Merge branch 'main' into chore/switch-e2e-sql-backend

Makefile.toml

@@ -1244,7 +1244,7 @@ cat <<EOF > "${INSTALL_PATH}"
 #!/usr/bin/env bash
 set -e
 cd "$DIR"
-./risedev "\\$@"
+RISEDEV_CMD="\\$(basename \\"$0")" ./risedev "\\$@"
 EOF
 chmod +x "${INSTALL_PATH}"
 

e2e_test/batch/types/map.slt.part

@@ -122,6 +122,63 @@ select to_jsonb(m1), to_jsonb(m2), to_jsonb(m3), to_jsonb(l), to_jsonb(s) from t
 {"a": 1.0, "b": 2.0, "c": 3.0}	null	null	null	null
 {"a": 1.0, "b": 2.0, "c": 3.0}	{"1": true, "2": false, "3": true}	{"a": {"a1": "a2"}, "b": {"b1": "b2"}}	[{"a": 1, "b": 2, "c": 3}, {"d": 4, "e": 5, "f": 6}]	{"m": {"a": {"x": 1}, "b": {"x": 2}, "c": {"x": 3}}}
 
+query ?
+select jsonb_populate_map(
+    null::map(varchar, int),
+    '{"a": 1, "b": 2}'::jsonb
+);
+----
+{a:1,b:2}
+
+
+query ?
+select jsonb_populate_map(
+    MAP {'a': 1, 'b': 2},
+    '{"b": 3, "c": 4}'::jsonb
+);
+----
+{a:1,b:3,c:4}
+
+
+# implicit cast (int -> varchar)
+query ?
+select jsonb_populate_map(
+    MAP {'a': 'a', 'b': 'b'},
+    '{"b": 3, "c": 4}'::jsonb
+);
+----
+{a:a,b:3,c:4}
+
+
+query error
+select jsonb_populate_map(
+    MAP {'a': 1, 'b': 2},
+    '{"b": "3", "c": 4}'::jsonb
+);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `jsonb_populate_map('{a:1,b:2}', '{"b": "3", "c": 4}')`
+  3: Parse error: cannot cast jsonb string to type number
+
+
+query error
+select jsonb_populate_map(
+    null::map(int, int),
+    '{"a": 1, "b": 2}'::jsonb
+);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `jsonb_populate_map(NULL, '{"a": 1, "b": 2}')`
+  3: Parse error: cannot convert jsonb to a map with non-string keys
+
+
+
 statement ok
 drop table t;
 

e2e_test/s3/fs_parquet_source_and_sink.py

@@ -137,7 +137,7 @@ def _table():
         s3.bucket_name = '{config['S3_BUCKET']}',
         s3.credentials.access = '{config['S3_ACCESS_KEY']}',
         s3.credentials.secret = '{config['S3_SECRET_KEY']}',
-        s3.endpoint_url = 'https://{config['S3_ENDPOINT']}'
+        s3.endpoint_url = 'https://{config['S3_ENDPOINT']}',
         s3.path = '',
         s3.file_type = 'parquet',
         type = 'append-only',

e2e_test/sink/iceberg_sink.slt

@@ -48,23 +48,65 @@ CREATE SOURCE iceberg_demo_source WITH (
     table.name='e2e_demo_table'
 );
 
