Merge branch 'main' into wangzheng/chore_gc_metric

Author: zwang28

.github/workflows/typo.yml

@@ -10,4 +10,4 @@ jobs:
       uses: actions/checkout@v3
 
     - name: Check spelling of the entire repository
-      uses: crate-ci/typos@v1.14.12
+      uses: crate-ci/typos@v1.15.0

.typos.toml

@@ -1,13 +1,16 @@
 [default.extend-words]
-indexs = "indices"
-Stichting = "Stichting"
-fo = "fo"
-FPR = "FPR"
+indexs = "indices"      # Both are valid, just pick one.
+Stichting = "Stichting" # This is Dutch for "Foundation". From DuckDB.
+FPR = "FPR"             # False Positive Rate
+inout = "inout"         # This is a SQL keyword!
+numer = "numer"         # numerator
+nd = "nd"               # N-dimentional / 2nd
+steam = "stream"        # You played with Steam games too much.
+# Some weird short variable names
 ot = "ot"
-inout = "inout"
-numer = "numer"
-nd = "nd"
-steam = "stream"
+bui = "bui"
+
+[default.extend-identifiers]
 
 [files]
 extend-exclude = [

Cargo.lock

@@ -910,7 +910,7 @@ if [ $# -gt 0 ]; then
   ARGS=("$@")
 
   echo "Applying clippy --fix for $@ (including dirty and staged files)"
-  cargo clippy ${ARGS[@]/#/--package risingwave_} ${RISINGWAVE_FEATURE_FLAGS} --fix --allow-dirty --allow-staged
+  cargo clippy ${ARGS[@]/#/--package risingwave_} --fix --allow-dirty --allow-staged
 else
   echo "Applying clippy --fix for all targets to all files (including dirty and staged files)"
   echo "Tip: run $(tput setaf 4)./risedev cf {package_names}$(tput sgr0) to only check-fix those packages (e.g. frontend, meta)."
@@ -922,7 +922,7 @@ fi
 private = true
 category = "RiseDev - Check"
 description = "Run cargo typos-cli check"
-install_crate = { min_version = "1.14.8", crate_name = "typos-cli", binary = "typos", test_arg = [
+install_crate = { min_version = "1.15.0", crate_name = "typos-cli", binary = "typos", test_arg = [
   "--help",
 ], install_command = "binstall" }
 script = """

Makefile.toml

@@ -0,0 +1,33 @@
+statement error
+select p, percentile_cont(p) within group (order by x::float8)
+from generate_series(1,5) x,
+     (values (0::float8),(0.1),(0.25),(0.4),(0.5),(0.6),(0.75),(0.9),(1)) v(p)
+group by p order by p;
+
+statement error
+select percentile_cont(array[0,1,0.25,0.75,0.5,1,0.3,0.32,0.35,0.38,0.4]) within group (order by x)
+from generate_series(1,6) x;
+
+statement error
+select percentile_disc(array[0.25,0.5,0.75]) within group (order by x)
+from unnest('{fred,jim,fred,jack,jill,fred,jill,jim,jim,sheila,jim,sheila}'::text[]) u(x);
+
+statement error
+select pg_collation_for(percentile_disc(1) within group (order by x collate "POSIX"))
+  from (values ('fred'),('jim')) v(x);
+
+query RR
+select
+  percentile_cont(0.5) within group (order by a),
+  percentile_disc(0.5) within group (order by a)
+from (values(1::float8),(3),(5),(7)) t(a);
+----
+4 3
+
+query RR
+select
+  percentile_cont(0.25) within group (order by a),
+  percentile_disc(0.5) within group (order by a)
+from (values(1::float8),(3),(5),(7)) t(a);
+----
+2.5 3

e2e_test/batch/aggregate/ordered_set_agg.slt.part

@@ -0,0 +1,14 @@
+query I
+select abs.abs from abs(-1);
+----
+1
+
+query I
+select alias.alias from abs(-1) alias;
+----
+1
+
+query I
+select alias.col from abs(-1) alias(col);
+----
+1

e2e_test/batch/functions/func_in_from.part

@@ -35,3 +35,21 @@ drop materialized view mv;
 
 statement ok
 drop table t;
+
+statement ok
+create materialized view mv as select * from abs(-1);
+
+# TODO: support this
+statement error not yet implemented: LogicalTableFunction::logical_rewrite_for_stream
+create materialized view mv2 as select * from range(1,2);
+
+statement ok
+flush;
+
+query IR
+select * from mv;
+----
+1
+
+statement ok
+drop materialized view mv;

e2e_test/streaming/values.slt

@@ -8,15 +8,15 @@ type SinkRecord interface {
 	// Convert the event to an INSERT INTO command.
 	ToPostgresSql() string
 
-	// Convert the event to a Kakfa message in JSON format.
+	// Convert the event to a Kafka message in JSON format.
 	// This interface will also be used for Pulsar and Kinesis.
 	ToJson() (topic string, key string, data []byte)
 
-	// Convert the event to a Kakfa message in Protobuf format.
+	// Convert the event to a Kafka message in Protobuf format.
 	// This interface will also be used for Pulsar and Kinesis.
 	ToProtobuf() (topic string, key string, data []byte)
 
-	// Convert the event to a Kakfa message in Avro format.
+	// Convert the event to a Kafka message in Avro format.
 	// This interface will also be used for Pulsar and Kinesis.
 	ToAvro() (topic string, key string, data []byte)
 }
@@ -40,7 +40,7 @@ func (r BaseSinkRecord) ToAvro() (topic string, key string, data []byte) {
 	panic("not implemented")
 }
 
-// Convert the event to a Kakfa message in the given format.
+// Convert the event to a Kafka message in the given format.
 // This interface will also be used for Pulsar and Kinesis.
 func RecordToKafka(r SinkRecord, format string) (topic string, key string, data []byte) {
 	if format == "json" {

integration_tests/datagen/sink/sink.go

@@ -14,4 +14,32 @@ FROM
         jdbc.url = 'jdbc:mysql://mysql:3306/mydb?user=root&password=123456',
         table.name = 'target_count',
         type = 'upsert'
-    );
+    );
+
+-- ingest the table back to RW
+CREATE TABLE rw_types (
+    id BIGINT PRIMARY KEY,
+    varchar_column VARCHAR,
+    text_column TEXT,
+    integer_column INTEGER,
+    smallint_column SMALLINT,
+    bigint_column BIGINT,
+    decimal_column DECIMAL,
+    real_column REAL,
+    double_column DOUBLE PRECISION,
+    boolean_column BOOLEAN,
+    date_column DATE,
+    time_column TIME,
+    timestamp_column TIMESTAMP,
+    jsonb_column JSONB,
+    bytea_column BYTEA
+) WITH (
+    connector = 'mysql-cdc',
+    hostname = 'mysql',
+    port = '3306',
+    username = 'root',
+    password = '123456',
+    database.name = 'mydb',
+    table.name = 'data_types',
+    server.id = '3'
+);

integration_tests/mysql-sink/create_mv.sql

@@ -11,4 +11,39 @@ CREATE SOURCE user_behaviors (
     topic = 'user_behaviors',
     properties.bootstrap.server = 'message_queue:29092',
     scan.startup.mode = 'earliest'
-) ROW FORMAT JSON;
+) ROW FORMAT JSON;
+
+CREATE TABLE data_types (
+    id BIGINT PRIMARY KEY,
+    varchar_column VARCHAR,
+    text_column TEXT,
+    integer_column INTEGER,
+    smallint_column SMALLINT,
+    bigint_column BIGINT,
+    decimal_column DECIMAL,
+    real_column REAL,
+    double_column DOUBLE PRECISION,
+    boolean_column BOOLEAN,
+    date_column DATE,
+    time_column TIME,
+    timestamp_column TIMESTAMP,
+    jsonb_column JSONB,
+    bytea_column BYTEA
+);
+
+CREATE SINK data_types_mysql_sink
+FROM
+    data_types WITH (
+        connector = 'jdbc',
+        jdbc.url = 'jdbc:mysql://mysql:3306/mydb?user=root&password=123456',
+        table.name = 'data_types',
+        type = 'upsert'
+    );
+
+INSERT INTO data_types (id, varchar_column, text_column, integer_column, smallint_column, bigint_column, decimal_column, real_column, double_column, boolean_column, date_column, time_column, timestamp_column, jsonb_column, bytea_column)
+VALUES
+    (1, 'Varchar value 1', 'Text value 1', 123, 456, 789, 12.34, 56.78, 90.12, TRUE, '2023-05-22', '12:34:56', '2023-05-22 12:34:56', '{"key": "value"}', E'\\xDEADBEEF'),
+    (2, 'Varchar value 2', 'Text value 2', 234, 567, 890, 23.45, 67.89, 01.23, FALSE, '2023-05-23', '23:45:01', '2023-05-23 23:45:01', '{"key": "value2"}', E'\\xFEEDBEEF'),
+    (3, 'Varchar value 3', 'Text value 3', 345, 678, 901, 34.56, 78.90, 12.34, TRUE, '2023-05-24', '12:34:56', '2023-05-24 12:34:56', '{"key": "value3"}', E'\\xCAFEBABE'),
+    (4, 'Varchar value 4', 'Text value 4', 456, 789, 012, 45.67, 89.01, 23.45, FALSE, '2023-05-25', '23:45:01', '2023-05-25 23:45:01', '{"key": "value4"}', E'\\xBABEC0DE'),
+    (5, 'Varchar value 5', 'Text value 5', 567, 890, 123, 56.78, 90.12, 34.56, TRUE, '2023-05-26', '12:34:56', '2023-05-26 12:34:56', '{"key": "value5"}', E'\\xDEADBABE');

integration_tests/mysql-sink/create_source.sql

@@ -1 +1 @@
-user_behaviors,target_count
+user_behaviors,target_count,rw_types

integration_tests/mysql-sink/data_check

@@ -1,4 +1,24 @@
 CREATE TABLE target_count (
   target_id VARCHAR(128) primary key,
   target_count BIGINT
+);
+
+
+-- sink table
+CREATE TABLE data_types (
+    id BIGINT PRIMARY KEY,
+    varchar_column VARCHAR(255),
+    text_column TEXT,
+    integer_column INT,
+    smallint_column SMALLINT,
+    bigint_column BIGINT,
+    decimal_column DECIMAL(10,2),
+    real_column FLOAT,
+    double_column DOUBLE,
+    boolean_column BOOLEAN,
+    date_column DATE,
+    time_column TIME,
+    timestamp_column TIMESTAMP,
+    jsonb_column JSON,
+    bytea_column BLOB
 );

integration_tests/mysql-sink/mysql_prepare.sql

@@ -2,5 +2,12 @@ SELECT
     *
 FROM
     target_count
+LIMIT
+    10;
+
+SELECT
+    *
+FROM
+    data_types
 LIMIT
     10;

integration_tests/mysql-sink/query.sql

@@ -14,4 +14,36 @@ FROM
         jdbc.url = 'jdbc:postgresql://postgres:5432/mydb?user=myuser&password=123456',
         table.name = 'target_count',
         type = 'upsert'
-    );
+    );
+
+-- ingest back to RW
+CREATE table rw_types (
+    id BIGINT PRIMARY KEY,
+    varchar_column VARCHAR,
+    text_column TEXT,
+    integer_column INTEGER,
+    smallint_column SMALLINT,
+    bigint_column BIGINT,
+    decimal_column DECIMAL,
+    real_column REAL,
+    double_column DOUBLE PRECISION,
+    boolean_column BOOLEAN,
+    date_column DATE,
+    time_column TIME,
+    timestamp_column TIMESTAMP,
+    timestamptz_column TIMESTAMPTZ,
+    interval_column INTERVAL,
+    jsonb_column JSONB,
+    bytea_column BYTEA,
+    array_column VARCHAR[]
+) WITH (
+      connector = 'postgres-cdc',
+      hostname = 'postgres',
+      port = '5432',
+      username = 'myuser',
+      password = '123456',
+      database.name = 'mydb',
+      schema.name = 'public',
+      table.name = 'data_types',
+      slot.name = 'data_types'
+);