feat(sink): introduce file sink [WIP]

src/connector/Cargo.toml

@@ -82,8 +82,9 @@ mysql_common = { version = "0.31", default-features = false, features = [
 ] }
 nexmark = { version = "0.2", features = ["serde"] }
 num-bigint = "0.4"
-opendal = "0.44"
+opendal = "0.44.2"
 parking_lot = "0.12"
+parquet = "50.0.0"
 paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
 prost = { version = "0.12", features = ["no-recursion-limit"] }

src/connector/src/sink/mod.rs

@@ -29,6 +29,7 @@ pub mod kinesis;
 pub mod log_store;
 pub mod mock_coordination_client;
 pub mod nats;
+pub mod opendal_sink;
 pub mod pulsar;
 pub mod redis;
 pub mod remote;
@@ -46,6 +47,8 @@ use ::deltalake::DeltaTableError;
 use ::redis::RedisError;
 use anyhow::anyhow;
 use async_trait::async_trait;
+use opendal::Error as OpendalError;
+use risingwave_common::array::ArrayError;
 use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::{ColumnDesc, Field, Schema};
 use risingwave_common::metrics::{
@@ -88,6 +91,8 @@ macro_rules! for_all_sinks {
                 { HttpJava, $crate::sink::remote::HttpJavaSink },
                 { Doris, $crate::sink::doris::DorisSink },
                 { Starrocks, $crate::sink::starrocks::StarrocksSink },
+                { S3, $crate::sink::opendal_sink::s3::S3Sink },
+                { Gcs, $crate::sink::opendal_sink::gcs::GcsSink },
                 { DeltaLake, $crate::sink::deltalake::DeltaLakeSink },
                 { BigQuery, $crate::sink::big_query::BigQuerySink },
                 { Test, $crate::sink::test_sink::TestSink },
@@ -525,6 +530,8 @@ pub enum SinkError {
     ),
     #[error("Starrocks error: {0}")]
     Starrocks(String),
+    #[error("Opendal error: {0}")]
+    Opendal(String),
     #[error("Pulsar error: {0}")]
     Pulsar(
         #[source]
@@ -557,6 +564,24 @@ impl From<icelake::Error> for SinkError {
     }
 }
 
+impl From<OpendalError> for SinkError {
+    fn from(error: OpendalError) -> Self {
+        SinkError::Opendal(error.to_string())
+    }
+}
+
+impl From<parquet::errors::ParquetError> for SinkError {
+    fn from(error: parquet::errors::ParquetError) -> Self {
+        SinkError::Opendal(error.to_string())
+    }
+}
+
+impl From<ArrayError> for SinkError {
+    fn from(error: ArrayError) -> Self {
+        SinkError::Opendal(error.to_string())
+    }
+}
+
 impl From<RpcError> for SinkError {
     fn from(value: RpcError) -> Self {
         SinkError::Remote(anyhow!(value))