Merge branch 'main' into xxh/decouple_default_1

.github/workflows/stale.yml

@@ -24,10 +24,26 @@ jobs:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
         stale-issue-message: >
           This issue has been open for 60 days with no activity.
-          Could you please update the status? Feel free to continue discussion or close as not planned.
+
+          If you think it is still relevant today, and needs to be done *in the near future*, you can comment to update the status, or just manually remove the `no-issue-activity` label.
+
+          You can also confidently close this issue as not planned to keep our backlog clean.
+          Don't worry if you think the issue is still valuable to continue in the future.
+          It's searchable and can be reopened when it's time. 😄
         stale-pr-message: >
           This PR has been open for 60 days with no activity.
-          Could you please update the status? Feel free to ping a reviewer if you are waiting for review.
+
+          If it's blocked by code review, feel free to ping a reviewer or ask someone else to review it.
+
+          If you think it is still relevant today, and have time to work on it *in the near future*, you can comment to update the status, or just manually remove the `no-pr-activity` label.
+
+          You can also confidently close this PR to keep our backlog clean. (If no further action taken, the PR will be automatically closed after 7 days. Sorry!  🙏)
+          Don't worry if you think the PR is still valuable to continue in the future.
+          It's searchable and can be reopened when it's time. 😄
+        close-pr-message: >
+          Close this PR as there's no further actions taken after it is marked as stale for 7 days. Sorry! 🙏
+
+          You can reopen it when you have time to continue working on it.
         stale-issue-label: 'no-issue-activity'
         stale-pr-label: 'no-pr-activity'
         days-before-close: -1

e2e_test/batch/over_window/main.slt.part

@@ -1 +1,2 @@
 include ./generated/main.slt.part
+include ./session/mod.slt.part

e2e_test/batch/over_window/session/mod.slt.part

@@ -0,0 +1,61 @@
+# Because currently general streaming version of session window is not supported yet,
+# we only add e2e for batch mode.
+
+statement ok
+create table t (
+    tm timestamp,
+    foo int,
+    bar int
+);
+
+statement ok
+create view v1 as
+select
+    *,
+    first_value(tm) over (partition by bar order by tm session with gap '10 minutes') as window_start,
+    last_value(tm) over (partition by bar order by tm session with gap '10 minutes') as window_end
+from t;
+
+statement ok
+insert into t values
+  ('2023-05-06 16:51:00', 1, 100)
+, ('2023-05-06 16:56:00', 8, 100)
+, ('2023-05-06 17:30:00', 3, 200)
+, ('2023-05-06 17:35:00', 5, 100)
+, ('2023-05-06 17:59:00', 4, 100)
+, ('2023-05-06 18:01:00', 6, 200)
+;
+
+query TiiTT
+select * from v1 order by tm;
+----
+2023-05-06 16:51:00	1	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 16:56:00	8	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 17:30:00	3	200	2023-05-06 17:30:00	2023-05-06 17:30:00
+2023-05-06 17:35:00	5	100	2023-05-06 17:35:00	2023-05-06 17:35:00
+2023-05-06 17:59:00	4	100	2023-05-06 17:59:00	2023-05-06 17:59:00
+2023-05-06 18:01:00	6	200	2023-05-06 18:01:00	2023-05-06 18:01:00
+
+statement ok
+insert into t values
+  ('2023-05-06 18:08:00', 7, 100)
+, ('2023-05-06 18:10:00', 9, 200)
+;
+
+query TiiTT
+select * from v1 order by tm;
+----
+2023-05-06 16:51:00	1	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 16:56:00	8	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 17:30:00	3	200	2023-05-06 17:30:00	2023-05-06 17:30:00
+2023-05-06 17:35:00	5	100	2023-05-06 17:35:00	2023-05-06 17:35:00
+2023-05-06 17:59:00	4	100	2023-05-06 17:59:00	2023-05-06 18:08:00
+2023-05-06 18:01:00	6	200	2023-05-06 18:01:00	2023-05-06 18:10:00
+2023-05-06 18:08:00	7	100	2023-05-06 17:59:00	2023-05-06 18:08:00
+2023-05-06 18:10:00	9	200	2023-05-06 18:01:00	2023-05-06 18:10:00
+
+statement ok
+drop view v1;
+
+statement ok
+drop table t;

