You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "EdgarModesto23 (via GitHub)" <gi...@apache.org> on 2025/09/23 00:41:34 UTC
[PR] feat(connector) Add Iceberg sink connector [iggy]
EdgarModesto23 opened a new pull request, #2191:
URL: https://github.com/apache/iggy/pull/2191
This PR adds and Iceberg sink connector with the following features:
- **Support for S3-compatible storage**
- **Support for REST catalogs**
- **Single destination table**
- **Multiple-table fan-out static routing**
- **Multiple-table fan-out dynamic routing**
I originally intended to include support for hms catalogs, but i had issues compiling the [current implementation of the hms iceberg catalog in rust](https://docs.rs/crate/iceberg-catalog-hms/latest)
It is also worth mentioning that although i included the code necessary to support glue, the iceberg glue catalog team [is still working towards appending data files](https://github.com/apache/iceberg-rust/issues/329)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371286716
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
+ fs,
+ gcs,
+ azdls,
+ oss,
+}
+
+impl fmt::Display for IcebergSinkStoreClass {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkStoreClass::s3 => "s3",
+ IcebergSinkStoreClass::fs => "fs",
+ IcebergSinkStoreClass::gcs => "gcs",
+ IcebergSinkStoreClass::oss => "oss",
+ IcebergSinkStoreClass::azdls => "azdls",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+impl fmt::Display for IcebergSinkTypes {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkTypes::rest => "rest",
+ IcebergSinkTypes::glue => "glue",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+sink_connector!(IcebergSink);
+
+#[derive(Debug)]
+pub struct IcebergSink {
+ id: u32,
+ config: IcebergSinkConfig,
+ props: HashMap<String, String>,
+ router: Option<Box<dyn Router>>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct IcebergSinkConfig {
+ pub tables: Vec<String>,
+ pub catalog_type: IcebergSinkTypes,
+ pub warehouse: String,
+ pub uri: String,
+ pub dynamic_routing: bool,
+ pub dynamic_route_field: String,
+ pub store_url: String,
+ pub store_access_key_id: String,
+ pub store_secret_access_key: String,
+ pub store_region: String,
+ pub store_class: IcebergSinkStoreClass,
+}
+
+fn slice_user_table(table: &str) -> Vec<String> {
+ table.split('.').map(|s| s.to_string()).collect()
+}
+
+impl IcebergSink {
+ #[inline(always)]
+ fn get_props_s3(&self) -> Result<HashMap<String, String>, Error> {
+ let mut props: HashMap<String, String> = HashMap::new();
+ props.insert("s3.region".to_string(), self.config.store_region.clone());
+ props.insert(
+ "s3.access-key-id".to_string(),
+ self.config.store_access_key_id.clone(),
+ );
+ props.insert(
+ "s3.secret-access-key".to_string(),
+ self.config.store_secret_access_key.clone(),
+ );
+ props.insert("s3.endpoint".to_string(), self.config.store_url.clone());
+ Ok(props)
+ }
+
+ pub fn new(id: u32, config: IcebergSinkConfig) -> Self {
+ let props = HashMap::new();
+ let router = None;
+
+ IcebergSink {
+ id,
+ config,
+ router,
+ props,
+ }
+ }
+
+ #[inline(always)]
+ fn get_rest_catalog(&self) -> RestCatalog {
+ let catalog_config = RestCatalogConfig::builder()
+ .uri(self.config.uri.clone())
+ .props(self.props.clone())
+ .warehouse(self.config.warehouse.clone())
+ .build();
+
+ RestCatalog::new(catalog_config)
+ }
+
+ //#[inline(always)]
Review Comment:
What was a reason to have this commented out? I think it'd best to either add the appropriate comment or remove this code to avoid any confusion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector): Add Iceberg sink connector [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on PR #2191:
URL: https://github.com/apache/iggy/pull/2191#issuecomment-3529228715
> Hey @EdgarModesto23 do you plan to finalize this PR?
Hey @spetz, this PR for the Iceberg Sink connector is ready.
As I mentioned in the original description, the Iceberg Rust team is still working on adding support for appending files on Glue. Once that’s available, I plan to add the necessary code in a separate PR. Other than that, this one should be good to go.
Regarding the Iceberg Source connector. My plan was to merge the Sink connector first, and then work on the Source one when I have time. However, if you prefer both connectors to be merged together, we can close this PR and reopen it once both are complete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector): Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2372148591
##########
core/connectors/sinks/iceberg_sink/iggy_iceberg_sink_config.toml:
##########
@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+[http_api] # Optional HTTP API configuration
+enabled = true
+address = "127.0.0.1:8081"
+# api_key = "secret" # Optional API key for authentication to be passed as `api-key` header
+
+[http_api.cors] # Optional CORS configuration for HTTP API
+enabled = false
+allowed_methods = ["GET", "POST", "PUT", "DELETE"]
+allowed_origins = ["*"]
+allowed_headers = ["content-type"]
+exposed_headers = [""]
+allow_credentials = false
+allow_private_network = false
+
+[http_api.tls] # Optional TLS configuration for HTTP API
+enabled = false
+cert_file = "core/certs/iggy_cert.pem"
+key_file = "core/certs/iggy_key.pem"
+
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+# token = "secret" # Personal Access Token (PAT) can be used instead of username and password
+
+[state]
+path = "local_state"
+
+[sinks.iceberg]
+enabled = true
+name = "Iceberg sink"
+path = "target/release/libiggy_connector_iceberg_sink"
+
+[[sinks.iceberg.streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "iceberg_sink_connector"
+
+# Local S3 example
+[sinks.iceberg.config]
+tables = ["nyc.users"]
+catalog_type = "rest"
+warehouse = "warehouse"
+uri = "http://localhost:8181"
+dynamic_routing = true
+dynamic_route_field = "db_table"
+store_url = "http://localhost:9000"
+store_access_key_id = "admin"
+store_secret_access_key = "password"
+store_region = "us-east-1"
+store_class = "s3"
+
+
+[sinks.iceberg.transforms.add_fields]
+enabled = true
+
+[[sinks.iceberg.transforms.add_fields.fields]]
+key = "db_table"
+value.static = "nyc.users"
+
+#
+# [sinks.stdout]
+# enabled = true
+# name = "Stdout sink"
+# path = "target/release/libiggy_connector_stdout_sink"
+#
+# [[sinks.stdout.streams]]
+# stream = "example_stream"
+# topics = ["example_topic"]
+# schema = "json"
+# batch_length = 100
+# poll_interval = "5ms"
+# consumer_group = "stdout_sink_connector"
+#
+# [sinks.stdout.config]
+# print_payload = false
+#
+# [sinks.stdout.transforms.add_fields]
+# enabled = true
+#
+# [[sinks.stdout.transforms.add_fields.fields]]
+# key = "message"
+# value.static = "hello"
+
+[sources.random]
Review Comment:
Maybe remove the whole sources section for example config?
##########
core/connectors/sinks/iceberg_sink/iggy_iceberg_sink_config.toml:
##########
@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+[http_api] # Optional HTTP API configuration
+enabled = true
+address = "127.0.0.1:8081"
+# api_key = "secret" # Optional API key for authentication to be passed as `api-key` header
Review Comment:
Keep this as `api_key = ""` (not commented out)
##########
core/connectors/sinks/iceberg_sink/iggy_iceberg_sink_config.toml:
##########
@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+[http_api] # Optional HTTP API configuration
+enabled = true
+address = "127.0.0.1:8081"
+# api_key = "secret" # Optional API key for authentication to be passed as `api-key` header
+
+[http_api.cors] # Optional CORS configuration for HTTP API
+enabled = false
+allowed_methods = ["GET", "POST", "PUT", "DELETE"]
+allowed_origins = ["*"]
+allowed_headers = ["content-type"]
+exposed_headers = [""]
+allow_credentials = false
+allow_private_network = false
+
+[http_api.tls] # Optional TLS configuration for HTTP API
+enabled = false
+cert_file = "core/certs/iggy_cert.pem"
+key_file = "core/certs/iggy_key.pem"
+
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+# token = "secret" # Personal Access Token (PAT) can be used instead of username and password
+
+[state]
+path = "local_state"
+
+[sinks.iceberg]
+enabled = true
+name = "Iceberg sink"
+path = "target/release/libiggy_connector_iceberg_sink"
+
+[[sinks.iceberg.streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "iceberg_sink_connector"
+
+# Local S3 example
+[sinks.iceberg.config]
+tables = ["nyc.users"]
+catalog_type = "rest"
+warehouse = "warehouse"
+uri = "http://localhost:8181"
+dynamic_routing = true
+dynamic_route_field = "db_table"
+store_url = "http://localhost:9000"
+store_access_key_id = "admin"
+store_secret_access_key = "password"
+store_region = "us-east-1"
+store_class = "s3"
+
+
+[sinks.iceberg.transforms.add_fields]
+enabled = true
+
+[[sinks.iceberg.transforms.add_fields.fields]]
+key = "db_table"
+value.static = "nyc.users"
+
+#
+# [sinks.stdout]
Review Comment:
Let's remove commented out stdout sink.
##########
core/connectors/sinks/iceberg_sink/src/sink.rs:
##########
@@ -0,0 +1,91 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{
+ IcebergSink,
+ catalog::init_catalog,
+ router::{dynamic_router::DynamicRouter, static_router::StaticRouter},
+};
+use async_trait::async_trait;
+use iceberg::Catalog;
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata};
+use tracing::{debug, error, info};
+
+#[async_trait]
+impl Sink for IcebergSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}",
Review Comment:
We could also log here the redacted values for access key ID and secret, e.g. just take the first few chars and add them to the log, it's especially helpful when you want to ensure, that the provided credentials are valid.
##########
core/connectors/sinks/iceberg_sink/src/router/mod.rs:
##########
@@ -0,0 +1,221 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::router::arrow_streamer::JsonArrowReader;
+use crate::slice_user_table;
+use arrow_json::ReaderBuilder;
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use std::sync::Arc;
+use tracing::{error, warn};
+use uuid::Uuid;
+
+mod arrow_streamer;
+pub mod dynamic_router;
+pub mod static_router;
+
+pub fn is_valid_namespaced_table(input: &str) -> bool {
+ let parts: Vec<&str> = input.split('.').collect();
+ parts.len() >= 2 && parts.iter().all(|part| !part.is_empty())
+}
+
+async fn table_exists(route_field_val: &str, catalog: &dyn Catalog) -> Option<Table> {
+ let sliced_table = slice_user_table(route_field_val);
+ let table_ident = TableIdent::from_strs(&sliced_table).ok()?;
+
+ catalog.load_table(&table_ident).await.ok()
+}
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().ok_or_else(|| {
Review Comment:
Maybe let's call this variable like `field_type` (for example), something a bit more meaningful than plain `t`.
##########
core/connectors/sinks/iceberg_sink/src/router/dynamic_router.rs:
##########
@@ -0,0 +1,169 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::router::{Router, is_valid_namespaced_table, table_exists, write_data};
+use async_trait::async_trait;
+use iceberg::Catalog;
+use iceberg::table::Table;
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload};
+use simd_json::base::ValueAsObject;
+use std::collections::HashMap;
+use tracing::{info, warn};
+
+#[derive(Debug)]
+pub struct DynamicRouter {
+ catalog: Box<dyn Catalog>,
+ route_field: String,
+}
+
+pub struct DynamicWriter {
+ pub tables_to_write: HashMap<String, Table>,
+ pub table_to_message: HashMap<String, Vec<ConsumedMessage>>,
+}
+
+impl DynamicWriter {
+ pub fn new() -> Self {
+ let tables_to_write = HashMap::new();
+ let table_to_message = HashMap::new();
+ Self {
+ tables_to_write,
+ table_to_message,
+ }
+ }
+
+ fn push_to_existing(
+ &mut self,
+ route_field_val: &str,
+ message: ConsumedMessage,
+ ) -> Option<ConsumedMessage> {
+ if let Some(message_vec) = self.table_to_message.get_mut(route_field_val) {
+ message_vec.push(message);
+ None
+ } else {
+ Some(message)
+ }
+ }
+
+ async fn load_table_if_exists(
+ &mut self,
+ route_field_val: &str,
+ catalog: &dyn Catalog,
+ ) -> Result<(), Error> {
+ let table = table_exists(route_field_val, catalog)
+ .await
+ .ok_or(Error::InvalidState)?;
+
+ self.tables_to_write
+ .insert(route_field_val.to_string(), table);
+
+ Ok(())
+ }
+}
+
+impl DynamicRouter {
+ pub fn new(catalog: Box<dyn Catalog>, route_field: String) -> Self {
+ Self {
+ catalog,
+ route_field,
+ }
+ }
+
+ fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String> {
+ match &message.payload {
+ Payload::Json(payload) => payload
+ .as_object()
+ .and_then(|obj| obj.get(&self.route_field))
+ .map(|val| val.to_string()),
+ _ => {
+ warn!("Unsupported format for iceberg connector");
+ None
+ }
+ }
+ }
+}
+
+#[async_trait]
+impl Router for DynamicRouter {
+ async fn route_data(
+ &self,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), crate::Error> {
+ let mut writer = DynamicWriter::new();
+ for message in messages {
+ let route_field_val = match self.extract_route_field(&message) {
+ Some(val) => val,
+ None => continue,
+ };
+
+ let message = match writer.push_to_existing(&route_field_val, message) {
+ Some(msg) => msg,
+ None => continue,
+ };
+
+ if !is_valid_namespaced_table(&route_field_val) {
+ warn!(
+ "Found invalid route field name on message: {}. Route fields should have at least 1 namespace separated by '.' character before the table",
+ route_field_val
Review Comment:
Let's use the string interpolation.
##########
core/connectors/sinks/iceberg_sink/src/router/arrow_streamer.rs:
##########
@@ -0,0 +1,79 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use simd_json::OwnedValue;
+use std::io::{self, BufRead, Cursor, Read};
+use std::slice::Iter;
+
+pub struct JsonArrowReader<'a> {
+ values: Iter<'a, &'a OwnedValue>,
+ cursor: Cursor<Vec<u8>>,
+}
+
+impl<'a> JsonArrowReader<'a> {
+ pub fn new(values: &'a [&OwnedValue]) -> Self {
+ Self {
+ values: values.iter(),
+ cursor: Cursor::new(Vec::new()),
+ }
+ }
+
+ fn load_next(&mut self) -> io::Result<bool> {
+ if let Some(val) = self.values.next() {
Review Comment:
You could also use `let Some(val) = .self.values.next() else { }` block, which might be a a bit more readable for extended logic :)
##########
core/connectors/sinks/iceberg_sink/iggy_iceberg_sink_config.toml:
##########
@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
Review Comment:
Please adjust this file for the new config format, where each sink/source has it's own dedicated configuration file with specified `id`, `type` and `version`. New, example configs are kept under `runtime/example_config/connectors`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2372146620
##########
core/connectors/sinks/iceberg_sink/example_config.toml:
##########
@@ -0,0 +1,130 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+[http_api] # Optional HTTP API configuration
+enabled = true
+address = "127.0.0.1:8081"
+# api_key = "secret" # Optional API key for authentication to be passed as `api-key` header
+
+[http_api.cors] # Optional CORS configuration for HTTP API
+enabled = false
+allowed_methods = ["GET", "POST", "PUT", "DELETE"]
+allowed_origins = ["*"]
+allowed_headers = ["content-type"]
+exposed_headers = [""]
+allow_credentials = false
+allow_private_network = false
+
+[http_api.tls] # Optional TLS configuration for HTTP API
+enabled = false
+cert_file = "core/certs/iggy_cert.pem"
+key_file = "core/certs/iggy_key.pem"
+
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+# token = "secret" # Personal Access Token (PAT) can be used instead of username and password
Review Comment:
Keep this as `token = ""` (not commented out)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector): add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #2191:
URL: https://github.com/apache/iggy/pull/2191
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2377308264
##########
Cargo.toml:
##########
@@ -100,77 +110,81 @@ flume = "0.11.1"
futures = "0.3.31"
futures-util = "0.3.31"
human-repr = "1.1.0"
-humantime = "2.3.0"
+humantime = "2.2.0"
+iceberg = "0.6.0"
+iceberg-catalog-glue = "0.6.0"
+iceberg-catalog-rest = "0.6.0"
iggy = { path = "core/sdk", version = "0.7.0" }
+
+# Path dependencies
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.7.0" }
iggy_common = { path = "core/common", version = "0.7.0" }
iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.1.0" }
integration = { path = "core/integration" }
keyring = { version = "3.6.3", features = ["sync-secret-service", "vendored"] }
lazy_static = "1.5.0"
-log = "0.4.28"
+log = "0.4.27"
+
+# Optional dependencies
mimalloc = "0.1"
mockall = "0.13.1"
nonzero_lit = "0.1.2"
once_cell = "1.21.3"
+parquet = "55.2.0"
passterm = "=2.0.1"
postcard = { version = "1.1.3", features = ["alloc"] }
predicates = "3.1.3"
-quinn = "0.11.9"
+quinn = "0.11.8"
rand = "0.9.2"
-regex = "1.11.2"
-reqwest = { version = "0.12.23", default-features = false, features = [
+regex = "1.11.1"
+reqwest = { version = "0.12.22", default-features = false, features = [
"json",
"rustls-tls",
] }
reqwest-middleware = { version = "0.4.2", features = ["json"] }
reqwest-retry = "0.7.0"
-rust-s3 = { version = "0.37.0", default-features = false, features = [
+rust-s3 = { version = "0.36.0-beta.2", default-features = false, features = [
"tokio-rustls-tls",
"tags",
] }
rustls = { version = "0.23.31", features = ["ring"] }
-serde = { version = "1.0.225", features = ["derive", "rc"] }
-serde_json = "1.0.145"
+serde = { version = "1.0.219", features = ["derive", "rc"] }
+serde_json = "1.0.142"
serde_with = { version = "3.14.0", features = ["base64", "macros"] }
serde_yml = "0.0.12"
serial_test = "3.2.0"
server = { path = "core/server" }
-simd-json = { version = "0.16.0", features = ["serde_impl"] }
+simd-json = { version = "0.15.1", features = ["serde_impl"] }
strum = { version = "0.27.2", features = ["derive"] }
strum_macros = "0.27.2"
sysinfo = "0.37.0"
-tempfile = "3.22.0"
+tempfile = "3.20.0"
test-case = "3.3.1"
-thiserror = "2.0.16"
+thiserror = "2.0.14"
tokio = { version = "1.47.1", features = ["full"] }
-tokio-rustls = "0.26.3"
+tokio-rustls = "0.26.2"
tokio-util = { version = "0.7.16", features = ["compat"] }
-toml = "0.9.6"
+toml = "0.9.5"
tower-http = { version = "0.6.6", features = [
"add-extension",
"cors",
"trace",
] }
tracing = "0.1.41"
tracing-appender = "0.2.3"
-tracing-subscriber = { version = "0.3.20", default-features = false, features = [
+tracing-subscriber = { version = "0.3.19", default-features = false, features = [
"fmt",
"env-filter",
"ansi",
] }
trait-variant = "0.1.2"
-twox-hash = { version = "2.1.2", features = ["xxhash32"] }
-uuid = { version = "1.18.1", features = [
+twox-hash = { version = "2.1.1", features = ["xxhash32"] }
Review Comment:
yeah I don't know what happened either haha, I'll fix this asap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector): Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #2191:
URL: https://github.com/apache/iggy/pull/2191#issuecomment-3529251612
@EdgarModesto23 thank you for the response, would you mind resolving these few comments I added back then during the initial review? And then, there are just the basic conflicts due to updated deps, and we could merge this PR :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371287992
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
+ fs,
+ gcs,
+ azdls,
+ oss,
+}
+
+impl fmt::Display for IcebergSinkStoreClass {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkStoreClass::s3 => "s3",
+ IcebergSinkStoreClass::fs => "fs",
+ IcebergSinkStoreClass::gcs => "gcs",
+ IcebergSinkStoreClass::oss => "oss",
+ IcebergSinkStoreClass::azdls => "azdls",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+impl fmt::Display for IcebergSinkTypes {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkTypes::rest => "rest",
+ IcebergSinkTypes::glue => "glue",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+sink_connector!(IcebergSink);
+
+#[derive(Debug)]
+pub struct IcebergSink {
+ id: u32,
+ config: IcebergSinkConfig,
+ props: HashMap<String, String>,
+ router: Option<Box<dyn Router>>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct IcebergSinkConfig {
+ pub tables: Vec<String>,
+ pub catalog_type: IcebergSinkTypes,
+ pub warehouse: String,
+ pub uri: String,
+ pub dynamic_routing: bool,
+ pub dynamic_route_field: String,
+ pub store_url: String,
+ pub store_access_key_id: String,
+ pub store_secret_access_key: String,
+ pub store_region: String,
+ pub store_class: IcebergSinkStoreClass,
+}
+
+fn slice_user_table(table: &str) -> Vec<String> {
+ table.split('.').map(|s| s.to_string()).collect()
+}
+
+impl IcebergSink {
+ #[inline(always)]
+ fn get_props_s3(&self) -> Result<HashMap<String, String>, Error> {
+ let mut props: HashMap<String, String> = HashMap::new();
+ props.insert("s3.region".to_string(), self.config.store_region.clone());
+ props.insert(
+ "s3.access-key-id".to_string(),
+ self.config.store_access_key_id.clone(),
+ );
+ props.insert(
+ "s3.secret-access-key".to_string(),
+ self.config.store_secret_access_key.clone(),
+ );
+ props.insert("s3.endpoint".to_string(), self.config.store_url.clone());
+ Ok(props)
+ }
+
+ pub fn new(id: u32, config: IcebergSinkConfig) -> Self {
+ let props = HashMap::new();
+ let router = None;
+
+ IcebergSink {
+ id,
+ config,
+ router,
+ props,
+ }
+ }
+
+ #[inline(always)]
+ fn get_rest_catalog(&self) -> RestCatalog {
+ let catalog_config = RestCatalogConfig::builder()
+ .uri(self.config.uri.clone())
+ .props(self.props.clone())
+ .warehouse(self.config.warehouse.clone())
+ .build();
+
+ RestCatalog::new(catalog_config)
+ }
+
+ //#[inline(always)]
+ //fn get_hms_catalog(&self) -> HmsCatalog {
+ // let config = HmsCatalogConfig::builder()
+ // .props(self.props)
+ // .warehouse(self.config.bucket_name.clone())
+ // .address(self.config.uri.clone())
+ // .thrift_transport(HmsThriftTransport::Buffered)
+ // .build();
+ //
+ // HmsCatalog::new(config)
+ //}
+
+ #[inline(always)]
+ async fn get_glue_catalog(&self) -> Result<GlueCatalog, Error> {
+ let config = GlueCatalogConfig::builder()
+ .props(self.props.clone())
+ .warehouse(self.config.warehouse.clone())
+ .build();
+
+ GlueCatalog::new(config).await.map_err(|err| {
+ error!("Failed to get glue catalog with error: {}. Make sure the catalog is correctly declared on the config file", err);
+ Error::InitError(err.to_string())
+ })
+ }
+}
+
+#[async_trait]
+impl Sink for IcebergSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ info!(
+ "Opened Iceberg sink connector with ID: {} for URL: {}",
+ self.id, self.config.uri
+ );
+
+ info!(
+ "Configuring Iceberg catalog with the following config:\n-region: {}\n-url: {}\n-store class: {}\n-catalog type: {}\n",
+ self.config.store_region,
+ self.config.store_url,
+ self.config.store_class,
+ self.config.catalog_type
+ );
+
+ // Insert adequate props for initializing file IO, else fail to open
+ self.props = match self.config.store_class {
+ IcebergSinkStoreClass::s3 => self.get_props_s3()?,
+ _ => {
+ error!(
+ "Store class {} is not supported yet",
+ self.config.store_class
+ );
+ return Err(Error::InvalidConfig);
+ }
+ };
+
+ let catalog: Box<dyn Catalog> = match self.config.catalog_type {
+ IcebergSinkTypes::rest => Box::new(self.get_rest_catalog()),
+ IcebergSinkTypes::glue => Box::new(self.get_glue_catalog().await?),
+ };
+
+ if self.config.dynamic_routing {
+ self.router = Some(Box::new(DynamicRouter::new(
+ catalog,
+ self.config.dynamic_route_field.clone(),
+ )))
+ } else {
+ self.router = Some(Box::new(
+ StaticRouter::new(catalog, &self.config.tables).await?,
+ ));
+ }
+
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ _topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ info!(
+ "Iceberg sink with ID: {} received: {} messages, format: {}",
Review Comment:
Maybe let's change both of these info logs, to the debug level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371702918
##########
Cargo.toml:
##########
@@ -100,77 +110,81 @@ flume = "0.11.1"
futures = "0.3.31"
futures-util = "0.3.31"
human-repr = "1.1.0"
-humantime = "2.3.0"
+humantime = "2.2.0"
+iceberg = "0.6.0"
+iceberg-catalog-glue = "0.6.0"
+iceberg-catalog-rest = "0.6.0"
iggy = { path = "core/sdk", version = "0.7.0" }
+
+# Path dependencies
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.7.0" }
iggy_common = { path = "core/common", version = "0.7.0" }
iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.1.0" }
integration = { path = "core/integration" }
keyring = { version = "3.6.3", features = ["sync-secret-service", "vendored"] }
lazy_static = "1.5.0"
-log = "0.4.28"
+log = "0.4.27"
+
+# Optional dependencies
mimalloc = "0.1"
mockall = "0.13.1"
nonzero_lit = "0.1.2"
once_cell = "1.21.3"
+parquet = "55.2.0"
passterm = "=2.0.1"
postcard = { version = "1.1.3", features = ["alloc"] }
predicates = "3.1.3"
-quinn = "0.11.9"
+quinn = "0.11.8"
rand = "0.9.2"
-regex = "1.11.2"
-reqwest = { version = "0.12.23", default-features = false, features = [
+regex = "1.11.1"
+reqwest = { version = "0.12.22", default-features = false, features = [
"json",
"rustls-tls",
] }
reqwest-middleware = { version = "0.4.2", features = ["json"] }
reqwest-retry = "0.7.0"
-rust-s3 = { version = "0.37.0", default-features = false, features = [
+rust-s3 = { version = "0.36.0-beta.2", default-features = false, features = [
"tokio-rustls-tls",
"tags",
] }
rustls = { version = "0.23.31", features = ["ring"] }
-serde = { version = "1.0.225", features = ["derive", "rc"] }
-serde_json = "1.0.145"
+serde = { version = "1.0.219", features = ["derive", "rc"] }
+serde_json = "1.0.142"
serde_with = { version = "3.14.0", features = ["base64", "macros"] }
serde_yml = "0.0.12"
serial_test = "3.2.0"
server = { path = "core/server" }
-simd-json = { version = "0.16.0", features = ["serde_impl"] }
+simd-json = { version = "0.15.1", features = ["serde_impl"] }
strum = { version = "0.27.2", features = ["derive"] }
strum_macros = "0.27.2"
sysinfo = "0.37.0"
-tempfile = "3.22.0"
+tempfile = "3.20.0"
test-case = "3.3.1"
-thiserror = "2.0.16"
+thiserror = "2.0.14"
tokio = { version = "1.47.1", features = ["full"] }
-tokio-rustls = "0.26.3"
+tokio-rustls = "0.26.2"
tokio-util = { version = "0.7.16", features = ["compat"] }
-toml = "0.9.6"
+toml = "0.9.5"
tower-http = { version = "0.6.6", features = [
"add-extension",
"cors",
"trace",
] }
tracing = "0.1.41"
tracing-appender = "0.2.3"
-tracing-subscriber = { version = "0.3.20", default-features = false, features = [
+tracing-subscriber = { version = "0.3.19", default-features = false, features = [
"fmt",
"env-filter",
"ansi",
] }
trait-variant = "0.1.2"
-twox-hash = { version = "2.1.2", features = ["xxhash32"] }
-uuid = { version = "1.18.1", features = [
+twox-hash = { version = "2.1.1", features = ["xxhash32"] }
Review Comment:
I think you messed something up during merge, please don't downgrade versions in Cargo.toml :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371864289
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
+ fs,
+ gcs,
+ azdls,
+ oss,
+}
+
+impl fmt::Display for IcebergSinkStoreClass {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkStoreClass::s3 => "s3",
Review Comment:
don't implement `Display` manually, please see `strum` usage, e.g. `core/common/src/types/cluster/status.rs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector): Add Iceberg sink connector [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #2191:
URL: https://github.com/apache/iggy/pull/2191#issuecomment-3525226858
Hey @EdgarModesto23 do you plan to finalize this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371896799
##########
core/connectors/sinks/iceberg_sink/src/router.rs:
##########
@@ -0,0 +1,438 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_json::ReaderBuilder;
+
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use simd_json::base::ValueAsObject;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::slice_user_table;
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().unwrap();
Review Comment:
please use expect, avoid unwraps when possible
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #2191:
URL: https://github.com/apache/iggy/pull/2191#issuecomment-3325138387
Also, fix CI :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "EdgarModesto23 (via GitHub)" <gi...@apache.org>.
EdgarModesto23 commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2377311397
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
+ fs,
+ gcs,
+ azdls,
+ oss,
+}
+
+impl fmt::Display for IcebergSinkStoreClass {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkStoreClass::s3 => "s3",
+ IcebergSinkStoreClass::fs => "fs",
+ IcebergSinkStoreClass::gcs => "gcs",
+ IcebergSinkStoreClass::oss => "oss",
+ IcebergSinkStoreClass::azdls => "azdls",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+impl fmt::Display for IcebergSinkTypes {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkTypes::rest => "rest",
+ IcebergSinkTypes::glue => "glue",
+ };
+ write!(f, "{}", s)
+ }
+}
+
+sink_connector!(IcebergSink);
+
+#[derive(Debug)]
+pub struct IcebergSink {
+ id: u32,
+ config: IcebergSinkConfig,
+ props: HashMap<String, String>,
+ router: Option<Box<dyn Router>>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct IcebergSinkConfig {
+ pub tables: Vec<String>,
+ pub catalog_type: IcebergSinkTypes,
+ pub warehouse: String,
+ pub uri: String,
+ pub dynamic_routing: bool,
+ pub dynamic_route_field: String,
+ pub store_url: String,
+ pub store_access_key_id: String,
+ pub store_secret_access_key: String,
+ pub store_region: String,
+ pub store_class: IcebergSinkStoreClass,
+}
+
+fn slice_user_table(table: &str) -> Vec<String> {
+ table.split('.').map(|s| s.to_string()).collect()
+}
+
+impl IcebergSink {
+ #[inline(always)]
+ fn get_props_s3(&self) -> Result<HashMap<String, String>, Error> {
+ let mut props: HashMap<String, String> = HashMap::new();
+ props.insert("s3.region".to_string(), self.config.store_region.clone());
+ props.insert(
+ "s3.access-key-id".to_string(),
+ self.config.store_access_key_id.clone(),
+ );
+ props.insert(
+ "s3.secret-access-key".to_string(),
+ self.config.store_secret_access_key.clone(),
+ );
+ props.insert("s3.endpoint".to_string(), self.config.store_url.clone());
+ Ok(props)
+ }
+
+ pub fn new(id: u32, config: IcebergSinkConfig) -> Self {
+ let props = HashMap::new();
+ let router = None;
+
+ IcebergSink {
+ id,
+ config,
+ router,
+ props,
+ }
+ }
+
+ #[inline(always)]
+ fn get_rest_catalog(&self) -> RestCatalog {
+ let catalog_config = RestCatalogConfig::builder()
+ .uri(self.config.uri.clone())
+ .props(self.props.clone())
+ .warehouse(self.config.warehouse.clone())
+ .build();
+
+ RestCatalog::new(catalog_config)
+ }
+
+ //#[inline(always)]
Review Comment:
This was the original code for having support for hms catalogs but later I commented it out because it's correct afaik but the current build for the i[ceberg-hms catalog is broken](https://docs.rs/crate/iceberg-catalog-hms/0.6.0), so I didn't really knew what to do with it. I'll remove it and add it later on when they fix it if that sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connector) Add Iceberg sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #2191:
URL: https://github.com/apache/iggy/pull/2191#discussion_r2371903226
##########
core/connectors/sinks/iceberg_sink/src/router.rs:
##########
@@ -0,0 +1,438 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_json::ReaderBuilder;
+
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use simd_json::base::ValueAsObject;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::slice_user_table;
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().unwrap();
+
+ let value = Some(Literal::Primitive(primitive_type_to_literal(t)?));
+
+ fields.push(value);
+ }
+ Ok(Some(Struct::from_iter(fields)))
+}
+
+async fn write_data<I, M>(
+ messages: I,
+ table: &Table,
+ catalog: &dyn Catalog,
+ messages_schema: Schema,
+) -> Result<(), Error>
+where
+ I: IntoIterator<Item = M>,
+ M: std::ops::Deref<Target = ConsumedMessage>,
+{
+ let location = DefaultLocationGenerator::new(table.metadata().clone()).map_err(|err| {
+ error!(
+ "Failed to get location on table: {}. Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidConfig
+ })?;
+
+ let file_name_gen = DefaultFileNameGenerator::new(
+ Uuid::new_v4().to_string(),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ WriterProperties::default(),
+ table.metadata().current_schema().clone(),
+ table.file_io().clone(),
+ location,
+ file_name_gen,
+ );
+
+ let data_file_writer_builder = DataFileWriterBuilder::new(
+ parquet_writer_builder,
+ get_partition_type_value(table.metadata().default_partition_type())?,
+ table.metadata().default_partition_spec_id(),
+ );
+
+ let mut writer = data_file_writer_builder.build().await.map_err(|err| {
+ error!("Error while constructing data file writer: {}", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ let json_messages = messages
+ .into_iter()
+ .filter_map(|record| match &record.payload {
+ Payload::Json(record) => simd_json::to_string(&record).ok(),
+ _ => {
+ warn!("Unsupported payload format: {}", messages_schema);
+ None
+ }
+ })
+ .collect::<Vec<_>>()
+ .join("\n");
+
+ if json_messages.is_empty() {
+ error!(
+ "Could not serialize payload, expected JSON format, got {} instead",
+ messages_schema
+ );
+ return Err(Error::InvalidPayloadType);
+ }
+
+ let cursor = Cursor::new(json_messages);
+
+ let reader = ReaderBuilder::new(Arc::new(
+ schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| {
+ error!(
+ "Error while mapping records to Iceberg table with uuid: {}. Error {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?,
+ ))
+ .build(cursor)
+ .map_err(|err| {
+ error!(
+ "Error while building Iceberg reader from message payload: {}",
+ err
+ );
+ Error::InitError(err.to_string())
+ })?;
+
+ for batch in reader {
+ let batch_data = batch.map_err(|err| {
+ error!("Error while getting record batch: {}", err);
+ Error::InvalidRecord
+ })?;
+ writer.write(batch_data).await.map_err(|err| {
+ error!("Error while writing record batch: {}", err);
+ Error::InvalidRecord
+ })?;
+ }
+
+ let data_files = writer.close().await.map_err(|err| {
+ error!("Error while writing data records to Parquet file: {}", err);
+ Error::InvalidRecord
+ })?;
+
+ let table_commit = Transaction::new(table);
+
+ let action = table_commit.fast_append().add_data_files(data_files);
+
+ let tx = action.apply(table_commit).map_err(|err| {
+ error!(
+ "Failed to apply transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?;
+
+ let _table = tx.commit(catalog).await.map_err(|err| {
+ error!(
+ "Failed to commit transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?;
+ Ok(())
+}
+
+#[derive(Debug)]
+pub(crate) struct DynamicRouter {
+ catalog: Box<dyn Catalog>,
+ route_field: String,
+}
+
+struct DynamicWriter {
+ pub tables_to_write: HashMap<String, Table>,
+ pub table_to_message: HashMap<String, Vec<Arc<ConsumedMessage>>>,
+}
+
+impl DynamicWriter {
+ pub fn new() -> Self {
+ let tables_to_write = HashMap::new();
+ let table_to_message = HashMap::new();
+ Self {
+ tables_to_write,
+ table_to_message,
+ }
+ }
+
+ fn push_to_existing(&mut self, route_field_val: &str, message: &Arc<ConsumedMessage>) -> bool {
+ if let Some(message_vec) = self.table_to_message.get_mut(route_field_val) {
+ message_vec.push(Arc::clone(message));
+ true
+ } else {
+ false
+ }
+ }
+
+ // This will:
+ // - Check if the table declared on the route field exists in the iceberg catalog.
+ // - If it does, it will try to load it to memory and map the name with the Table object so
+ // that we can dynamically send messages to it's correct destination.
+ async fn ensure_table_exists(
+ &mut self,
+ route_field_val: &str,
+ catalog: &dyn Catalog,
+ ) -> Result<bool, Error> {
+ let sliced_table = slice_user_table(route_field_val);
+ let table_ident = &TableIdent::from_strs(&sliced_table).map_err(|err| {
+ error!("Failed to load table from catalog: {}. ", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ if !catalog.table_exists(table_ident).await.map_err(|err| {
+ error!("Failed to load table from catalog: {}. ", err);
+ Error::InitError(err.to_string())
+ })? {
+ return Ok(false);
+ }
+
+ let table_ident = TableIdent::from_strs(&sliced_table).map_err(|err| {
+ error!("Failed to load table from catalog: {}.", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ let table = catalog.load_table(&table_ident).await.map_err(|err| {
+ error!("Failed to load table from catalog: {}", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ self.tables_to_write
+ .insert(route_field_val.to_string(), table);
+
+ Ok(true)
+ }
+}
+
+impl DynamicRouter {
+ pub fn new(catalog: Box<dyn Catalog>, route_field: String) -> Self {
+ Self {
+ catalog,
+ route_field,
+ }
+ }
+
+ fn extract_route_field(&self, message: &ConsumedMessage) -> Option<String> {
+ match &message.payload {
+ Payload::Json(payload) => payload
+ .as_object()
+ .and_then(|obj| obj.get(&self.route_field))
+ .map(|val| val.to_string()),
+ _ => {
+ warn!("Unsupported format for iceberg connector");
+ None
+ }
+ }
+ }
+}
+
+#[async_trait]
+impl Router for DynamicRouter {
+ async fn route_data(
+ &self,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), crate::Error> {
+ let mut writer = DynamicWriter::new();
+ for message in messages {
+ let message = Arc::new(message);
+ let route_field_val = match self.extract_route_field(&message) {
+ Some(val) => val,
+ None => continue,
Review Comment:
all messages without routing field will be skipped silently. is it intended behavior?
##########
core/connectors/sinks/iceberg_sink/example_config.toml:
##########
Review Comment:
consider changing the name of this file to e.g. `iggy_iceberg_sink_config.toml`
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
Review Comment:
please split this file into smaller modules, aim for around maximum 150 lines per file (if possible, this is not a hard requirement from me, just a suggestion)
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
Review Comment:
preferably, use "S3", "FS", etc., (all capital), remove allow non camel case types and and use serde directive: `#[serde(rename_all = "lowercase")]`
##########
core/connectors/sinks/iceberg_sink/src/router.rs:
##########
@@ -0,0 +1,438 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_json::ReaderBuilder;
+
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use simd_json::base::ValueAsObject;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::slice_user_table;
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().unwrap();
+
+ let value = Some(Literal::Primitive(primitive_type_to_literal(t)?));
+
+ fields.push(value);
+ }
+ Ok(Some(Struct::from_iter(fields)))
+}
+
+async fn write_data<I, M>(
+ messages: I,
+ table: &Table,
+ catalog: &dyn Catalog,
+ messages_schema: Schema,
+) -> Result<(), Error>
+where
+ I: IntoIterator<Item = M>,
+ M: std::ops::Deref<Target = ConsumedMessage>,
+{
+ let location = DefaultLocationGenerator::new(table.metadata().clone()).map_err(|err| {
+ error!(
+ "Failed to get location on table: {}. Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidConfig
+ })?;
+
+ let file_name_gen = DefaultFileNameGenerator::new(
+ Uuid::new_v4().to_string(),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ WriterProperties::default(),
+ table.metadata().current_schema().clone(),
+ table.file_io().clone(),
+ location,
+ file_name_gen,
+ );
+
+ let data_file_writer_builder = DataFileWriterBuilder::new(
+ parquet_writer_builder,
+ get_partition_type_value(table.metadata().default_partition_type())?,
+ table.metadata().default_partition_spec_id(),
+ );
+
+ let mut writer = data_file_writer_builder.build().await.map_err(|err| {
+ error!("Error while constructing data file writer: {}", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ let json_messages = messages
+ .into_iter()
+ .filter_map(|record| match &record.payload {
+ Payload::Json(record) => simd_json::to_string(&record).ok(),
+ _ => {
+ warn!("Unsupported payload format: {}", messages_schema);
+ None
+ }
+ })
+ .collect::<Vec<_>>()
+ .join("\n");
Review Comment:
i dont like this approach. look how many allocations you perform.
- each message gets deserialized to a new String (allocation per message)
- these strings are collected into a Vec (another allocation)
- then joined into one giant string (final massive allocation)
if i'm correct, for 10,000 messages of 1KB each, this allocates ~30MB+ unnecessarily
can you maybe stream directly to arrow json reader, without intermediate allocations?
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
Review Comment:
please remove empty lines between use items and let rustfmt format it, correct in other files too
##########
core/connectors/sinks/iceberg_sink/src/router.rs:
##########
@@ -0,0 +1,438 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_json::ReaderBuilder;
+
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use simd_json::base::ValueAsObject;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::slice_user_table;
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().unwrap();
Review Comment:
please use expect
##########
core/connectors/sinks/iceberg_sink/src/lib.rs:
##########
@@ -0,0 +1,249 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use core::fmt;
+use std::collections::HashMap;
+
+use async_trait::async_trait;
+
+use iceberg::Catalog;
+use iceberg_catalog_glue::{GlueCatalog, GlueCatalogConfig};
+use iceberg_catalog_rest::{RestCatalog, RestCatalogConfig};
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info};
+
+use crate::router::{DynamicRouter, Router, StaticRouter};
+
+mod router;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkTypes {
+ rest,
+ glue,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[allow(non_camel_case_types)]
+pub enum IcebergSinkStoreClass {
+ s3,
+ fs,
+ gcs,
+ azdls,
+ oss,
+}
+
+impl fmt::Display for IcebergSinkStoreClass {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let s = match self {
+ IcebergSinkStoreClass::s3 => "s3",
Review Comment:
don't implement `Display` manually, please use `strum` usage, e.g. `core/common/src/types/cluster/status.rs`
##########
core/connectors/sinks/iceberg_sink/src/router.rs:
##########
@@ -0,0 +1,438 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_json::ReaderBuilder;
+
+use async_trait::async_trait;
+use iceberg::TableIdent;
+use iceberg::arrow::schema_to_arrow_schema;
+use iceberg::spec::{Literal, PrimitiveLiteral, PrimitiveType, Struct, StructType};
+use iceberg::table::Table;
+use iceberg::transaction::{ApplyTransactionAction, Transaction};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
+use iceberg::writer::file_writer::ParquetWriterBuilder;
+use iceberg::writer::{IcebergWriter, IcebergWriterBuilder};
+use iceberg::{
+ Catalog,
+ writer::file_writer::location_generator::{DefaultFileNameGenerator, DefaultLocationGenerator},
+};
+use iggy_connector_sdk::{ConsumedMessage, Error, MessagesMetadata, Payload, Schema};
+use parquet::file::properties::WriterProperties;
+use simd_json::base::ValueAsObject;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::slice_user_table;
+
+pub fn primitive_type_to_literal(pt: &PrimitiveType) -> Result<PrimitiveLiteral, Error> {
+ match pt {
+ PrimitiveType::Boolean => Ok(PrimitiveLiteral::Boolean(false)),
+ PrimitiveType::Int => Ok(PrimitiveLiteral::Int(0)),
+ PrimitiveType::Long => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::Decimal { .. } => Ok(PrimitiveLiteral::Int128(0)),
+ PrimitiveType::Date => Ok(PrimitiveLiteral::Int(0)), // e.g. days since epoch
+ PrimitiveType::Time => Ok(PrimitiveLiteral::Long(0)), // microseconds since midnight
+ PrimitiveType::Timestamp => Ok(PrimitiveLiteral::Long(0)), // microseconds since epoch
+ PrimitiveType::Timestamptz => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestampNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::TimestamptzNs => Ok(PrimitiveLiteral::Long(0)),
+ PrimitiveType::String => Ok(PrimitiveLiteral::String(String::new())),
+ PrimitiveType::Uuid => Ok(PrimitiveLiteral::Binary(vec![0; 16])),
+ PrimitiveType::Fixed(len) => Ok(PrimitiveLiteral::Binary(vec![0; *len as usize])),
+ PrimitiveType::Binary => Ok(PrimitiveLiteral::Binary(Vec::new())),
+ _ => {
+ error!("Partition type not supported");
+ Err(Error::InvalidConfig)
+ }
+ }
+}
+
+fn get_partition_type_value(default_partition_type: &StructType) -> Result<Option<Struct>, Error> {
+ let mut fields: Vec<Option<Literal>> = Vec::new();
+
+ if default_partition_type.fields().is_empty() {
+ return Ok(None);
+ };
+
+ for field in default_partition_type.fields() {
+ let t = field.field_type.as_primitive_type().unwrap();
+
+ let value = Some(Literal::Primitive(primitive_type_to_literal(t)?));
+
+ fields.push(value);
+ }
+ Ok(Some(Struct::from_iter(fields)))
+}
+
+async fn write_data<I, M>(
+ messages: I,
+ table: &Table,
+ catalog: &dyn Catalog,
+ messages_schema: Schema,
+) -> Result<(), Error>
+where
+ I: IntoIterator<Item = M>,
+ M: std::ops::Deref<Target = ConsumedMessage>,
+{
+ let location = DefaultLocationGenerator::new(table.metadata().clone()).map_err(|err| {
+ error!(
+ "Failed to get location on table: {}. Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidConfig
+ })?;
+
+ let file_name_gen = DefaultFileNameGenerator::new(
+ Uuid::new_v4().to_string(),
+ None,
+ iceberg::spec::DataFileFormat::Parquet,
+ );
+
+ let parquet_writer_builder = ParquetWriterBuilder::new(
+ WriterProperties::default(),
+ table.metadata().current_schema().clone(),
+ table.file_io().clone(),
+ location,
+ file_name_gen,
+ );
+
+ let data_file_writer_builder = DataFileWriterBuilder::new(
+ parquet_writer_builder,
+ get_partition_type_value(table.metadata().default_partition_type())?,
+ table.metadata().default_partition_spec_id(),
+ );
+
+ let mut writer = data_file_writer_builder.build().await.map_err(|err| {
+ error!("Error while constructing data file writer: {}", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ let json_messages = messages
+ .into_iter()
+ .filter_map(|record| match &record.payload {
+ Payload::Json(record) => simd_json::to_string(&record).ok(),
+ _ => {
+ warn!("Unsupported payload format: {}", messages_schema);
+ None
+ }
+ })
+ .collect::<Vec<_>>()
+ .join("\n");
+
+ if json_messages.is_empty() {
+ error!(
+ "Could not serialize payload, expected JSON format, got {} instead",
+ messages_schema
+ );
+ return Err(Error::InvalidPayloadType);
+ }
+
+ let cursor = Cursor::new(json_messages);
+
+ let reader = ReaderBuilder::new(Arc::new(
+ schema_to_arrow_schema(&table.metadata().current_schema().clone()).map_err(|err| {
+ error!(
+ "Error while mapping records to Iceberg table with uuid: {}. Error {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?,
+ ))
+ .build(cursor)
+ .map_err(|err| {
+ error!(
+ "Error while building Iceberg reader from message payload: {}",
+ err
+ );
+ Error::InitError(err.to_string())
+ })?;
+
+ for batch in reader {
+ let batch_data = batch.map_err(|err| {
+ error!("Error while getting record batch: {}", err);
+ Error::InvalidRecord
+ })?;
+ writer.write(batch_data).await.map_err(|err| {
+ error!("Error while writing record batch: {}", err);
+ Error::InvalidRecord
+ })?;
+ }
+
+ let data_files = writer.close().await.map_err(|err| {
+ error!("Error while writing data records to Parquet file: {}", err);
+ Error::InvalidRecord
+ })?;
+
+ let table_commit = Transaction::new(table);
+
+ let action = table_commit.fast_append().add_data_files(data_files);
+
+ let tx = action.apply(table_commit).map_err(|err| {
+ error!(
+ "Failed to apply transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?;
+
+ let _table = tx.commit(catalog).await.map_err(|err| {
+ error!(
+ "Failed to commit transaction on table with UUID: {}, Error: {}",
+ table.metadata().uuid(),
+ err
+ );
+ Error::InvalidRecord
+ })?;
+ Ok(())
+}
+
+#[derive(Debug)]
+pub(crate) struct DynamicRouter {
+ catalog: Box<dyn Catalog>,
+ route_field: String,
+}
+
+struct DynamicWriter {
+ pub tables_to_write: HashMap<String, Table>,
+ pub table_to_message: HashMap<String, Vec<Arc<ConsumedMessage>>>,
+}
+
+impl DynamicWriter {
+ pub fn new() -> Self {
+ let tables_to_write = HashMap::new();
+ let table_to_message = HashMap::new();
+ Self {
+ tables_to_write,
+ table_to_message,
+ }
+ }
+
+ fn push_to_existing(&mut self, route_field_val: &str, message: &Arc<ConsumedMessage>) -> bool {
+ if let Some(message_vec) = self.table_to_message.get_mut(route_field_val) {
+ message_vec.push(Arc::clone(message));
+ true
+ } else {
+ false
+ }
+ }
+
+ // This will:
+ // - Check if the table declared on the route field exists in the iceberg catalog.
+ // - If it does, it will try to load it to memory and map the name with the Table object so
+ // that we can dynamically send messages to it's correct destination.
+ async fn ensure_table_exists(
+ &mut self,
+ route_field_val: &str,
+ catalog: &dyn Catalog,
+ ) -> Result<bool, Error> {
+ let sliced_table = slice_user_table(route_field_val);
+ let table_ident = &TableIdent::from_strs(&sliced_table).map_err(|err| {
+ error!("Failed to load table from catalog: {}. ", err);
+ Error::InitError(err.to_string())
+ })?;
+
+ if !catalog.table_exists(table_ident).await.map_err(|err| {
Review Comment:
this is inefficient, just create function that loads this table and returns `Option` because right now you are loading it twice for each unique table
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org