Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

…nto li0k/storage_split_group_v2
risingwavelabs · Sep 29, 2024 · fe8cc7e · fe8cc7e
2 parents 48eb6f3 + 04af708
commit fe8cc7e
Show file tree

Hide file tree

Showing 175 changed files with 2,446 additions and 1,336 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/docker/docker-compose-distributed-etcd.yml b/docker/docker-compose-distributed-etcd.yml
@@ -1,7 +1,7 @@
 ---
 version: "3"
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   compactor-0:
     <<: *image

diff --git a/docker/docker-compose-distributed.yml b/docker/docker-compose-distributed.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   compactor-0:
     <<: *image

diff --git a/docker/docker-compose-etcd.yml b/docker/docker-compose-etcd.yml
@@ -1,7 +1,7 @@
 ---
 version: "3"
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-azblob.yml b/docker/docker-compose-with-azblob.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-gcs.yml b/docker/docker-compose-with-gcs.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-local-fs.yml b/docker/docker-compose-with-local-fs.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-obs.yml b/docker/docker-compose-with-obs.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-oss.yml b/docker/docker-compose-with-oss.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-s3.yml b/docker/docker-compose-with-s3.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose-with-sqlite.yml b/docker/docker-compose-with-sqlite.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -1,6 +1,6 @@
 ---
 x-image: &image
-  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.0}
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:v2.0.1}
 services:
   risingwave-standalone:
     <<: *image

diff --git a/e2e_test/batch/catalog/pg_settings.slt.part b/e2e_test/batch/catalog/pg_settings.slt.part
@@ -58,6 +58,7 @@ user sink_decouple
 user source_rate_limit
 user standard_conforming_strings
 user statement_timeout
+user streaming_max_parallelism
 user streaming_parallelism
 user streaming_use_arrangement_backfill
 user streaming_use_snapshot_backfill

diff --git a/e2e_test/streaming/on_conflict.slt b/e2e_test/streaming/on_conflict.slt
@@ -2,7 +2,7 @@ statement ok
 SET RW_IMPLICIT_FLUSH TO true;
 
 statement ok
-create table t1 (v1 int, v2 int, v3 int, primary key(v1)) APPEND ONLY on conflict ignore;
+create table t1 (v1 int, v2 int, v3 int, primary key(v1)) APPEND ONLY on conflict do nothing;
 
 statement ok
 insert into t1 values (1,4,2), (2,3,3);
@@ -30,7 +30,7 @@ statement error
 create table t2 (v1 int, v2 int, v3 int, primary key(v1)) APPEND ONLY on conflict overwrite;
 
 statement ok
-create table t2 (v1 int, v2 int, v3 int, primary key(v1)) on conflict overwrite;
+create table t2 (v1 int, v2 int, v3 int, primary key(v1)) on conflict do update full;
 
 statement ok
 insert into t2 values (1,4,2), (2,3,3);

diff --git a/e2e_test/streaming/with_version_column.slt b/e2e_test/streaming/with_version_column.slt
@@ -12,7 +12,7 @@ statement error
 create table t0 (v1 int, v2 int, v3 int, v4 bool, primary key(v1)) on conflict do update if not null with version column v4;
 
 statement error
-create table t0 (v1 int, v2 int, v3 int, v4 int, primary key(v1)) on conflict ignore with version column(v4);
+create table t0 (v1 int, v2 int, v3 int, v4 int, primary key(v1)) on conflict do nothing with version column(v4);
 
 statement ok
 create table t1 (v1 int, v2 int, v3 int, v4 int, primary key(v1)) on conflict do update if not null with version column(v4);
@@ -62,7 +62,7 @@ statement ok
 SET RW_IMPLICIT_FLUSH TO true;
 
 statement ok
-create table t2 (v1 int, v2 int, v3 int, v4 int, primary key(v1)) on conflict overwrite with version column(v4);
+create table t2 (v1 int, v2 int, v3 int, v4 int, primary key(v1)) on conflict do update full with version column(v4);
 
 statement ok
 insert into t2 values (1,null,2, 4), (2,3,null, 1);

diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -2442,6 +2442,20 @@ def section_hummock_read(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Safe Version Fetch Count",
+                    "",
+                    [
+                        panels.target(
+                            f"{metric('state_store_safe_version_hit')}",
+                            "",
+                        ),
+                        panels.target(
+                            f"{metric('state_store_safe_version_miss')}",
+                            "",
+                        ),
+                    ],
+                ),
             ],
         )
     ]
@@ -3206,6 +3220,10 @@ def section_hummock_manager(outer_panels):
                             f"{metric('storage_current_version_object_count')}",
                             "referenced by current version",
                         ),