e2e_test/source/basic/nosim_kafka.slt

@@ -39,6 +39,22 @@ CREATE TABLE kafka_json_schema_plain with (
     kafka.scan.startup.mode = 'earliest'
 ) FORMAT PLAIN ENCODE JSON (schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}');
 
+
+query
+describe kafka_json_schema_plain;
+----
+dimensions (empty) false NULL
+map jsonb false NULL
+notMap (empty) false NULL
+price double precision false NULL
+productId bigint false NULL
+productName character varying false NULL
+tags character varying[] false NULL
+_row_id serial true NULL
+primary key _row_id NULL NULL
+distribution key _row_id NULL NULL
+table description kafka_json_schema_plain NULL NULL
+
 statement ok
 CREATE TABLE kafka_json_schema_upsert (PRIMARY KEY(rw_key))
 INCLUDE KEY AS rw_key
@@ -83,10 +99,10 @@ select count(*) from debezium_compact;
 
 query TFITT
 select
-  "dimensions", "price", "productId", "productName", "tags"
+  *
 from kafka_json_schema_plain
 ----
-(9.5,7,12) 12.5 1 An ice sculpture {cold,ice}
+(9.5,7,12) {"foo": "bar"} (b) 12.5 1 An ice sculpture {cold,ice}
 
 query TFITT
 select

e2e_test/streaming/eowc/eowc_over_window.slt

@@ -42,6 +42,15 @@ select
 from t
 emit on window close;
 
+statement ok
+create materialized view mv4 as
+select
+    *,
+    first_value(tm) over (partition by bar order by tm session with gap '10 minutes') as window_start,
+    last_value(tm) over (partition by bar order by tm session with gap '10 minutes') as window_end
+from t
+emit on window close;
+
 statement ok
 insert into t values
   ('2023-05-06 16:51:00', 1, 100)
@@ -71,6 +80,12 @@ select * from mv3 order by tm;
 2023-05-06 17:30:00  3  200  1
 2023-05-06 17:35:00  5  100  3
 
+query TiiTT
+select * from mv4 order by tm;
+----
+2023-05-06 16:51:00	1	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 16:56:00	8	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+
 statement ok
 insert into t values
   ('2023-05-06 18:10:00', 7, 100)
@@ -100,6 +115,14 @@ select * from mv3 order by tm;
 2023-05-06 17:59:00  4  100  4
 2023-05-06 18:01:00  6  200  2
 
+query TiiTT
+select * from mv4 order by tm;
+----
+2023-05-06 16:51:00	1	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 16:56:00	8	100	2023-05-06 16:51:00	2023-05-06 16:56:00
+2023-05-06 17:30:00	3	200	2023-05-06 17:30:00	2023-05-06 17:30:00
+2023-05-06 17:35:00	5	100	2023-05-06 17:35:00	2023-05-06 17:35:00
+
 statement ok
 drop materialized view mv1;
 
@@ -109,5 +132,8 @@ drop materialized view mv2;
 statement ok
 drop materialized view mv3;
 
+statement ok
+drop materialized view mv4;
+
 statement ok
 drop table t;

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/JdbcUtils.java

