Merge branch 'main' of github.com:singularity-data/risingwave into yu…

…hao/pg_get_viewdef
risingwavelabs · Jan 3, 2024 · 163d65e · 163d65e
2 parents cee6ff6 + 0523b41
commit 163d65e
Show file tree

Hide file tree

Showing 67 changed files with 793 additions and 601 deletions.
diff --git a/ci/scripts/connector-node-integration-test.sh b/ci/scripts/connector-node-integration-test.sh
@@ -113,45 +113,6 @@ for ((i=0; i<${#type[@]}; i++)); do
       echo "File sink ${type[i]} test failed"
       exit 1
     fi
-
-    # test upsert mode
-    echo "--- running iceberg upsert mode ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_iceberg
-    if python3 integration_tests.py --upsert_iceberg_sink ${upsert_sink_input_feature[i]}; then
-      python3 pyspark-util.py test_upsert_iceberg --input_file="./data/upsert_sink_input.json"
-      echo "Upsert iceberg sink ${type[i]} test passed"
-    else
-      echo "Upsert iceberg sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py drop_iceberg
-
-    # test append-only mode
-    echo "--- running iceberg append-only mode ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_iceberg
-    if python3 integration_tests.py --iceberg_sink ${sink_input_feature[i]}; then
-      python3 pyspark-util.py test_iceberg
-      echo "Iceberg sink ${type[i]} test passed"
-    else
-      echo "Iceberg sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py drop_iceberg
-
-    # test append-only mode
-    echo "--- running deltalake append-only mod ${type[i]} integration tests"
-    cd ${RISINGWAVE_ROOT}/java/connector-node/python-client
-    python3 pyspark-util.py create_deltalake
-    if python3 integration_tests.py --deltalake_sink ${sink_input_feature[i]}; then
-      python3 pyspark-util.py test_deltalake
-      echo "Deltalake sink ${type[i]} test passed"
-    else
-      echo "Deltalake sink ${type[i]} test failed"
-      exit 1
-    fi
-    python3 pyspark-util.py clean_deltalake
 done
 
 

diff --git a/ci/scripts/e2e-sink-test.sh b/ci/scripts/e2e-sink-test.sh
@@ -61,6 +61,7 @@ sqllogictest -p 4566 -d dev './e2e_test/sink/append_only_sink.slt'
 sqllogictest -p 4566 -d dev './e2e_test/sink/create_sink_as.slt'
 sqllogictest -p 4566 -d dev './e2e_test/sink/blackhole_sink.slt'
 sqllogictest -p 4566 -d dev './e2e_test/sink/remote/types.slt'
+sqllogictest -p 4566 -d dev './e2e_test/sink/sink_into_table.slt'
 sleep 1
 
 echo "--- testing remote sinks"

diff --git a/docker/dashboards/risingwave-dev-dashboard.json b/docker/dashboards/risingwave-dev-dashboard.json
diff --git a/docker/dashboards/risingwave-user-dashboard.json b/docker/dashboards/risingwave-user-dashboard.json
diff --git a/e2e_test/sink/deltalake_rust_sink.slt b/e2e_test/sink/deltalake_rust_sink.slt
@@ -7,7 +7,7 @@ CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
 statement ok
 create sink s6 as select * from mv6
 with (
-    connector = 'deltalake_rust',
+    connector = 'deltalake',
     type = 'append-only',
     force_append_only = 'true',
     location = 's3a://deltalake/deltalake-test',

diff --git a/e2e_test/sink/iceberg_sink.slt b/e2e_test/sink/iceberg_sink.slt
@@ -6,15 +6,17 @@ CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
 
 statement ok
 CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH (
-    connector = 'iceberg_java',
+    connector = 'iceberg',
     type = 'upsert',
     primary_key = 'v1',
     warehouse.path = 's3://iceberg',
     s3.endpoint = 'http://127.0.0.1:9301',
     s3.access.key = 'hummockadmin',
     s3.secret.key = 'hummockadmin',
-    database.name='demo_db',
-    table.name='demo_table'
+    s3.region = 'us-east-1',
+    catalog.type = 'storage',
+    database.name='demo',
+    table.name='demo_db.demo_table'
 );
 
 statement ok

diff --git a/e2e_test/sink/sink_into_table.slt b/e2e_test/sink/sink_into_table.slt
@@ -6,6 +6,9 @@ SET RW_IMPLICIT_FLUSH TO true;
 statement ok
 create table t_simple (v1 int, v2 int);
 
+statement error unsupported sink type table
+create sink table_sink from t_simple with (connector = 'table');
+
 statement ok
 create table m_simple (v1 int primary key, v2 int);
 

diff --git a/e2e_test/streaming/over_window/main.slt b/e2e_test/streaming/over_window/main.slt
@@ -5,22 +5,20 @@ set rw_streaming_over_window_cache_policy = full;
 
 include ./generated/main.slt.part
 
-# TODO(rc): The following tests are temporarily commented out because of recovery test failure.
-
-#statement ok
-#set rw_streaming_over_window_cache_policy = recent;
-#
-#include ./generated/main.slt.part
-#
-#statement ok
-#set rw_streaming_over_window_cache_policy = recent_first_n;
-#
-#include ./generated/main.slt.part
-#
-#statement ok
-#set rw_streaming_over_window_cache_policy = recent_last_n;
-#
-#include ./generated/main.slt.part
+statement ok
+set rw_streaming_over_window_cache_policy = recent;
+
+include ./generated/main.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_first_n;
+
+include ./generated/main.slt.part
+
+statement ok
+set rw_streaming_over_window_cache_policy = recent_last_n;
+
+include ./generated/main.slt.part
 
 statement ok
 set rw_streaming_over_window_cache_policy = default;
diff --git a/grafana/risingwave-dev-dashboard.dashboard.py b/grafana/risingwave-dev-dashboard.dashboard.py
@@ -265,7 +265,13 @@ def section_compaction(outer_panels):
                     [
                         panels.target(
                             f"avg({metric('storage_compact_task_pending_num')}) by({COMPONENT_LABEL}, {NODE_LABEL})",
-                            "compactor_task_split_count - {{%s}} @ {{%s}}"
+                            "compactor_task_count - {{%s}} @ {{%s}}"
+                            % (COMPONENT_LABEL, NODE_LABEL),
+                        ),
+
+                        panels.target(
+                            f"avg({metric('storage_compact_task_pending_parallelism')}) by({COMPONENT_LABEL}, {NODE_LABEL})",
+                            "compactor_task_pending_parallelism - {{%s}} @ {{%s}}"
                             % (COMPONENT_LABEL, NODE_LABEL),
                         ),
                     ],

diff --git a/grafana/risingwave-dev-dashboard.json b/grafana/risingwave-dev-dashboard.json
diff --git a/grafana/risingwave-user-dashboard.json b/grafana/risingwave-user-dashboard.json
diff --git a/integration_tests/iceberg-sink/create_sink.sql b/integration_tests/iceberg-sink/create_sink.sql
@@ -1,13 +1,15 @@
 CREATE SINK bhv_iceberg_sink
 FROM
     bhv_mv WITH (
-    connector = 'iceberg_java',
+    connector = 'iceberg',
     type = 'upsert',
     primary_key = 'user_id, target_id, event_timestamp',
     warehouse.path = 's3://hummock001/iceberg-data',
     s3.endpoint = 'http://minio-0:9301',
     s3.access.key = 'hummockadmin',
     s3.secret.key = 'hummockadmin',
-    database.name='demo_db',
-    table.name='demo_table'
+    s3.region = 'us-east-1',
+    catalog.type = 'storage',
+    database.name='demo',
+    table.name='demo_db.demo_table'
 );
diff --git a/java/connector-node/assembly/assembly.xml b/java/connector-node/assembly/assembly.xml
@@ -42,11 +42,8 @@
                 <include>*:risingwave-sink-es-7</include>
                 <include>*:risingwave-sink-cassandra</include>
                 <include>*:risingwave-sink-jdbc</include>
-                <include>*:risingwave-sink-iceberg</include>
-                <include>*:risingwave-sink-deltalake</include>
+                <include>*:risingwave-sink-mock-flink-http-sink</include>
 
-                <!-- For S3-->
-                <include>*:s3-common</include>
             </includes>
             <useTransitiveDependencies>true</useTransitiveDependencies>
             <useTransitiveFiltering>true</useTransitiveFiltering>

diff --git a/java/connector-node/risingwave-connector-service/pom.xml b/java/connector-node/risingwave-connector-service/pom.xml
@@ -38,6 +38,11 @@
             <groupId>com.risingwave</groupId>
             <artifactId>connector-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
 
         <dependency>
             <groupId>com.google.code.gson</groupId>
@@ -77,16 +82,6 @@
             <artifactId>risingwave-sink-jdbc</artifactId>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-iceberg</artifactId>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.risingwave</groupId>
-            <artifactId>risingwave-sink-deltalake</artifactId>
-            <scope>provided</scope>
-        </dependency>
         <dependency>
             <groupId>com.risingwave</groupId>
             <artifactId>risingwave-sink-es-7</artifactId>
@@ -104,6 +99,7 @@
         <dependency>
             <groupId>com.risingwave</groupId>
             <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
+            <scope>provided</scope>
         </dependency>
     </dependencies>
 </project>
diff --git a/...r-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java b/...r-node/risingwave-connector-service/src/main/java/com/risingwave/connector/SinkUtils.java
@@ -39,10 +39,6 @@ public static SinkFactory getSinkFactory(String sinkName) {
                 return new FileSinkFactory();
             case "jdbc":
                 return new JDBCSinkFactory();
-            case "iceberg_java":
-                return new IcebergSinkFactory();
-            case "deltalake":
-                return new DeltaLakeSinkFactory();
             case "elasticsearch":
                 return new EsSinkFactory();
             case "cassandra":

diff --git a/java/connector-node/risingwave-sink-es-7/pom.xml b/java/connector-node/risingwave-sink-es-7/pom.xml
@@ -51,6 +51,10 @@
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+        </dependency>
     </dependencies>
 
 </project>