Merge branch 'main' into add-sqrt

.github/workflows/connector-node-integration.yml

@@ -77,13 +77,6 @@ jobs:
             echo "Jdbc sink test failed"
             exit 1
           fi
-
-          if python3 integration_tests.py --print_sink; then
-            echo "Print sink test passed"
-          else
-            echo "Print sink test failed"
-            exit 1
-          fi
           cd ..
           echo "all jdbc tests passed"
           

Cargo.toml

@@ -125,4 +125,4 @@ tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "0c25710"
 tokio-retry = { git = "https://github.com/madsim-rs/rust-tokio-retry.git", rev = "95e2fd3" }
 tokio-postgres = { git = "https://github.com/madsim-rs/rust-postgres.git", rev = "87ca1dc" }
 postgres-types = { git = "https://github.com/madsim-rs/rust-postgres.git", rev = "87ca1dc" }
-madsim-rdkafka = { git = "https://github.com/madsim-rs/madsim.git", rev = "43e025d" }
+madsim-rdkafka = { git = "https://github.com/madsim-rs/madsim.git", rev = "52adb98" }

e2e_test/ddl/invalid_operation.slt

@@ -184,7 +184,7 @@ drop view src;
 
 # 4.6 sink 
 statement ok
-CREATE SINK sink FROM mv WITH (connector='console', prefix='DEBUG', suffix='');
+CREATE SINK sink FROM mv WITH (connector='blackhole');
 
 # FIXME: improve the error message
 statement error not found

grafana/risingwave-dev-dashboard.dashboard.py

@@ -996,15 +996,15 @@ def section_streaming_actors(outer_panels):
                     [
                         panels.target(
                             f"rate({metric('stream_join_lookup_miss_count')}[$__rate_interval])",
-                            "cache miss {{actor_id}} {{side}}",
+                            "cache miss table - {{side}} side, join_table_id {{join_table_id}} degree_table_id {{degree_table_id}} actor {{actor_id}} ",
                         ),
                         panels.target(
                             f"rate({metric('stream_join_lookup_total_count')}[$__rate_interval])",
-                            "total lookups {{actor_id}} {{side}}",
+                            "total lookups {{side}} side, join_table_id {{join_table_id}} degree_table_id {{degree_table_id}} actor {{actor_id}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_join_insert_cache_miss_count')}[$__rate_interval])",
-                            "cache miss when insert{{actor_id}} {{side}}",
+                            "cache miss when insert {{side}} side, join_table_id {{join_table_id}} degree_table_id {{degree_table_id}} actor {{actor_id}}",
                         ),
                     ],
                 ),
@@ -1080,11 +1080,11 @@ def section_streaming_actors(outer_panels):
                     [
                         panels.target(
                             f"rate({metric('stream_agg_lookup_miss_count')}[$__rate_interval])",
-                            "cache miss {{actor_id}}",
+                            "cache miss - table {{table_id}} actor {{actor_id}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_agg_lookup_total_count')}[$__rate_interval])",
-                            "total lookups {{actor_id}}",
+                            "total lookups - table {{table_id}} actor {{actor_id}}",
                         ),
                     ],
                 ),
@@ -1094,11 +1094,11 @@ def section_streaming_actors(outer_panels):
                     [
                         panels.target(
                             f"rate({metric('stream_agg_chunk_lookup_miss_count')}[$__rate_interval])",
-                            "chunk-level cache miss {{actor_id}}",
+                            "chunk-level cache miss  - table {{table_id}} actor {{actor_id}}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_agg_chunk_lookup_total_count')}[$__rate_interval])",
-                            "chunk-level total lookups {{actor_id}}",
+                            "chunk-level total lookups  - table {{table_id}} actor {{actor_id}}",
                         ),
                     ],
                 ),
@@ -1107,7 +1107,7 @@ def section_streaming_actors(outer_panels):
                     "The number of keys cached in each hash aggregation executor's executor cache.",
                     [
                         panels.target(f"{metric('stream_agg_cached_keys')}",
-                                      "{{actor_id}}"),
+                                      "table {{table_id}} actor {{actor_id}}"),
                     ],
                 ),
             ],

grafana/risingwave-user-dashboard.dashboard.py

@@ -240,19 +240,19 @@ def section_memory(outer_panels):
                     [
                         panels.target(
                             f"rate({metric('stream_join_lookup_miss_count')}[$__rate_interval])",
-                            "Join - cache miss {{actor_id}} {{side}}",
+                            "Join - cache miss - {{side}} side, join_table_id {{join_table_id}} degree_table_id {{degree_table_id}} actor {{actor_id}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_join_lookup_total_count')}[$__rate_interval])",
-                            "Join - total lookups {{actor_id}} {{side}}",
+                            "Join - total lookups - {{side}} side, join_table_id {{join_table_id}} degree_table_id {{degree_table_id}} actor {{actor_id}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_agg_lookup_miss_count')}[$__rate_interval])",
-                            "Agg - cache miss {{actor_id}}",
+                            "Agg - cache miss - table {{table_id}} actor {{actor_id}}",
                         ),
                         panels.target(
                             f"rate({metric('stream_agg_lookup_total_count')}[$__rate_interval])",
-                            "Agg - total lookups {{actor_id}}",
+                            "Agg - total lookups - table {{table_id}} actor {{actor_id}}",
                         ),
                     ],
                 ),

java/connector-node/python-client/integration_tests.py

@@ -156,10 +156,6 @@ def validate_jdbc_sink(input_file):
                 exit(1)
 
 
-def test_print_sink(input_file):
-    test_sink("print", {}, input_file)
-
-
 def test_iceberg_sink(input_file):
     test_sink("iceberg",
               {"type":"append-only",
@@ -192,7 +188,6 @@ def test_deltalake_sink(input_file):
     parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
     parser.add_argument('--file_sink', action='store_true', help="run file sink test")
     parser.add_argument('--jdbc_sink', action='store_true', help="run jdbc sink test")
-    parser.add_argument('--print_sink', action='store_true', help="run print sink test")
     parser.add_argument('--iceberg_sink', action='store_true', help="run iceberg sink test")
     parser.add_argument('--upsert_iceberg_sink', action='store_true', help="run upsert iceberg sink test")
     parser.add_argument('--deltalake_sink', action='store_true', help="run deltalake sink test")
@@ -202,8 +197,6 @@ def test_deltalake_sink(input_file):
         test_file_sink(args.input_file)
     if args.jdbc_sink:
         test_jdbc_sink(args.input_file)
-    if args.print_sink:
-        test_print_sink(args.input_file)
     if args.iceberg_sink:
         test_iceberg_sink(args.input_file)
     if args.upsert_iceberg_sink:

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java

@@ -21,9 +21,6 @@
 public class SinkUtils {
     public static SinkFactory getSinkFactory(String sinkType) {
         switch (sinkType) {
-            case "print":
-            case "connector-node-print":
-                return new PrintSinkFactory();
             case "file":
                 return new FileSinkFactory();
             case "jdbc":