@@ -25,6 +25,9 @@
 
 public abstract class JdbcUtils {
 
+    static final int CONNECTION_TIMEOUT = 30;
+    static final int SOCKET_TIMEOUT = 300;
+
     public static Optional<JdbcDialectFactory> getDialectFactory(String jdbcUrl) {
         if (jdbcUrl.startsWith("jdbc:mysql")) {
             return Optional.of(new MySqlDialectFactory());
@@ -43,6 +46,16 @@ public static Connection getConnection(String jdbcUrl) throws SQLException {
         // https://jdbc.postgresql.org/documentation/use/
         // https://dev.mysql.com/doc/connectors/en/connector-j-connp-props-networking.html#cj-conn-prop_tcpKeepAlive
         props.setProperty("tcpKeepAlive", "true");
+
+        // default timeout in seconds
+        boolean isPg = jdbcUrl.startsWith("jdbc:postgresql");
+
+        // postgres use seconds and mysql use milliseconds
+        int connectTimeout = isPg ? CONNECTION_TIMEOUT : CONNECTION_TIMEOUT * 1000;
+        int socketTimeout = isPg ? SOCKET_TIMEOUT : SOCKET_TIMEOUT * 1000;
+        props.setProperty("connectTimeout", String.valueOf(connectTimeout));
+        props.setProperty("socketTimeout", String.valueOf(socketTimeout));
+
         var conn = DriverManager.getConnection(jdbcUrl, props);
         // disable auto commit can improve performance
         conn.setAutoCommit(false);

java/pom.xml

@@ -83,7 +83,7 @@
         <flink.version>1.18.0</flink.version>
         <testcontainers.version>1.17.6</testcontainers.version>
         <postgresql.version>42.5.5</postgresql.version>
-        <mysql.connector.java.version>8.0.33</mysql.connector.java.version>
+        <mysql.connector.java.version>8.3.0</mysql.connector.java.version>
         <mongodb.driver.sync.version>4.11.1</mongodb.driver.sync.version>
         <sqlite.version>3.45.0.0</sqlite.version>
         <aws.version>2.21.42</aws.version>
@@ -570,4 +570,4 @@
             <url>https://s01.oss.sonatype.org/service/local/staging/deploy/maven2/</url>
         </repository>
     </distributionManagement>
-</project>
+</project>

proto/backup_service.proto

@@ -2,6 +2,8 @@ syntax = "proto3";
 
 package backup_service;
 
+import "hummock.proto";
+
 option java_package = "com.risingwave.proto";
 option optimize_for = SPEED;
 
@@ -50,6 +52,7 @@ message MetaSnapshotMetadata {
   optional uint32 format_version = 5;
   optional string remarks = 6;
   optional string rw_version = 7;
+  map<uint32, hummock.StateTableInfo> state_table_info = 8;
 }
 
 service BackupService {

proto/expr.proto

@@ -452,6 +452,7 @@ message WindowFrame {
 
     TYPE_ROWS = 5;
     TYPE_RANGE = 10;
+    TYPE_SESSION = 15;
   }
   enum BoundType {
     BOUND_TYPE_UNSPECIFIED = 0;
@@ -497,6 +498,13 @@ message WindowFrame {
     BoundType type = 1;
     optional data.Datum offset = 3;
   }
+  message SessionFrameBounds {
+    data.Datum gap = 1;
+
+    data.DataType order_data_type = 10;
+    common.OrderType order_type = 15;
+    data.DataType gap_data_type = 20;
+  }
 
   Type type = 1;
 
@@ -508,6 +516,7 @@ message WindowFrame {
   oneof bounds {
     RowsFrameBounds rows = 10;
     RangeFrameBounds range = 15;
+    SessionFrameBounds session = 20;
   }
 }
 

proto/hummock.proto

@@ -69,25 +69,28 @@ message IntraLevelDelta {
 enum CompatibilityVersion {
   VERSION_UNSPECIFIED = 0;
   NO_TRIVIAL_SPLIT = 1;
+  NO_MEMBER_TABLE_IDS = 2;
 }
 
 message GroupConstruct {
   CompactionConfig group_config = 1;
   // If parent_group_id is not 0, it means parent_group_id splits into parent_group_id and this group, so this group is not empty initially.
   uint64 parent_group_id = 2;
-  repeated uint32 table_ids = 3;
+  repeated uint32 table_ids = 3 [deprecated = true];
   uint64 group_id = 4;
   uint64 new_sst_start_id = 5;
   CompatibilityVersion version = 6;
 }
 
 message GroupMetaChange {
-  repeated uint32 table_ids_add = 1;
-  repeated uint32 table_ids_remove = 2;
+  option deprecated = true;
+  repeated uint32 table_ids_add = 1 [deprecated = true];
+  repeated uint32 table_ids_remove = 2 [deprecated = true];
 }
 
 message GroupTableChange {
-  repeated uint32 table_ids = 1;
+  option deprecated = true;
+  repeated uint32 table_ids = 1 [deprecated = true];
   uint64 target_group_id = 2;
   uint64 origin_group_id = 3;
   uint64 new_sst_start_id = 4;
@@ -102,7 +105,7 @@ message GroupDelta {
     GroupConstruct group_construct = 2;
     GroupDestroy group_destroy = 3;
     GroupMetaChange group_meta_change = 4;
-    GroupTableChange group_table_change = 5;
+    GroupTableChange group_table_change = 5 [deprecated = true];
   }
 }
 
@@ -152,11 +155,13 @@ message TableChangeLog {
 message StateTableInfo {
   uint64 committed_epoch = 1;
   uint64 safe_epoch = 2;
+  uint64 compaction_group_id = 3;
 }
 
 message StateTableInfoDelta {
   uint64 committed_epoch = 1;
   uint64 safe_epoch = 2;
+  uint64 compaction_group_id = 3;
 }
 
 message HummockVersion {
@@ -165,7 +170,7 @@ message HummockVersion {
     OverlappingLevel l0 = 2;
     uint64 group_id = 3;
     uint64 parent_group_id = 4;
-    repeated uint32 member_table_ids = 5;
+    repeated uint32 member_table_ids = 5 [deprecated = true];
   }
   uint64 id = 1;
   // Levels of each compaction group
@@ -340,6 +345,7 @@ message CompactTask {
     JOIN_HANDLE_FAILED = 11;
     TRACK_SST_OBJECT_ID_FAILED = 12;
     NO_AVAIL_CPU_RESOURCE_CANCELED = 13;
+    HEARTBEAT_PROGRESS_CANCELED = 14;
   }
   // SSTs to be compacted, which will be removed from LSM after compaction
   repeated InputLevel input_ssts = 1;

scripts/source/test_data/kafka_json_schema.1

@@ -1,2 +1,2 @@
-{"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"https://example.com/product.schema.json","title":"Product","description":"A product from Acme's catalog","type":"object","properties":{"productId":{"description":"The unique identifier for a product","type":"integer"},"productName":{"description":"Name of the product","type":"string"},"price":{"description":"The price of the product","type":"number","exclusiveMinimum":0},"tags":{"description":"Tags for the product","type":"array","items":{"type":"string"},"minItems":1,"uniqueItems":true},"dimensions":{"type":"object","properties":{"length":{"type":"number"},"width":{"type":"number"},"height":{"type":"number"}},"required":["length","width","height"]}},"required":["productId","productName","price"]}
-{"productId":1,"productName":"An ice sculpture","price":12.5,"tags":["cold","ice"],"dimensions":{"length":7,"width":12,"height":9.5}}
+{"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"https://example.com/product.schema.json","title":"Product","description":"A product from Acme's catalog","type":"object","properties":{"productId":{"description":"The unique identifier for a product","type":"integer"},"productName":{"description":"Name of the product","type":"string"},"price":{"description":"The price of the product","type":"number","exclusiveMinimum":0},"tags":{"description":"Tags for the product","type":"array","items":{"type":"string"},"minItems":1,"uniqueItems":true},"dimensions":{"type":"object","properties":{"length":{"type":"number"},"width":{"type":"number"},"height":{"type":"number"}},"required":["length","width","height"]},"map":{"type":"object","additionalProperties":{"type":"string"}},"notMap":{"type":"object","additionalProperties":{"type":"string"},"properties":{"a":{"type":"string"}}}},"required":["productId","productName","price"]}
+{"productId":1,"productName":"An ice sculpture","price":12.5,"tags":["cold","ice"],"dimensions":{"length":7,"width":12,"height":9.5},"map":{"foo":"bar"},"notMap":{"a":"b","ignored":"c"}}