+statement ok
+CREATE SINK s7 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH (
+    connector = 'iceberg',
+    type = 'upsert',
+    primary_key = 'v1',
+    warehouse.path = 's3a://hummock001',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.access.key = secret iceberg_s3_access_key,
+    s3.secret.key = secret iceberg_s3_secret_key,
+    s3.region = 'us-east-1',
+    catalog.name = 'demo',
+    catalog.type = 'storage',
+    database.name='demo_db',
+    table.name='e2e_auto_create_table',
+    commit_checkpoint_interval = 1,
+    create_table_if_not_exists = 'true'
+);
+
+statement ok
+CREATE SOURCE iceberg_e2e_auto_create_table WITH (
+    connector = 'iceberg',
+    warehouse.path = 's3a://hummock001',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.access.key = secret iceberg_s3_access_key,
+    s3.secret.key = secret iceberg_s3_secret_key,
+    s3.region = 'us-east-1',
+    catalog.name = 'demo',
+    catalog.type = 'storage',
+    database.name='demo_db',
+    table.name='e2e_auto_create_table'
+);
+
 statement ok
 INSERT INTO t6 VALUES (1, 2, '1-2'), (2, 2, '2-2'), (3, 2, '3-2'), (5, 2, '5-2'), (8, 2, '8-2'), (13, 2, '13-2'), (21, 2, '21-2');
 
 statement ok
 FLUSH;
 
-sleep 5s
+sleep 20s
+
+query I
+select count(*) from rw_iceberg_snapshots where source_name = 'iceberg_demo_source';
+----
+1
+
+query I
+select count(*) from rw_iceberg_snapshots where source_name = 'iceberg_e2e_auto_create_table';
+----
+1
 
 query I
 select sum(record_count) from rw_iceberg_files where source_name = 'iceberg_demo_source';
 ----
 7
 
 query I
-select count(*) from rw_iceberg_snapshots where source_name = 'iceberg_demo_source';
+select sum(record_count) from rw_iceberg_files where source_name = 'iceberg_e2e_auto_create_table';
 ----
-1
+7
 
 statement ok
 INSERT INTO t6 VALUES (1, 50, '1-50');
@@ -78,10 +120,7 @@ statement ok
 DROP SOURCE iceberg_demo_source;
 
 statement ok
-DROP SINK s6;
-
-statement ok
-DROP MATERIALIZED VIEW mv6;
+DROP SOURCE iceberg_e2e_auto_create_table;
 
 statement ok
-DROP TABLE t6;
+DROP TABLE t6 cascade;

e2e_test/source/cdc_inline/alter/cdc_table_alter.slt

@@ -131,13 +131,13 @@ select order_id, product_id, shipment_id from enriched_orders order by order_id;
 system ok
 mysql -e "
   USE testdb1;
-  ALTER TABLE products ADD COLUMN weight DECIMAL(10, 2) NOT NULL DEFAULT 0.0;
+  ALTER TABLE products ADD COLUMN weight DECIMAL(10, 2) NOT NULL DEFAULT 1.1;
   ALTER TABLE orders ADD COLUMN order_comment VARCHAR(255);
 "
 
 # alter cdc tables
 statement ok
-ALTER TABLE my_products ADD COLUMN weight DECIMAL;
+ALTER TABLE my_products ADD COLUMN weight DECIMAL DEFAULT 1.1;
 
 statement ok
 ALTER TABLE my_orders ADD COLUMN order_comment VARCHAR;
@@ -148,9 +148,9 @@ sleep 3s
 query ITTT
 SELECT id,name,description,weight FROM my_products order by id limit 3
 ----
-101 scooter Small 2-wheel scooter NULL
-102 car battery 12V car battery NULL
-103 12-pack drill 12-pack of drill bits with sizes ranging from #40 to #3 NULL
+101 scooter Small 2-wheel scooter 1.1
+102 car battery 12V car battery 1.1
+103 12-pack drill 12-pack of drill bits with sizes ranging from #40 to #3 1.1
 
 
 # update mysql tables
@@ -169,7 +169,7 @@ SELECT id,name,description,weight FROM my_products order by id limit 3
 ----
 101 scooter Small 2-wheel scooter 10.50
 102 car battery 12V car battery 12.50
-103 12-pack drill 12-pack of drill bits with sizes ranging from #40 to #3 NULL
+103 12-pack drill 12-pack of drill bits with sizes ranging from #40 to #3 1.1
 
 query ITTT
 SELECT order_id,order_date,customer_name,product_id,order_status,order_comment FROM my_orders order by order_id limit 2

e2e_test/source/cdc_inline/auto_schema_change_mysql.slt

