temp recover to main

ci/scripts/connector-node-integration-test.sh

@@ -115,43 +115,43 @@ for ((i=0; i<${#type[@]}; i++)); do
     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
+    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
+    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
+    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
 
 

e2e_test/iceberg/test_case/cdc/load.slt

@@ -38,8 +38,6 @@ CREATE SINK s1 AS select * from products WITH (
 statement ok
 flush;
 
-sleep 10s
-
 query I
 select count(*) from products;
 ----

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',
+    connector = 'deltalake_rust',
     type = 'append-only',
     force_append_only = 'true',
     location = 's3a://deltalake/deltalake-test',

e2e_test/sink/iceberg_sink.slt

@@ -6,17 +6,15 @@ 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',
+    connector = 'iceberg_java',
     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',
-    s3.region = 'us-east-1',
-    catalog.type = 'storage',
-    database.name='demo',
-    table.name='demo_db.demo_table'
+    database.name='demo_db',
+    table.name='demo_table'
 );
 
 statement ok

integration_tests/iceberg-sink/create_sink.sql

@@ -1,15 +1,13 @@
 CREATE SINK bhv_iceberg_sink
 FROM
     bhv_mv WITH (
-    connector = 'iceberg',
+    connector = 'iceberg_java',
     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',
-    s3.region = 'us-east-1',
-    catalog.type = 'storage',
-    database.name='demo',
-    table.name='demo_db.demo_table'
+    database.name='demo_db',
+    table.name='demo_table'
 );

java/connector-node/assembly/assembly.xml

@@ -42,8 +42,11 @@
                 <include>*:risingwave-sink-es-7</include>
                 <include>*:risingwave-sink-cassandra</include>
                 <include>*:risingwave-sink-jdbc</include>
-                <include>*:risingwave-sink-mock-flink-http-sink</include>
+                <include>*:risingwave-sink-iceberg</include>
+                <include>*:risingwave-sink-deltalake</include>
 
+                <!-- For S3-->
+                <include>*:s3-common</include>
             </includes>
             <useTransitiveDependencies>true</useTransitiveDependencies>
             <useTransitiveFiltering>true</useTransitiveFiltering>

java/connector-node/risingwave-connector-service/pom.xml

@@ -38,11 +38,6 @@
             <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>
@@ -82,6 +77,16 @@
             <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>
@@ -99,7 +104,6 @@
         <dependency>
             <groupId>com.risingwave</groupId>
             <artifactId>risingwave-sink-mock-flink-http-sink</artifactId>
-            <scope>provided</scope>
         </dependency>
     </dependencies>
 </project>

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

@@ -39,6 +39,10 @@ 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":

java/connector-node/risingwave-sink-es-7/pom.xml

@@ -51,10 +51,6 @@
             <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>

java/pom.xml

@@ -303,16 +303,6 @@
                 <artifactId>gson</artifactId>
                 <version>${gson.version}</version>
             </dependency>
-            <dependency>
-                <groupId>com.google.guava</groupId>
-                <artifactId>guava</artifactId>
-                <version>33.0.0-jre</version>
-            </dependency>
-            <dependency>
-                <groupId>org.apache.httpcomponents</groupId>
-                <artifactId>httpclient</artifactId>
-                <version>4.5.10</version>
-            </dependency>
             <dependency>
                 <groupId>io.prometheus</groupId>
                 <artifactId>simpleclient_httpserver</artifactId>

src/connector/src/sink/deltalake.rs

@@ -46,7 +46,7 @@ use super::{
 };
 use crate::sink::writer::SinkWriterExt;
 
-pub const DELTALAKE_SINK: &str = "deltalake";
+pub const DELTALAKE_SINK: &str = "deltalake_rust";
 pub const DEFAULT_REGION: &str = "us-east-1";
 
 #[derive(Deserialize, Serialize, Debug, Clone, WithOptions)]
@@ -534,7 +534,7 @@ mod test {
             .unwrap();
 
         let properties = hashmap! {
-            "connector".to_string() => "deltalake".to_string(),
+            "connector".to_string() => "deltalake_rust".to_string(),
             "force_append_only".to_string() => "true".to_string(),
             "type".to_string() => "append-only".to_string(),
             "location".to_string() => format!("file://{}", path),

src/connector/src/sink/iceberg/mod.rs

@@ -51,11 +51,22 @@ use super::{
 };
 use crate::deserialize_bool_from_string;
 use crate::sink::coordinate::CoordinatedSinkWriter;
+use crate::sink::remote::{CoordinatedRemoteSink, RemoteSinkTrait};
 use crate::sink::writer::{LogSinkerOf, SinkWriter, SinkWriterExt};
 use crate::sink::{Result, SinkCommitCoordinator, SinkParam};
 
 /// This iceberg sink is WIP. When it ready, we will change this name to "iceberg".
 pub const ICEBERG_SINK: &str = "iceberg";
+pub const REMOTE_ICEBERG_SINK: &str = "iceberg_java";
+
+#[derive(Debug)]
+pub struct RemoteIceberg;
+
+impl RemoteSinkTrait for RemoteIceberg {
+    const SINK_NAME: &'static str = REMOTE_ICEBERG_SINK;
+}
+
+pub type RemoteIcebergSink = CoordinatedRemoteSink<RemoteIceberg>;
 
 #[derive(Debug, Clone, Deserialize, WithOptions)]
 #[serde(deny_unknown_fields)]

src/connector/src/sink/mod.rs

@@ -79,13 +79,15 @@ macro_rules! for_all_sinks {
                 { ClickHouse, $crate::sink::clickhouse::ClickHouseSink },
                 { Iceberg, $crate::sink::iceberg::IcebergSink },
                 { Nats, $crate::sink::nats::NatsSink },
+                { RemoteIceberg, $crate::sink::iceberg::RemoteIcebergSink },
                 { Jdbc, $crate::sink::remote::JdbcSink },
+                { DeltaLake, $crate::sink::remote::DeltaLakeSink },
                 { ElasticSearch, $crate::sink::remote::ElasticSearchSink },
                 { Cassandra, $crate::sink::remote::CassandraSink },
                 { HttpJava, $crate::sink::remote::HttpJavaSink },
                 { Doris, $crate::sink::doris::DorisSink },
                 { Starrocks, $crate::sink::starrocks::StarrocksSink },
-                { DeltaLake, $crate::sink::deltalake::DeltaLakeSink },
+                { DeltaLakeRust, $crate::sink::deltalake::DeltaLakeSink },
                 { BigQuery, $crate::sink::big_query::BigQuerySink },
                 { Test, $crate::sink::test_sink::TestSink },
                 { Table, $crate::sink::table::TableSink }

src/frontend/src/optimizer/plan_node/stream_sink.rs

@@ -34,6 +34,7 @@ use risingwave_connector::sink::{
     SINK_TYPE_UPSERT, SINK_USER_FORCE_APPEND_ONLY_OPTION,
 };
 use risingwave_pb::stream_plan::stream_node::PbNodeBody;
+use tracing::info;
 
 use super::derive::{derive_columns, derive_pk};
 use super::generic::GenericPlanRef;
@@ -149,6 +150,13 @@ impl StreamSink {
             Distribution::Single => RequiredDist::single(),
             _ => {
                 match properties.get("connector") {
+                    Some(s) if s == "deltalake" => {
+                        // iceberg with multiple parallelism will fail easily with concurrent commit
+                        // on metadata
+                        // TODO: reset iceberg sink to have multiple parallelism
+                        info!("setting iceberg sink parallelism to singleton");
+                        RequiredDist::single()
+                    }
                     Some(s) if s == "jdbc" && sink_type == SinkType::Upsert => {
                         if sink_type == SinkType::Upsert && downstream_pk.is_empty() {
                             return Err(ErrorCode::SinkError(Box::new(Error::new(