Merge branch 'main' into xinjing/bind_ordered_set_agg

e2e_test/batch/catalog/issue_10177.slt.part

@@ -54,5 +54,6 @@ pg_catalog 1700 numeric b f NULL
 pg_catalog 20 int8 b f NULL
 pg_catalog 21 int2 b f NULL
 pg_catalog 23 int4 b f NULL
+pg_catalog 25 text b f NULL
 pg_catalog 700 float4 b f NULL
 pg_catalog 701 float8 b f NULL

e2e_test/batch/catalog/pg_cast.slt.part

@@ -1,95 +1,95 @@
 query IIIT
 SELECT * FROM pg_catalog.pg_cast;
 ----
-0 16 23 EXPLICIT
-1 16 1043 ASSIGN
-2 21 23 IMPLICIT
-3 21 20 IMPLICIT
-4 21 700 IMPLICIT
-5 21 701 IMPLICIT
-6 21 1700 IMPLICIT
-7 21 1043 ASSIGN
-8 21 1301 IMPLICIT
-9 23 16 EXPLICIT
-10 23 21 ASSIGN
-11 23 20 IMPLICIT
-12 23 700 IMPLICIT
-13 23 701 IMPLICIT
-14 23 1700 IMPLICIT
-15 23 1043 ASSIGN
-16 23 1301 IMPLICIT
-17 20 21 ASSIGN
-18 20 23 ASSIGN
-19 20 700 IMPLICIT
-20 20 701 IMPLICIT
-21 20 1700 IMPLICIT
-22 20 1043 ASSIGN
-23 20 1301 IMPLICIT
-24 700 21 ASSIGN
-25 700 23 ASSIGN
-26 700 20 ASSIGN
-27 700 701 IMPLICIT
-28 700 1700 ASSIGN
-29 700 1043 ASSIGN
-30 701 21 ASSIGN
-31 701 23 ASSIGN
-32 701 20 ASSIGN
-33 701 700 ASSIGN
-34 701 1700 ASSIGN
-35 701 1043 ASSIGN
-36 1700 21 ASSIGN
-37 1700 23 ASSIGN
-38 1700 20 ASSIGN
-39 1700 700 IMPLICIT
-40 1700 701 IMPLICIT
-41 1700 1043 ASSIGN
-42 1082 1043 ASSIGN
-43 1082 1114 IMPLICIT
-44 1082 1184 IMPLICIT
-45 1043 16 EXPLICIT
-46 1043 21 EXPLICIT
-47 1043 23 EXPLICIT
-48 1043 20 EXPLICIT
-49 1043 700 EXPLICIT
-50 1043 701 EXPLICIT
-51 1043 1700 EXPLICIT
-52 1043 1082 EXPLICIT
-53 1043 1083 EXPLICIT
-54 1043 1114 EXPLICIT
-55 1043 1184 EXPLICIT
-56 1043 1186 EXPLICIT
-57 1043 17 EXPLICIT
-58 1043 3802 EXPLICIT
-59 1043 1301 EXPLICIT
-60 1083 1043 ASSIGN
-61 1083 1186 IMPLICIT
-62 1114 1082 ASSIGN
-63 1114 1043 ASSIGN
-64 1114 1083 ASSIGN
-65 1114 1184 IMPLICIT
-66 1184 1082 ASSIGN
-67 1184 1043 ASSIGN
-68 1184 1083 ASSIGN
-69 1184 1114 ASSIGN
-70 1186 1043 ASSIGN
-71 1186 1083 ASSIGN
-72 17 1043 ASSIGN
-73 3802 16 EXPLICIT
-74 3802 21 EXPLICIT
-75 3802 23 EXPLICIT
-76 3802 20 EXPLICIT
-77 3802 700 EXPLICIT
-78 3802 701 EXPLICIT
-79 3802 1700 EXPLICIT
-80 3802 1043 ASSIGN
-81 1301 701 EXPLICIT
-82 1301 1043 ASSIGN
+0 16 23 e
+1 16 1043 a
+2 21 23 i
+3 21 20 i
+4 21 700 i
+5 21 701 i
+6 21 1700 i
+7 21 1043 a
+8 21 1301 i
+9 23 16 e
+10 23 21 a
+11 23 20 i
+12 23 700 i
+13 23 701 i
+14 23 1700 i
+15 23 1043 a
+16 23 1301 i
+17 20 21 a
+18 20 23 a
+19 20 700 i
+20 20 701 i
+21 20 1700 i
+22 20 1043 a
+23 20 1301 i
+24 700 21 a
+25 700 23 a
+26 700 20 a
+27 700 701 i
+28 700 1700 a
+29 700 1043 a
+30 701 21 a
+31 701 23 a
+32 701 20 a
+33 701 700 a
+34 701 1700 a
+35 701 1043 a
+36 1700 21 a
+37 1700 23 a
+38 1700 20 a
+39 1700 700 i
+40 1700 701 i
+41 1700 1043 a
+42 1082 1043 a
+43 1082 1114 i
+44 1082 1184 i
+45 1043 16 e
+46 1043 21 e
+47 1043 23 e
+48 1043 20 e
+49 1043 700 e
+50 1043 701 e
+51 1043 1700 e
+52 1043 1082 e
+53 1043 1083 e
+54 1043 1114 e
+55 1043 1184 e
+56 1043 1186 e
+57 1043 17 e
+58 1043 3802 e
+59 1043 1301 e
+60 1083 1043 a
+61 1083 1186 i
+62 1114 1082 a
+63 1114 1043 a
+64 1114 1083 a
+65 1114 1184 i
+66 1184 1082 a
+67 1184 1043 a
+68 1184 1083 a
+69 1184 1114 a
+70 1186 1043 a
+71 1186 1083 a
+72 17 1043 a
+73 3802 16 e
+74 3802 21 e
+75 3802 23 e
+76 3802 20 e
+77 3802 700 e
+78 3802 701 e
+79 3802 1700 e
+80 3802 1043 a
+81 1301 701 e
+82 1301 1043 a
 
 query TT rowsort
 SELECT s.typname, t.typname
 FROM pg_catalog.pg_cast as c, pg_catalog.pg_type as s, pg_catalog.pg_type as t
 WHERE