@@ -10,8 +10,11 @@ mysql -e "
     CREATE TABLE customers(
          id BIGINT PRIMARY KEY,
          modified DATETIME,
+         name VARCHAR(32),
          custinfo JSON
     );
+    INSERT INTO customers VALUES(1, NOW(), 'John', NULL);
+    INSERT INTO customers VALUES(2, NOW(), 'Doe', NULL);
     ALTER TABLE customers ADD INDEX zipsa( (CAST(custinfo->'zipcode' AS UNSIGNED ARRAY)) );
     "
 
@@ -28,14 +31,15 @@ create source mysql_source with (
 );
 
 statement ok
-create table rw_customers (id bigint, modified timestamp, custinfo jsonb, primary key (id)) from mysql_source table 'mytest.customers';
+create table rw_customers (id bigint, modified timestamp, name varchar, custinfo jsonb, primary key (id)) from mysql_source table 'mytest.customers';
 
 # Name, Type, Is Hidden, Description
 query TTTT
 describe rw_customers;
 ----
 id bigint false NULL
 modified timestamp without time zone false NULL
+name character varying false NULL
 custinfo jsonb false NULL
 primary key id NULL NULL
 distribution key id NULL NULL
@@ -46,8 +50,8 @@ table description rw_customers NULL NULL
 system ok
 mysql -e "
   USE mytest;
-  ALTER TABLE customers ADD COLUMN v1 VARCHAR(255);
-  ALTER TABLE customers ADD COLUMN v2 double(5,2);
+  ALTER TABLE customers ADD COLUMN v1 VARCHAR(255) DEFAULT 'hello';
+  ALTER TABLE customers ADD COLUMN v2 double(5,2) DEFAULT 88.9;
 "
 
 sleep 3s
@@ -58,13 +62,20 @@ describe rw_customers;
 ----
 id bigint false NULL
 modified timestamp without time zone false NULL
+name character varying false NULL
 custinfo jsonb false NULL
 v1 character varying false NULL
 v2 double precision false NULL
 primary key id NULL NULL
 distribution key id NULL NULL
 table description rw_customers NULL NULL
 
+query TTTT
+select id,v1,v2,name from rw_customers order by id;
+----
+1 hello 88.9 John
+2 hello 88.9 Doe
+
 # rename column on upstream will not be replicated, since we do not support rename column
 system ok
 mysql -e "
@@ -81,6 +92,7 @@ describe rw_customers;
 ----
 id bigint false NULL
 modified timestamp without time zone false NULL
+name character varying false NULL
 custinfo jsonb false NULL
 v1 character varying false NULL
 v2 double precision false NULL
@@ -112,6 +124,7 @@ query TTTT
 describe rw_customers;
 ----
 id bigint false NULL
+name character varying false NULL
 custinfo jsonb false NULL
 primary key id NULL NULL
 distribution key id NULL NULL

e2e_test/source_inline/kafka/issue_18308.slt

@@ -0,0 +1,50 @@
+control substitution on
+
+system ok
+rpk topic create test-topic-18308
+
+statement ok
+CREATE SOURCE kafkasource (
+    id int,
+    name string,
+)
+WITH (
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = 'test-topic-18308',
+    scan.startup.mode = 'earliest'
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+CREATE TABLE compact_table (
+  id int,
+  name varchar,
+  PRIMARY KEY (id)
+);
+
+statement ok
+CREATE SINK table_sink INTO compact_table AS SELECT * FROM kafkasource;
+
+system ok
+echo '{ "id": 1, "name": "xxchan" }' | rpk topic produce test-topic-18308
+
+sleep 5s
+
+statement ok
+flush;
+
+query IT
+SELECT * FROM compact_table;
+----
+1 xxchan
+
+statement ok
+DROP SINK table_sink;
+
+statement ok
+DROP TABLE compact_table;
+
+statement ok
+DROP SOURCE kafkasource;
+
+system ok
+rpk topic delete test-topic-18308