Add pulsar iceberg table reader

risingwavelabs · Oct 18, 2023 · 5d5e496 · 5d5e496
1 parent 52f4c63
commit 5d5e496
Show file tree

Hide file tree

Showing 8 changed files with 379 additions and 10 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -111,14 +111,15 @@ criterion = { version = "0.5", features = ["async_futures"] }
 tonic = { package = "madsim-tonic", version = "0.4.0" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.0" }
 prost = { version = "0.12" }
-icelake = { git = "https://github.com/icelake-io/icelake", rev = "16dab0e36ab337e58ee8002d828def2d212fa116" }
+icelake = { git = "https://github.com/icelake-io/icelake", rev = "186fde7663545d1d6a5856ce9fbbc541224eadfb" }
 arrow-array = "47"
 arrow-cast = "47"
 arrow-schema = "47"
 arrow-buffer = "47"
 arrow-flight = "47"
 arrow-select = "47"
 arrow-ord = "47"
+arrow-json = "47"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }
 tikv-jemallocator = { git = "https://github.com/risingwavelabs/jemallocator.git", features = [
   "profiling",

diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
@@ -22,6 +22,7 @@ apache-avro = { git = "https://github.com/risingwavelabs/avro", branch = "idx0de
     "xz",
 ] }
 arrow-array = { workspace = true }
+arrow-json = { workspace = true }
 arrow-schema = { workspace = true }
 async-nats = "0.32"
 async-trait = "0.1"

diff --git a/src/connector/src/aws_utils.rs b/src/connector/src/aws_utils.rs
@@ -24,12 +24,16 @@ use url::Url;
 
 use crate::aws_auth::AwsAuthProps;
 
+pub const REGION: &str = "region";
+pub const ACCESS_KEY: &str = "access_key";
+pub const SECRET_ACCESS: &str = "secret_access";
+
 pub const AWS_DEFAULT_CONFIG: [&str; 7] = [
-    "region",
+    REGION,
     "arn",
     "profile",
-    "access_key",
-    "secret_access",
+    ACCESS_KEY,
+    SECRET_ACCESS,
     "session_token",
     "endpoint_url",
 ];

diff --git a/src/connector/src/error.rs b/src/connector/src/error.rs
@@ -35,6 +35,9 @@ pub enum ConnectorError {
     #[error("MySQL error: {0}")]
     MySql(#[from] mysql_async::Error),
 
+    #[error("Pulsar error: {0}")]
+    Pulsar(anyhow::Error),
+
     #[error(transparent)]
     Internal(#[from] anyhow::Error),
 }

diff --git a/src/connector/src/source/base.rs b/src/connector/src/source/base.rs
@@ -505,6 +505,9 @@ pub struct SourceMessage {
     pub meta: SourceMeta,
 }
 
+#[derive(Debug, Clone)]
+pub struct SourceMessages(pub Vec<SourceMessage>);
+
 #[derive(Debug, Clone)]
 pub enum SourceMeta {
     Kafka(KafkaMeta),

diff --git a/src/connector/src/source/pulsar/mod.rs b/src/connector/src/source/pulsar/mod.rs
@@ -21,8 +21,8 @@ pub use enumerator::*;
 use serde::Deserialize;
 pub use split::*;
 
+use self::source::reader::PulsarSplitReader;
 use crate::common::PulsarCommon;
-use crate::source::pulsar::source::reader::PulsarSplitReader;
 use crate::source::SourceProperties;
 
 pub const PULSAR_CONNECTOR: &str = "pulsar";
@@ -45,4 +45,10 @@ pub struct PulsarProperties {
 
     #[serde(flatten)]
     pub common: PulsarCommon,
+
+    #[serde(rename = "iceberg.enabled", default)]
+    pub iceberg_loader_enabled: Option<String>,
+
+    #[serde(rename = "iceberg.bucket", default)]
+    pub iceberg_bucket: Option<String>,
 }