-	castcontext = 'IMPLICIT'
+	castcontext = 'i'
 	and c.castsource = s.oid and c.casttarget = t.oid;
 ----
 date	timestamp

e2e_test/batch/catalog/pg_type.slt.part

@@ -5,6 +5,7 @@ SELECT oid, typname, typelem, typnotnull, typtype FROM pg_catalog.pg_type;
 20 int8 0 f b
 21 int2 0 f b
 23 int4 0 f b
+25 text 0 f b
 700 float4 0 f b
 701 float8 0 f b
 1043 varchar 0 f b

e2e_test/source/cdc/cdc.check.slt

@@ -27,7 +27,12 @@ select count(*) as cnt from mytable;
 ----
 4
 
-query IIII
+query VI
+select count(*) from orders_2;
+----
+3
+
+query VII
 select count(*) from shipments_2;
 ----
 3

e2e_test/source/cdc/cdc.load.slt

@@ -88,6 +88,24 @@ create table mytable (
  server.id = '5087'
 );
 
+# Some columns missing and reordered (mysql-cdc)
+statement ok
+create table orders_2 (
+   order_id int,
+   price decimal,
+   customer_name string,
+   PRIMARY KEY (order_id)
+) with (
+ connector = 'mysql-cdc',
+ hostname = 'mysql',
+ port = '3306',
+ username = 'root',
+ password = '123456',
+ database.name = 'mydb',
+ table.name = 'orders',
+ server.id = '5088'
+);
+
 # Some columns missing and reordered (postgres-cdc)
 statement ok
 create table shipments_2 (

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/PostgresValidator.java

@@ -306,7 +306,11 @@ private void validatePrivileges() throws SQLException {
                     stmt.setString(1, tableOwner);
                     var res = stmt.executeQuery();
                     while (res.next()) {
-                        String[] users = (String[]) res.getArray("members").getArray();
+                        var usersArray = res.getArray("members");
+                        if (usersArray == null) {
+                            break;
+                        }
+                        String[] users = (String[]) usersArray.getArray();
                         if (null != users
                                 && Arrays.asList(users)
                                         .contains(userProps.get(DbzConnectorConfig.USER))) {

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/core/SourceHandlerFactory.java

@@ -17,6 +17,7 @@
 import com.risingwave.connector.api.source.SourceHandler;
 import com.risingwave.connector.api.source.SourceTypeE;
 import com.risingwave.connector.source.common.DbzConnectorConfig;
+import java.util.HashMap;
 import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,7 +27,10 @@ public abstract class SourceHandlerFactory {
 
     public static SourceHandler createSourceHandler(
             SourceTypeE source, long sourceId, String startOffset, Map<String, String> userProps) {
-        var config = new DbzConnectorConfig(source, sourceId, startOffset, userProps);
+        // userProps extracted from grpc request, underlying implementation is UnmodifiableMap
+        Map<String, String> modifiableUserProps = new HashMap<>(userProps);
+        modifiableUserProps.put("source.id", Long.toString(sourceId));
+        var config = new DbzConnectorConfig(source, sourceId, startOffset, modifiableUserProps);
         LOG.info("resolved config for source#{}: {}", sourceId, config.getResolvedDebeziumProps());
         return new DbzSourceHandler(config);
     }

java/connector-node/risingwave-connector-service/src/main/resources/mysql.properties

@@ -16,6 +16,6 @@ table.include.list=${database.name}.${table.name}
 # default to disable schema change events
 include.schema.changes=${debezium.include.schema.changes:-false}
 database.server.id=${server.id}
-database.server.name=RW_CDC_${database.name}.${table.name}
+database.server.name=RW_CDC_${database.name}.${table.name}.${source.id}
 
 name=${hostname}:${port}:${database.name}.${table.name}

src/common/src/types/postgres_type.rs

@@ -52,6 +52,7 @@ impl DataType {
             21 => Ok(DataType::Int16),
             23 => Ok(DataType::Int32),
             20 => Ok(DataType::Int64),
+            25 => Ok(DataType::Varchar), // workaround to support text in extended mode.
             700 => Ok(DataType::Float32),
             701 => Ok(DataType::Float64),
             1700 => Ok(DataType::Decimal),

src/expr/src/sig/cast.rs

@@ -29,10 +29,12 @@ pub struct CastSig {
 /// that allows explicit casts, but not vice versa. See details in
 /// [PG](https://www.postgresql.org/docs/current/catalog-pg-cast.html).
 #[derive(Clone, Copy, Debug, Display, Eq, Ord, PartialEq, PartialOrd)]
-#[display(style = "UPPERCASE")]
 pub enum CastContext {
+    #[display("i")]
     Implicit,
+    #[display("a")]
     Assign,
+    #[display("e")]
     Explicit,
 }
 

src/frontend/src/binder/relation/join.rs

@@ -35,8 +35,7 @@ impl RewriteExprsRecursive for BoundJoin {
     fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
         self.left.rewrite_exprs_recursive(rewriter);
         self.right.rewrite_exprs_recursive(rewriter);
-        let dummy = ExprImpl::literal_bool(true);
-        self.cond = rewriter.rewrite_expr(std::mem::replace(&mut self.cond, dummy));
+        self.cond = rewriter.rewrite_expr(self.cond.take());
     }
 }
 

src/frontend/src/catalog/system_catalog/pg_catalog/pg_type.rs

@@ -56,6 +56,9 @@ pub const PG_TYPE_DATA: &[(i32, &str)] = &[
     (20, "int8"),
     (21, "int2"),
     (23, "int4"),
+    // Note: rw doesn't support `text` type, returning it is just a workaround to be compatible
+    // with PostgreSQL.
+    (25, "text"),
     (700, "float4"),
     (701, "float8"),
     (1043, "varchar"),