s3 连接器，转为指定 AWS S3 source 连接器为 s3_v2等。此外，本版本还新增了许多实用的 SQL 命令和函数，例如创建订阅功能等。一起来了解本次更新的主要亮点吧！1Recover 命令
RECOVER 命令。该命令可以触发临时性的恢复，当存在较高延迟时可能需要使用。需要注意的是，只有超级用户才能使用 RECOVER 命令，以限制可以触发恢复的用户范围。此外，RisingWave 仅能从已提交的时间点进行恢复，且该命令不会等待恢复完成。DROP 或 CANCEL 命令都将立即生效。RECOVER命令[1]
2配置全局参数
ALTER SYSTEM 命令在整个服务器上设置运行时参数或系统参数。使用此命令后，新参数值也将成为每个新会话的默认值。如果您喜欢使用不同的参数运行每个 RisingWave 会话，用此命令可以更轻松地设置每个会话。rw_enable_join_ordering 设置为 true。ALTER SYSTEM SET rw_enable_join_ordering TO true;
SHOW ALL 命令查看所有运行时参数，如果想查看当前运行时某个特定运行时参数的值，可使用 SHOW parameter_name 命令。如果想查看所有系统参数及其当前值，则可使用 SHOW PARAMETERS 命令。ALTER SYSTEM 命令[2] 查看和配置运行时参数[3] 查看和配置系统参数[4]
3增强时间连接
stream_source 是一个非仅追加的源，而 prod 是一个表，则以下 SQL 命令使用了时间连接来连接两个表。SELECT *
FROM stream_source
JOIN prod FOR SYSTEM_TIME AS OF PROC_TIME()
ON source_id = prod_id;
非仅追加模式的时间连接[5]
4支持订阅和订阅游标
CREATE SUBSCRIPTION 命令创建订阅。您还可以删除、修改和显示现有的订阅。创建订阅时可以选择增量数据的保留时间以及订阅可以访问的时间。以下是 SQL 查询在物化视图上创建订阅，并将数据保留一天的示例。CREATE SUBSCRIPTION sub1 FROM tbl1 WITH (retention = '1D' );
DECLARE cursor1 SUBSCRIPTION CURSOR FOR sub1;
FETCH NEXT FROM cursor1;
----结果
col1 | col2 | col3 | op | rw_timestamp
----+----+----+----+-------------------
1 | 2 | 3 | 1 |
(1 行)
col1、col2 和 col3 是订阅的表中的列，op 则是生成的特殊列。op 的值可以是 1、2、3 或 4，分别对应于 INSERT、DELETE、UPDATE_DELETE 和 UPDATE_INSERT。UPDATE 语句在这里被转换为 UPDATE_DELETE 和 UPDATE_INSERT。请注意，每次从游标中提取时，只会返回表中的一行数据。要查看其他行的数据，必须再次调用 FETCH 命令。订阅[6]
5使用 Iceberg source 进行 Time travel
AS OF 语法从特定时间段或版本中选择数据。RisingWave 目前仅支持从 Iceberg source 进行批量查询，拥有此功能后，随着时间推移跟踪数据集的变化会更容易。同时，从特定版本选择数据则使得数据可重现性增强，这对于调试非常重要：如果在数据中检测到错误，您可以轻松选择之前版本的数据。SELECT * FROM iceberg_source FOR system_version AS OF 1567123456789;
SELECT * FROM iceberg_source FOR system_time AS OF '2006-09-21 00:00:00+00:00';
从 Apache Iceberg 中摄取数据[7]
6PostgreSQL 和 MySQL 连接器的新配置
配置 SSL/TLS
ssl.mode 参数设置 SSL/TLS 加密级别。配置 SSL/TLS 可确保传输的数据的完整性和机密性，保护敏感信息免受威胁和攻击。许多监管标准和最佳实践也使用 SSL/TLS，这使得它成为许多用户关心的重要功能。ssl.mode 参数接受disabled、preferred 和 required这三个值，默认值为 disabled。CREATE SOURCE pg_source WITH (
connector = 'postgres-cdc',
hostname = '127.0.0.1',
port = '5432',
username = 'user',
password = 'password',
database.name = 'mydb',
slot.name = 'mydb_slot',
ssl.mode = 'required'
);
配置快照
除了为 CDC 表配置快照之外，您现在还可以配置屏障间隔和批次大小。在与 MySQL 或 PostgreSQL 数据库建立连接时，您可以创建 CDC source 或 CDC 表。CDC source 连接到整个数据库，而 CDC 表连接到单个 MySQL 或 PostgreSQL 表。快照相关的参数选项仅适用于创建表时。
snapshot.interval 可以设置开始快照读取时的屏障间隔；snapshot.batch_size 可以配置快照读取的批次大小。配置超时
cdc_source_wait_streaming_start_timeout 来增加超时阈值。SET cdc_source_wait_streaming_start_timeout to 90;
从 MySQL CDC 中摄取数据[8] 从 PostgreSQL CDC 中摄取数据[9]
7支持更多 Sink 连接器
支持 Snowflake sink 连接器
CREATE SINK SQL 语句即可完成。CREATE SINK snowflake_sink FROM mv WITH (
connector = 'snowflake',
type = 'append-only',
snowflake.database = 'db',
snowflake.schema = 'schema',
snowflake.pipe = 'pipe',
snowflake.account_identifier = '<ORG_NAME>-<ACCOUNT_NAME>',
snowflake.user = 'user',
snowflake.rsa_public_key_fp = 'fp',
snowflake.private_key = 'pk',
snowflake.s3_bucket = 's3_bucket',
snowflake.aws_access_key_id = 'aws_id',
snowflake.aws_secret_access_key = 'secret_key',
snowflake.aws_region = 'region',
snowflake.max_batch_row_num = '1030',
snowflake.s3_path = 's3_path',
);
支持 BigQuery sink 的 upsert 类型
append-only 类型的 BigQuery sink，这意味着只有 INSERT 操作会传递到 BigQuery 表中。现在，1.9 版本新增支持 upsert 类型的 Sink，允许传递 UPDATE 和 DELETE 操作。创建 upsert Sink 时，需要在 BigQuery 中进行额外的配置，并且必须设置相应的权限和主键。CREATE SINK big_query_sink
FROM mv
WITH (
connector = 'bigquery',
type = 'upsert',
bigquery.s3.path= '${s3_service_account_json_path}',
bigquery.project= '${project_id}',
bigquery.dataset= '${dataset_id}',
bigquery.table= '${table_id}',
access_key = '${aws_access_key}',
secret_access = '${aws_secret_access}',
region = '${aws_region}',
);
Delta Lake sink 支持 GCS
CREATE SINK 命令。CREATE SINK s1_sink FROM s1_source
WITH (
connector = 'deltalake',
type = 'append-only',
location = 'gs://bucket-name/path/to/file',
gcs.service.account = '{type: service_acct,
project_id: id,
private_key:key,
client_email: email}'
);
将数据从 RisingWave 导出到 Snowflake[11] 将数据从 RisingWave 导出到 BigQuery[12] 将数据从 RisingWave 导出到 Delta Lake[13]
8总结
RECOVER 命令: https://docs.risingwave.com/docs/current/sql-recover/
[2]ALTER SYSTEM 命令: https://docs.risingwave.com/docs/current/sql-alter-system/
[3]查看和配置运行时参数: https://docs.risingwave.com/docs/current/view-configure-runtime-parameters/
[4]查看和配置系统参数: https://docs.risingwave.com/docs/current/view-configure-system-parameters/
[5]非仅追加模式的时间连接: https://docs.risingwave.com/docs/current/query-syntax-join-clause/#process-time-temporal-joins
[6]订阅: https://docs.risingwave.com/docs/current/subscription/
[7]从 Apache Iceberg 中摄取数据: https://docs.risingwave.com/docs/current/ingest-from-iceberg/#time-travel
[8]从 MySQL CDC 中摄取数据: https://docs.risingwave.com/docs/current/ingest-from-mysql-cdc/
[9]从 PostgreSQL CDC 中摄取数据: https://docs.risingwave.com/docs/current/ingest-from-postgres-cdc/
[10]集成页面: https://docs.risingwave.com/docs/current/rw-integration-summary/
[11]将数据从 RisingWave 导出到 Snowflake: https://docs.risingwave.com/docs/current/sink-to-snowflake/
[12]将数据从 RisingWave 导出到 BigQuery: https://docs.risingwave.com/docs/current/sink-to-bigquery/
[13]将数据从 RisingWave 导出到 Delta Lake: https://docs.risingwave.com/docs/current/sink-to-delta-lake/
[14]发布说明: https://docs.risingwave.com/release-notes/
关于 RisingWave
往期推荐
技术内幕
