Merge remote-tracking branch 'origin' into tab/drop-table-connector

Cargo.toml

@@ -379,8 +379,6 @@ tokio-postgres = { git = "https://github.com/madsim-rs/rust-postgres.git", rev =
 sqlx = { git = "https://github.com/madsim-rs/sqlx.git", rev = "3efe6d0065963db2a2b7f30dee69f647e28dec81" }
 # patch to remove preserve_order from serde_json
 bson = { git = "https://github.com/risingwavelabs/bson-rust", rev = "e5175ec" }
-# TODO: unpatch after PR merged https://github.com/tokio-rs/prost/pull/1210
-prost-build = { git = "https://github.com/xxchan/prost.git", rev = "0eb1c7b09976cf6b5231e4b8d87bb5908ae6a163" }
 
 [workspace.metadata.dylint]
 libraries = [{ path = "./lints" }]

ci/scripts/e2e-iceberg-sink-v2-test.sh

@@ -35,6 +35,7 @@ echo "--- preparing iceberg"
 cd e2e_test/iceberg
 bash ./start_spark_connect_server.sh
 
+echo "--- Running tests"
 # Don't remove the `--quiet` option since poetry has a bug when printing output, see
 # https://github.com/python-poetry/poetry/issues/3412
 poetry update --quiet
@@ -52,6 +53,8 @@ poetry run python main.py -t ./test_case/iceberg_source_all_delete.toml
 poetry run python main.py -t ./test_case/iceberg_source_explain_for_delete.toml
 poetry run python main.py -t ./test_case/iceberg_predicate_pushdown.toml
 
+echo "--- Running benchmarks"
+poetry run python main.py -t ./benches/predicate_pushdown.toml
 
 echo "--- Kill cluster"
 cd ../../

ci/workflows/main-cron.yml

@@ -190,7 +190,7 @@ steps:
             <<: *docker-compose-common
             run: source-test-env
         - ./ci/plugins/upload-failure-logs
-      timeout_in_minutes: 15
+      timeout_in_minutes: 20
       retry: *auto-retry
 
     - label: "end-to-end sink test ({{matrix.backend}} backend)"

e2e_test/ddl/alter_owner.slt

@@ -67,6 +67,9 @@ v user1
 statement ok
 CREATE MATERIALIZED VIEW mv AS SELECT v1, (t.v2).v1 AS v21 FROM t;
 
+statement ok
+CREATE INDEX mv_idx ON mv(v1);
+
 statement ok
 ALTER MATERIALIZED VIEW mv OWNER TO user1;
 
@@ -85,6 +88,21 @@ WHERE
 ----
 mv user1
 
+query TT
+SELECT
+    pg_class.relname AS rel_name,
+    pg_roles.rolname AS owner
+FROM
+    pg_class
+    JOIN pg_namespace ON pg_namespace.oid = pg_class.relnamespace
+    JOIN pg_roles ON pg_roles.oid = pg_class.relowner
+WHERE
+    pg_namespace.nspname NOT LIKE 'pg_%'
+    AND pg_namespace.nspname != 'information_schema'
+    AND pg_class.relname = 'mv_idx';
+----
+mv_idx user1
+
 statement ok
 CREATE SOURCE src (v INT) WITH (
     connector = 'datagen',

e2e_test/ddl/alter_set_schema.slt

@@ -23,12 +23,21 @@ CREATE INDEX test_index1 ON test_table(u);
 statement ok
 CREATE INDEX test_index2 ON test_table(v);
 
+statement ok
+CREATE MATERIALIZED VIEW test_mv AS SELECT u FROM test_table;
+
+statement ok
+CREATE INDEX test_mv_index ON test_mv(u);
+
 statement ok
 ALTER TABLE test_table SET SCHEMA public;
 
 statement ok
 ALTER TABLE test_table SET SCHEMA test_schema;
 
+statement ok
+ALTER MATERIALIZED VIEW test_mv SET SCHEMA test_schema;
+
 query TT
 SELECT tablename, schemaname FROM pg_tables WHERE schemaname = 'test_schema';
 ----
@@ -39,6 +48,7 @@ SELECT indexname, schemaname FROM pg_indexes WHERE schemaname = 'test_schema';
 ----
 test_index1 test_schema
 test_index2 test_schema
+test_mv_index test_schema
 
 statement ok
 CREATE SOURCE test_source (v INT) WITH (
@@ -104,7 +114,7 @@ statement ok
 DROP SOURCE test_schema.test_source;
 
 statement ok
-DROP TABLE test_schema.test_table;
+DROP TABLE test_schema.test_table cascade;
 
 statement ok
 DROP SCHEMA test_schema;

e2e_test/iceberg/benches/predicate_pushdown.slt

@@ -0,0 +1,65 @@
+statement ok
+CREATE TABLE t1 (i1 int, i2 varchar, i3 varchar);
+
+statement ok
+INSERT INTO t1 select key, 'some long string of text', 'another long string of text' from generate_series(1, 1000000) as key;
+
+statement ok
+CREATE SINK sink1 AS select * from t1 WITH (
+    connector = 'iceberg',
+    type = 'append-only',
+    force_append_only = 'true',
+    database.name = 'demo_db',
+    table.name = 't1',
+    catalog.name = 'demo',
+    catalog.type = 'storage',
+    warehouse.path = 's3a://icebergdata/demo',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.region = 'us-east-1',
+    s3.access.key = 'hummockadmin',
+    s3.secret.key = 'hummockadmin',
+    commit_checkpoint_interval = 1,
+    create_table_if_not_exists = 'true'
+);
+
+statement ok
+CREATE SOURCE iceberg_t1_source
+WITH (
+    connector = 'iceberg',
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.region = 'us-east-1',
+    s3.access.key = 'hummockadmin',
+    s3.secret.key = 'hummockadmin',
+    s3.path.style.access = 'true',
+    catalog.type = 'storage',
+    warehouse.path = 's3a://icebergdata/demo',
+    database.name = 'demo_db',
+    table.name = 't1',
+);
+
+statement ok
+flush;
+
+query I
+select count(*) from iceberg_t1_source;
+----
+1000000
+
+# warmup
+include ./predicate_pushdown/point_get.slt.part
+# bench
+include ./predicate_pushdown/point_get.slt.part
+
+# warmup
+include ./predicate_pushdown/filter.slt.part
+# bench
+include ./predicate_pushdown/filter.slt.part
+
+statement ok
+DROP SINK sink1;
+
+statement ok
+DROP SOURCE iceberg_t1_source;
+
+statement ok
+DROP TABLE t1 cascade;

e2e_test/iceberg/benches/predicate_pushdown.toml

@@ -0,0 +1,11 @@
+init_sqls = [
+    'CREATE SCHEMA IF NOT EXISTS demo_db',
+    'DROP TABLE IF EXISTS demo_db.t1',
+]
+
+slt = 'benches/predicate_pushdown.slt'
+
+drop_sqls = [
+    'DROP TABLE IF EXISTS demo_db.t1',
+    'DROP SCHEMA IF EXISTS demo_db',
+]

e2e_test/iceberg/benches/predicate_pushdown/filter.slt.part

@@ -0,0 +1,4 @@
+query I
+select count(*) from iceberg_t1_source where i1 > 1001 and i1 < 1110;
+----
+108

e2e_test/iceberg/benches/predicate_pushdown/point_get.slt.part

@@ -0,0 +1,4 @@
+query I
+select * from iceberg_t1_source where i1 = 100000;
+----
+  100000 some long string of text another long string of text

e2e_test/sink/elasticsearch/elasticsearch_sink.slt

@@ -22,6 +22,7 @@ CREATE TABLE test_route (
 
 statement ok
 CREATE SINK test_route_sink from test_route WITH (
+    type = 'upsert',
     connector = 'elasticsearch',
     index = 'test_route',
     url = 'http://elasticsearch:9200',
@@ -32,6 +33,7 @@ CREATE SINK test_route_sink from test_route WITH (
 
 statement ok
 CREATE SINK s7 from t7 WITH (
+    type = 'upsert',
     connector = 'elasticsearch',
     index = 'test',
     url = 'http://elasticsearch:9200',
@@ -41,6 +43,7 @@ CREATE SINK s7 from t7 WITH (
 
 statement ok
 CREATE SINK s8 from t7 WITH (
+    type = 'upsert',
     connector = 'elasticsearch',
     index = 'test1',
     primary_key = 'v1,v3',

e2e_test/udf/create_and_drop.slt

@@ -0,0 +1,115 @@
+# https://github.com/risingwavelabs/risingwave/issues/17263
+
+statement ok
+create table t (a int, b int);
+
+statement ok
+create function add(a int, b int) returns int language python as $$
+def add(a, b):
+  return a+b
+$$;
+
+statement error function with name add\(integer,integer\) exists
+create function add(int, int) returns int language sql as $$select $1 + $2$$;
+
+statement ok
+create function if not exists add(int, int) returns int language sql as $$select $1 + $2$$;
+
+statement ok
+create function add_v2(int, int) returns int language sql as $$select $1 + $2$$;
+
+statement ok
+create aggregate mysum(value int) returns int language python as $$
+def create_state():
+    return 0
+def accumulate(state, value):
+    return state + value
+def finish(state):
+    return state
+$$;
+
+statement error function with name mysum\(integer\) exists
+create aggregate mysum(value int) returns int language python as $$
+def create_state():
+    return 0
+def accumulate(state, value):
+    return state + value
+def finish(state):
+    return state
+$$;
+
+statement ok
+create aggregate if not exists mysum(value int) returns int language python as $$
+def create_state():
+    return 0
+def accumulate(state, value):
+    return state + value
+def finish(state):
+    return state
+$$;
+
+statement ok
+create materialized view mv as select add(a, b) + add_v2(a, b) as c from t;
+
+statement ok
+create materialized view mv2 as select mysum(a) as s from t;
+
+statement error function used by 1 other objects
+drop function add;
+
+statement error function used by 1 other objects
+drop function if exists add;
+
+statement error function used by 1 other objects
+drop function add_v2;
+
+statement error function used by 1 other objects
+drop aggregate mysum;
+
+statement ok
+drop materialized view mv;
+
+statement ok
+drop materialized view mv2;
+
+statement ok
+drop function add;
+
+statement error function not found
+drop function add;
+
+statement ok
+drop function if exists add;
+
+statement ok
+drop function add_v2;
+
+statement ok
+drop function if exists add_v2;
+
+statement ok
+drop aggregate mysum;
+
+statement ok
+drop aggregate if exists mysum;
+
+statement ok
+create function add(a int, b int) returns int language python as $$
+def add(a, b):
+  return a+b
+$$;
+
+statement ok
+create sink s as select add(a, b) as c from t with (connector = 'blackhole');
+
+statement error function used by 1 other objects
+drop function add;
+
+statement ok
+drop sink s;
+
+statement ok
+drop function add;
+
+statement ok
+drop table t;