+                        panels.target(
+                            f"{metric('storage_time_travel_object_count')}",
+                            "referenced by time travel",
+                        ),
                         panels.target(
                             f"{metric('storage_total_object_count')}",
                             "all objects (including dangling ones)",
@@ -3364,6 +3382,33 @@ def section_hummock_manager(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_latency(
+                    "Time Travel Replay Latency",
+                    "The latency of replaying a hummock version for time travel",
+                    quantile(
+                        lambda quantile, legend: panels.target(
+                            f"histogram_quantile({quantile}, sum(rate({metric('storage_time_travel_version_replay_latency_bucket')}[$__rate_interval])) by (le))",
+                            f"time_travel_version_replay_latency_p{legend}",
+                        ),
+                        [50, 90, 99, "max"],
+                    )
+                    + [
+                        panels.target(
+                            f"rate({metric('storage_time_travel_version_replay_latency_sum')}[$__rate_interval]) / rate({metric('storage_time_travel_version_replay_latency_count')}[$__rate_interval]) > 0",
+                            "time_travel_version_replay_avg",
+                        ),
+                    ],
+                ),
+                panels.timeseries_ops(
+                    "Time Travel Replay Ops",
+                    "The frequency of replaying a hummock version for time travel",
+                    [
+                        panels.target(
+                            f"sum(rate({metric('storage_time_travel_version_replay_latency_count')}[$__rate_interval]))",
+                            "time_travel_version_replay_ops",
+                        ),
+                    ],
+                ),
             ],
         )
     ]
@@ -3498,11 +3543,6 @@ def section_grpc_meta_hummock_manager(outer_panels):
                     "UnpinVersionBefore",
                     "path='/meta.HummockManagerService/UnpinVersionBefore'",
                 ),
-                grpc_metrics_target(
-                    panels,
-                    "UnpinSnapshotBefore",
-                    "path='/meta.HummockManagerService/UnpinSnapshotBefore'",
-                ),
                 grpc_metrics_target(
                     panels,
                     "ReportCompactionTasks",

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/java/com_risingwave_java_binding_Binding.h b/java/com_risingwave_java_binding_Binding.h
diff --git a/...a-binding-integration-test/src/main/java/com/risingwave/java/binding/HummockReadDemo.java b/...a-binding-integration-test/src/main/java/com/risingwave/java/binding/HummockReadDemo.java
@@ -50,7 +50,10 @@ public static void main(String[] args) {
             HummockVersion version = metaClient.pinVersion();
             Table tableCatalog = metaClient.getTable(dbName, tableName);
 
-            int vnodeCount = Binding.vnodeCount();
+            int vnodeCount = Binding.defaultVnodeCount();
+            if (tableCatalog.hasMaybeVnodeCount()) {
+                vnodeCount = tableCatalog.getMaybeVnodeCount();
+            }
 
             List<Integer> vnodeList = new ArrayList<>();
             for (int i = 0; i < vnodeCount; i++) {

diff --git a/java/java-binding/src/main/java/com/risingwave/java/binding/Binding.java b/java/java-binding/src/main/java/com/risingwave/java/binding/Binding.java
@@ -33,7 +33,11 @@ public static native void tracingSlf4jEvent(
 
     public static native boolean tracingSlf4jEventEnabled(int level);
 
-    public static native int vnodeCount();
+    /**
+     * Used to get the default number of vnodes for a table, if its `maybeVnodeCount` field is not
+     * set.
+     */
+    public static native int defaultVnodeCount();
 
     static native long iteratorNewStreamChunk(long pointer);
 

diff --git a/java/pom.xml b/java/pom.xml
@@ -78,7 +78,7 @@
         <jackson.version>2.15.0</jackson.version>
         <spark_sql.version>3.3.1</spark_sql.version>
         <hadoop.version>3.4.0</hadoop.version>
-        <elasticsearch.version>7.17.19</elasticsearch.version>
+        <elasticsearch.version>7.17.23</elasticsearch.version>
         <opensearch.version>2.11.1</opensearch.version>
         <datastax.version>4.15.0</datastax.version>
         <flink.version>1.18.0</flink.version>

diff --git a/proto/common.proto b/proto/common.proto
@@ -105,9 +105,14 @@ message WorkerSlotMapping {
   repeated uint64 data = 2;
 }
 
+message BatchQueryCommittedEpoch {
+  uint64 epoch = 1;
+  uint64 hummock_version_id = 2;
+}
+
 message BatchQueryEpoch {
   oneof epoch {
-    uint64 committed = 1;
+    BatchQueryCommittedEpoch committed = 1;
     uint64 current = 2;
     uint64 backup = 3;
     uint64 time_travel = 4;

diff --git a/proto/hummock.proto b/proto/hummock.proto
@@ -553,6 +553,8 @@ message VacuumTask {
 message FullScanTask {
   uint64 sst_retention_time_sec = 1;
   optional string prefix = 2;
+  optional string start_after = 3;
+  optional uint64 limit = 4;
 }
 
 // Cancel compact task
@@ -588,6 +590,7 @@ message ReportFullScanTaskRequest {
   uint64 total_object_count = 2;
   // Total size of objects before filtered by conditions specified by FullScanTask.
   uint64 total_object_size = 3;
+  optional string next_start_after = 4;
 }
 
 message ReportFullScanTaskResponse {