Merge remote-tracking branch 'origin/main' into siyuan/cdc-backfill-pg

ci/scripts/gen-integration-test-yaml.py

@@ -35,6 +35,7 @@
     'nats': ['json'],
     'doris-sink': ['json'],
     'starrocks-sink': ['json'],
+    'deltalake-sink': ['json'],
 }
 
 def gen_pipeline_steps():

ci/scripts/notify.py

@@ -58,6 +58,7 @@
     "vector-json": ["tao"],
     "doris-sink": ["xinhao"],
     "starrocks-sink": ["xinhao"],
+    "deltalake-sink": ["xinhao"],
 }
 
 def get_failed_tests(get_test_status, test_map):

e2e_test/batch/catalog/pg_attribute.slt.part

@@ -38,9 +38,7 @@ select i.relname, a.attname, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_attribute a on t.oid = a.attrelid and a.attnum = ANY(ix.indkey)
     where t.relname = 'tmp' order by a.attnum;
 ----
-tmp_idx id2 {2,3,4,5}
-tmp_idx id3 {2,3,4,5}
-tmp_idx id4 {2,3,4,5}
+tmp_idx id2 {2}
 
 statement ok
 drop table tmp;

e2e_test/batch/catalog/pg_index.slt.part

@@ -10,7 +10,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id2_idx';
 ----
-2 {2,3}
+1 {2}
 
 statement ok
 create index tmp_id2_idx_include_id1 on tmp(id2) include(id1);
@@ -21,7 +21,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id2_idx_include_id1';
 ----
-3 {2,3,4}
+1 {2}
 
 statement ok
 create index tmp_id1_id2_idx on tmp(id1, id2);
@@ -32,7 +32,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id1_id2_idx';
 ----
-3 {2,3,4}
+2 {1,2}
 
 statement ok
 drop table tmp;

e2e_test/over_window/generated/batch/main.slt.part

@@ -1,11 +1,27 @@
 # This file is generated by `gen.py`. Do not edit it manually!
 
 statement ok
-SET RW_IMPLICIT_FLUSH TO true;
-
-include ./basic/mod.slt.part
-include ./rank_func/mod.slt.part
-include ./expr_in_win_func/mod.slt.part
-include ./agg_in_win_func/mod.slt.part
-include ./opt_agg_then_join/mod.slt.part
-include ./with_filter/mod.slt.part
+set rw_implicit_flush = true;
+
+statement ok
+set rw_streaming_over_window_cache_policy = full;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_first_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_last_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = default;

e2e_test/over_window/generated/batch/run_all.slt.part

@@ -0,0 +1,8 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+include ./basic/mod.slt.part
+include ./rank_func/mod.slt.part
+include ./expr_in_win_func/mod.slt.part
+include ./agg_in_win_func/mod.slt.part
+include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

e2e_test/over_window/generated/streaming/main.slt.part

@@ -1,11 +1,27 @@
 # This file is generated by `gen.py`. Do not edit it manually!
 
 statement ok
-SET RW_IMPLICIT_FLUSH TO true;
-
-include ./basic/mod.slt.part
-include ./rank_func/mod.slt.part
-include ./expr_in_win_func/mod.slt.part
-include ./agg_in_win_func/mod.slt.part
-include ./opt_agg_then_join/mod.slt.part
-include ./with_filter/mod.slt.part
+set rw_implicit_flush = true;
+
+statement ok
+set rw_streaming_over_window_cache_policy = full;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_first_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_last_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = default;

e2e_test/over_window/generated/streaming/run_all.slt.part

@@ -0,0 +1,8 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+include ./basic/mod.slt.part
+include ./rank_func/mod.slt.part
+include ./expr_in_win_func/mod.slt.part
+include ./agg_in_win_func/mod.slt.part
+include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

e2e_test/over_window/templates/main.slt.part

@@ -1,9 +1,25 @@
 statement ok
-SET RW_IMPLICIT_FLUSH TO true;
-
-include ./basic/mod.slt.part
-include ./rank_func/mod.slt.part
-include ./expr_in_win_func/mod.slt.part
-include ./agg_in_win_func/mod.slt.part
-include ./opt_agg_then_join/mod.slt.part
-include ./with_filter/mod.slt.part
+set rw_implicit_flush = true;
+
+statement ok
+set rw_streaming_over_window_cache_policy = full;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_first_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_last_n;
+
+include ./run_all.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = default;

e2e_test/over_window/templates/run_all.slt.part

@@ -0,0 +1,6 @@
+include ./basic/mod.slt.part
+include ./rank_func/mod.slt.part
+include ./expr_in_win_func/mod.slt.part
+include ./agg_in_win_func/mod.slt.part
+include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

integration_tests/deltalake-sink/README.md

@@ -0,0 +1,25 @@
+# Demo: Sinking to Delta Lake
+
+In this demo, we will create an append-only source via our datagen source,
+and sink the data generated from source to the downstream delta lake table
+stored on minio.
+
+1. Launch the cluster via docker compose
+```
+docker compose up -d
+```
+
+2. Create a delta lake table on minio
+```
+docker compose exec minio-0 mkdir /data/deltalake
+docker compose exec spark bash /spark-script/run-sql-file.sh create-table
+```
+
+3. Execute the SQL queries in sequence:
+    - create_source.sql
+    - create_sink.sql
+
+4. Query delta lake table. The following command will query the total count of records.
+```
+docker compose exec spark bash /spark-script/run-sql-file.sh query-table
+```

integration_tests/deltalake-sink/create_sink.sql

@@ -0,0 +1,10 @@
+create sink delta_lake_sink from source
+with (
+    connector = 'deltalake',
+    type = 'append-only',
+    force_append_only='true',
+    location = 's3a://deltalake/delta',
+    s3.access.key = 'hummockadmin',
+    s3.secret.key = 'hummockadmin',
+    s3.endpoint = 'http://minio-0:9301'
+);

integration_tests/deltalake-sink/create_source.sql

@@ -0,0 +1,3 @@
+CREATE table source (id int, name varchar);
+
+INSERT into source values (1, 'a'), (2, 'b'), (3, 'c');

integration_tests/deltalake-sink/docker-compose.yml

@@ -0,0 +1,43 @@
+---
+version: "3"
+services:
+  spark:
+    image: apache/spark:3.3.1
+    command: tail -f /dev/null
+    depends_on:
+      - minio-0
+    volumes:
+      - "./spark-script:/spark-script"
+    container_name: spark
+  risingwave-standalone:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: risingwave-standalone
+  etcd-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: etcd-0
+  grafana-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: grafana-0
+  minio-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: minio-0
+  prometheus-0:
+    extends:
+      file: ../../docker/docker-compose.yml
+      service: prometheus-0
+volumes:
+  compute-node-0:
+    external: false
+  etcd-0:
+    external: false
+  grafana-0:
+    external: false
+  minio-0:
+    external: false
+  prometheus-0:
+    external: false
+name: risingwave-compose

integration_tests/deltalake-sink/prepare.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -euo pipefail
+
+# build minio dir and create table
+docker compose exec minio-0 mkdir /data/deltalake
+docker compose exec spark bash /spark-script/run-sql-file.sh create-table