You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "realonbebeto (via GitHub)" <gi...@apache.org> on 2026/07/12 04:44:02 UTC
[PR] feat(connectors): add Redshift sink connector [iggy]
realonbebeto opened a new pull request, #3654:
URL: https://github.com/apache/iggy/pull/3654
## Summary
Adds a Redshift sink connector for writing Apache Iggy messages into Amazon Redshift using S3-staged Parquet files and Redshift's COPY command.
The connector batches messages into Parquet files, uploads them to a configurable S3 bucket/prefix, and performs bulk COPY loads into the target Redshift table. It supports configurable batch sizing, connection pooling, optional metadata/checksum/origin timestamp fields, payload encoding modes, retry/backoff handling for failed loads, optional archival of staged Parquet files, and runtime metrics logging.
The implementation treats S3 purely as a staging layer for efficient bulk ingestion rather than a destination system. Staged files can either be deleted after a successful load or retained under an archive prefix for replay, auditing, or downstream processing.
It also integrates the connector throughout the project, including workspace membership, connector documentation, release and binary artifact pipelines, version bump automation, and Docker-based integration test infrastructure.
### Disclosures
1. `pgwire`, `sqlparser`, and `tokio-postgres `are introduced as workspace dependencies solely to support the RedshiftFixture used by the integration test suite.
2. The Redshift integration tests are inspired by and closely follow the existing PostgreSQL test setup to maintain consistency across sink connector implementations.
Closes #2540
## AI Usage
A blend of ChatGPT and Claude was used to help build, refactor the connector and associated test suite
--
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(connectors): add Redshift sink connector [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4976676379
@realonbebeto can you please fix the CI failures? After that we can review the code.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611503807
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
Review Comment:
**Meta (root cause of 1–4): false-green CI.** The hand-rolled pgwire mock (fixtures container.rs + mock_*) diverges from real Redshift on Decimal handling, COPY semantics, and PK enforcement. Prod code shape is even dictated by the mock ("Simple Query Handler doesn't support binds". Test tail wags prod dog. **Fix:** add a real / Redshift-compatible smoke gate, or document the mock's known divergences and label the suite "wire-shape only, not semantics."
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5329797941
@realonbebeto looks good. can you regenerate `Cargo.lock` - i think we can 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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611504951
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
+ // uses a Simple Query Handler that doesn't support prepared statements with binds.
+ let credentials = format!(
+ "CREDENTIALS 'ACCESS_KEY_ID={};SECRET_ACCESS_KEY={}'",
+ self.config.aws_access_key_id.expose_secret(),
+ self.config.aws_secret_access_key.expose_secret()
+ );
+
+ format!(
Review Comment:
lib.rs:377 vs :332 — COPY uses raw unquoted target_table; CREATE quotes it. Mixed-case/special name → CREATE "MyTable" but COPY mytable → wrong/missing table; quote char → injection. **Fix**: reuse quote_identifier in build_copy_sql.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #3654:
URL: https://github.com/apache/iggy/pull/3654
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5403028604
/ready
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5406633270
@realonbebeto looks like test is failing.
Summary [ 479.458s] 1581 tests run: 1580 passed (7 slow), 1 failed, 3170 skipped
TRY 4 FAIL [ 0.005s] ( 869/1581) iggy_connector_redshift_sink tests::given_payload_format_should_return_correct_sql_type
error: test run failed
https://github.com/apache/iggy/actions/runs/32817147871/job/97707918535?pr=3654
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4996897815
@slbotbm @krishvishal prek passed by allowing the commit. Thanks for the hint.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611507429
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
Review Comment:
lib.rs:294-296 — COPY failure ?-returns before delete → staged parquet orphaned forever (no runtime retry consumes it). **Fix**: best-effort delete on COPY failure.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611506117
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
Review Comment:
lib.rs:294-296,231 — non-atomic upload→copy→archive: post-COPY archive/delete failure returns Err → whole batch counted insertion_errors though rows loaded; staged file leaks; archive-then-delete-fail double-stores. **Fix:** separate commit from cleanup; cleanup failure = warning, not batch error.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4989491660
@realonbebeto looks like your lockfile is stale, and that is why a lot of the tests are failing
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4992963337
@slbotbm I made a cargo update on the PR. I think this resolves the staleness.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611523454
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
+ // uses a Simple Query Handler that doesn't support prepared statements with binds.
+ let credentials = format!(
+ "CREDENTIALS 'ACCESS_KEY_ID={};SECRET_ACCESS_KEY={}'",
+ self.config.aws_access_key_id.expose_secret(),
+ self.config.aws_secret_access_key.expose_secret()
+ );
+
+ format!(
+ "COPY {} FROM '{}' {} FORMAT AS PARQUET REGION '{}';",
+ self.config.target_table, s3_path, credentials, self.config.aws_region
+ )
+ }
+
+ async fn upload_parquet(&self, content: &[u8]) -> Result<String, Error> {
+ let file_id = Uuid::now_v7();
+ let key = build_s3_key(&self.config.s3_prefix, &format!("{file_id}.parquet"));
+ let bucket = self.get_bucket()?;
+
+ tracing::debug!(key = %key, bytes = content.len(), "uploading parquet to S3");
+
+ let response = bucket.put_object(&key, content).await.map_err(|e| {
+ tracing::error!("Failed to upload to S3 key '{key}': {e}");
+ Error::Storage(format!("S3 upload failed: {e}"))
+ })?;
+
+ ensure_s3_status(response.status_code(), 200, "S3 upload")?;
+
+ let path = format!("s3://{}{}", bucket.name(), key);
+ tracing::info!(path = %path, bytes = content.len(), "uploaded parquet to S3");
+
+ Ok(path)
+ }
+
+ async fn archive_parquet(&self, key: &str) -> Result<(), Error> {
+ let old_key = key
+ .strip_prefix(&format!("s3://{}/", self.config.s3_bucket))
+ .unwrap_or(key);
+
+ if !self.get_archive() {
+ self.delete_object(old_key).await?;
+ tracing::info!(key = old_key, "deleted parquet file (archiving disabled)");
+ return Ok(());
+ }
+
+ let bucket = self.get_bucket()?;
+ let prefix = self.config.s3_prefix.trim_matches('/');
+ let archived_key = old_key.replacen(prefix, DEFAULT_ARCHIVE_PREFIX.trim_matches('/'), 1);
+
+ tracing::debug!(from = old_key, to = %archived_key, "archiving parquet file");
+
+ let status_code = bucket
+ .copy_object_internal(old_key, &archived_key)
+ .await
+ .map_err(|e| {
+ tracing::error!(key = old_key, error = %e, "failed to copy object for archiving");
+ Error::Storage(format!("S3 archiving failed: {e}"))
+ })?;
+
+ ensure_s3_status(status_code, 200, "S3 archive copy")?;
+
+ self.delete_object(old_key).await?;
+ tracing::info!(archived_to = %archived_key, "archived parquet file");
+
+ Ok(())
+ }
+
+ async fn delete_object(&self, key: &str) -> Result<(), Error> {
+ let bucket = self.get_bucket()?;
+
+ let response = bucket.delete_object(key).await.map_err(|e| {
+ tracing::error!(key, error = %e, "failed to delete S3 object");
+ Error::Storage(format!("S3 deleting failed: {e}"))
+ })?;
+ ensure_s3_status(response.status_code(), 204, "S3 object deletion")?;
+
+ tracing::debug!(key, "deleted S3 object");
+ Ok(())
+ }
+
+ fn get_pool(&self) -> Result<&Pool<Postgres>, Error> {
+ self.pool
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Database not connected".to_string()))
+ }
+
+ fn get_bucket(&self) -> Result<&Bucket, Error> {
+ let r = self
+ .bucket
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Database not connected".to_string()))?;
+
+ Ok(r)
+ }
+
+ fn payload_format(&self) -> PayloadFormat {
+ PayloadFormat::from_config(self.config.payload_format.as_deref())
+ }
+
+ fn get_max_retries(&self) -> u32 {
+ self.config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES)
+ }
+
+ fn get_retry_delay(&self) -> Duration {
+ self.config
+ .retry_delay
+ .as_deref()
+ .unwrap_or(DEFAULT_RETRY_DELAY)
+ .parse::<HumanDuration>()
+ .map(Into::into)
+ .unwrap_or_else(|_| Duration::from_secs(1))
+ }
+
+ fn get_archive(&self) -> bool {
+ self.config.archive.unwrap_or(false)
+ }
+}
+
+#[derive(Debug, Default)]
+struct State {
+ messages_processed: u64,
+ batches_loaded: u64,
+ insertion_errors: u64,
+}
+
+/// Generic retry helper with linear backoff, used for transient warehouse errors.
+async fn retry_with_backoff<F, Fut, T>(
+ operation: &str,
+ max_retries: u32,
+ base_delay: Duration,
+ is_transient: impl Fn(&sqlx::Error) -> bool,
+ mut op: F,
+) -> Result<T, Error>
+where
+ F: FnMut() -> Fut,
+ Fut: Future<Output = Result<T, sqlx::Error>>,
+{
+ let mut attempts = 0u32;
+
+ loop {
+ match op().await {
+ Ok(value) => return Ok(value),
+ Err(e) => {
+ attempts += 1;
+ let transient = is_transient(&e);
+
+ if !transient || attempts >= max_retries {
+ tracing::error!(operation, attempts, error = %e, "operation failed permanently");
+ return Err(Error::CannotStoreData(format!(
+ "{operation} failed after {attempts} attempts: {e}"
+ )));
+ }
+
+ tracing::warn!(operation, attempts, max_retries, error = %e, "transient error, retrying");
+ tokio::time::sleep(base_delay * attempts).await;
+ }
+ }
+ }
+}
+
+fn encode_parquet(batch: &RecordBatch) -> Result<Vec<u8>, Error> {
+ let mut content = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut content, batch.schema(), None).map_err(|e| {
+ tracing::error!(error = %e, "failed to create parquet writer");
+ Error::WriteFailure(format!("Failed to create parquet writer: {e}"))
+ })?;
+
+ writer.write(batch).map_err(|e| {
+ tracing::error!(error = %e, "failed to write parquet batch");
+ Error::WriteFailure(format!("Failed to write parquet: {e}"))
+ })?;
+
+ writer.close().map_err(|e| {
+ tracing::error!(error = %e, "failed to close parquet writer");
+ Error::WriteFailure(format!("Failed to close writer: {e}"))
+ })?;
+
+ Ok(content)
+}
+
+fn build_s3_key(prefix: &str, filename: &str) -> String {
+ if prefix.is_empty() {
+ format!("/{filename}")
+ } else {
+ format!("/{}/{filename}", prefix.trim_end_matches('/'))
+ }
+}
+
+fn ensure_s3_status<T: PartialEq + std::fmt::Display>(
+ status: T,
+ expected: T,
+ context: &str,
+) -> Result<(), Error> {
+ if status != expected {
+ tracing::error!(context, %status, "unexpected S3 response status");
+ return Err(Error::Storage(format!(
+ "{context} failed with status {status}"
+ )));
+ }
+ Ok(())
+}
+
+fn create_record_batch(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ include_metadata: bool,
+ include_checksum: bool,
+ include_origin_timestamp: bool,
+ payload_format: PayloadFormat,
+) -> Result<RecordBatch, Error> {
+ let mut fields = vec![Field::new("id", DataType::Decimal256(39, 0), false)];
+ let mut columns: Vec<ArrayRef> = vec![id_column(messages)?];
+
+ if include_metadata {
+ let (mut metadata_fields, mut metadata_columns) =
+ metadata_columns(topic_metadata, messages_metadata, messages);
+ fields.append(&mut metadata_fields);
+ columns.append(&mut metadata_columns);
+ }
+
+ if include_checksum {
+ fields.push(Field::new("iggy_checksum", DataType::Utf8, false));
+ columns.push(checksum_column(messages));
+ }
+
+ if include_origin_timestamp {
+ fields.push(Field::new(
+ "iggy_origin_timestamp",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ false,
+ ));
+ columns.push(origin_timestamp_column(messages));
+ }
+
+ fields.push(Field::new("payload", payload_format.arrow_type(), false));
+ columns.push(payload_column(messages, payload_format)?);
+
+ let schema = Arc::new(Schema::new(fields));
+ let batch =
+ RecordBatch::try_new(schema, columns).map_err(|e| Error::CannotStoreData(e.to_string()))?;
+
+ tracing::debug!(
+ rows = batch.num_rows(),
+ columns = batch.num_columns(),
+ "built record batch"
+ );
+
+ Ok(batch)
+}
+
+fn id_column(messages: &[ConsumedMessage]) -> Result<ArrayRef, Error> {
+ let ids = Decimal256Array::from_iter_values(
+ messages
+ .iter()
+ .map(|v| arrow::datatypes::i256::from_parts(v.id, 0)),
+ )
+ .with_precision_and_scale(39, 0)
+ .map_err(|e| Error::CannotStoreData(e.to_string()))?;
+
+ Ok(Arc::new(ids))
+}
+
+fn metadata_columns(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+) -> (Vec<Field>, Vec<ArrayRef>) {
+ let fields = vec![
+ Field::new("iggy_offset", DataType::Int64, false),
+ Field::new(
+ "iggy_timestamp",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ false,
+ ),
+ Field::new("iggy_stream", DataType::Utf8, false),
+ Field::new("iggy_topic", DataType::Utf8, false),
+ Field::new("iggy_partition_id", DataType::Int32, false),
+ ];
+
+ let columns: Vec<ArrayRef> = vec![
+ Arc::new(Int64Array::from_iter_values(
+ messages.iter().map(|v| v.offset as i64),
+ )),
+ Arc::new(TimestampMicrosecondArray::from_iter_values(
+ messages.iter().map(|v| v.timestamp as i64),
+ )),
+ Arc::new(StringArray::from_iter_values(
+ (0..messages.len()).map(|_| topic_metadata.stream.clone()),
+ )),
+ Arc::new(StringArray::from_iter_values(
+ (0..messages.len()).map(|_| topic_metadata.topic.clone()),
+ )),
+ Arc::new(Int32Array::from_iter_values(
+ (0..messages.len()).map(|_| messages_metadata.partition_id as i32),
+ )),
+ ];
+
+ (fields, columns)
+}
+
+fn checksum_column(messages: &[ConsumedMessage]) -> ArrayRef {
+ Arc::new(StringArray::from_iter_values(
+ messages.iter().map(|v| v.checksum.to_string()),
+ ))
+}
+
+fn origin_timestamp_column(messages: &[ConsumedMessage]) -> ArrayRef {
+ Arc::new(TimestampMicrosecondArray::from_iter_values(
+ messages.iter().map(|v| v.origin_timestamp as i64),
+ ))
+}
+
+fn payload_column(messages: &[ConsumedMessage], format: PayloadFormat) -> Result<ArrayRef, Error> {
+ match format {
+ PayloadFormat::Varbyte => {
Review Comment:
**Perf hot path:** lib.rs:685 v.payload.clone().try_to_bytes() double-materializes per message; lib.rs:654-659 clones stream/topic String N× per batch. **Fix:** drop the clone (try_to_bytes on borrow) and use .as_str() or a DictionaryArray.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5262121042
Schema difference queries display the same types and the queries run against Redshift Serverless.
<img width="791" height="557" alt="redshift-target" src="https://github.com/user-attachments/assets/6aa8a4c9-41b2-42c4-90c6-adfecf6b3418" />
<img width="776" height="519" alt="redshift" src="https://github.com/user-attachments/assets/3fc31eb1-bd9d-4f9e-ba5d-0ebb0020a645" />
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3850365749
##########
core/integration/tests/connectors/fixtures/redshift/redshift_mock/copy.rs:
##########
@@ -0,0 +1,146 @@
+// 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 sqlparser::{
+ ast::ObjectName,
+ keywords::Keyword,
+ parser::{Parser, ParserError},
+ tokenizer::Token,
+};
+
+use crate::connectors::fixtures::redshift::redshift_mock::{
+ expect_word, parse_number_literal, parse_string_literal,
Review Comment:
same for `create.rs`. neither file is declared in `mod.rs`, so they never compile - `expect_word`, `parse_string_literal` and `parse_number_literal` don't exist anywhere in the tree. adding the `mod` lines back fails with E0432. delete both files.
##########
core/connectors/sinks/redshift_sink/README.md:
##########
@@ -0,0 +1,217 @@
+# Redshift Sink Connector
+
+Writes Apache Iggy stream messages into Amazon Redshift via S3-staged Parquet
+files and a `COPY` load.
+
+Each connector batch is serialized to a Parquet file and uploaded to the
+configured S3 bucket/prefix, then loaded into the target Redshift table with a
+`COPY` statement. This makes S3 a staging area rather than a destination in
+its own right — Redshift is the system of record for the data.
+
+Persistent load failures are at-most-once from the runtime's perspective:
+messages may already be committed in Iggy before this connector exhausts its
+write attempts, so failed loads are logged but not redelivered.
+
+## Configuration
+
+```toml
+type = "sink"
+key = "redshift"
+enabled = true
+version = 0
+name = "Redshift sink"
+path = "../../target/release/libiggy_connector_redshift_sink"
+verbose = false
+
+[[streams]]
+stream = "user_events"
+topics = ["users", "orders"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "redshift_sink"
+
+[plugin_config]
+connection_string = "postgresql://user:pass@localhost:5439/database"
+target_table = "iggy_messages"
+batch_size = 100
+max_connections = 10
+include_metadata = true
+include_checksum = true
+include_origin_timestamp = true
+payload_format = "varbyte"
+aws_access_key_id = "admin"
+aws_secret_access_key = "password"
+aws_iam_role ="aws_iam_role=arn:aws:iam::0123456789012:role/iggyRole"
Review Comment:
`build_copy_sql` already prepends `aws_iam_role=`, so this example ends up as `CREDENTIALS 'aws_iam_role=aws_iam_role=arn:...'` and the COPY fails. use the bare arn like `config.toml` does.
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1493 @@
+// 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.
+
+mod config;
+
+use std::{collections::BTreeMap, str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{ArrayRef, BinaryBuilder, Int64Array, RecordBatch, StringArray, StringBuilder},
+ datatypes::{DataType, Field, Schema},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::{
+ arrow::ArrowWriter,
+ basic::{Compression, ZstdLevel},
+ file::properties::WriterProperties,
+};
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, Row, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ // Validating config
+ self.config.validate()?;
+
+ self.connect().await?;
+ // Ensuring tables exist
+ self.ensure_tables_exist().await?;
+ // Checking for schema drift
+ self.ensure_schema_match().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ self.config
+ .aws_access_key_id
+ .as_ref()
+ .map(|v| v.expose_secret()),
+ self.config
+ .aws_secret_access_key
+ .as_ref()
+ .map(|v| v.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_tables_exist(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let target_table = quote_identifier(&self.config.target_table)?;
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+
+ let payload_type = self.payload_format().sql_type();
+
+ let target_query = self.build_create_table_sql(&target_table)?;
+
+ let staging_query = self.build_create_table_sql(&staging_table)?;
+
+ tracing::debug!("ensuring staging and target tables exist");
+
+ sqlx::query(AssertSqlSafe(staging_query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{staging_table}': {e}"))
+ })?;
+
+ tracing::debug!("Staging table created");
+
+ sqlx::query(AssertSqlSafe(target_query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{target_table}': {e}"))
+ })?;
+
+ tracing::info!(
+ staging_table = staging_table,
+ target_table = target_table,
+ payload_type,
+ "staging and target tables ready"
+ );
+
+ Ok(())
+ }
+
+ // This method ensures that the target table schema matches the expected schema.
+ // it also verifies there is a created_at column
+ async fn ensure_schema_match(&self) -> Result<(), Error> {
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let target_table = quote_identifier(&self.config.target_table)?;
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+ let payload_type = self.payload_format().sql_type();
+ let pool = self.get_pool()?;
+
+ let mut expected_cols: BTreeMap<&str, &str> = BTreeMap::new();
+ expected_cols.insert("id", "VARCHAR");
+ if include_metadata {
+ expected_cols.insert("iggy_offset", "VARCHAR");
+ expected_cols.insert("iggy_timestamp", "VARCHAR");
+ expected_cols.insert("iggy_stream", "VARCHAR");
+ expected_cols.insert("iggy_topic", "VARCHAR");
+ expected_cols.insert("iggy_partition_id", "BIGINT");
+ }
+ if include_checksum {
+ expected_cols.insert("iggy_checksum", "VARCHAR");
+ }
+ if include_origin_timestamp {
+ expected_cols.insert("iggy_origin_timestamp", "VARCHAR");
+ }
+ expected_cols.insert("payload", Self::normalize_type(payload_type));
+ expected_cols.insert("created_at", "VARCHAR");
+
+ let target_cols = Self::load_columns(pool, &target_table).await?;
+ let staging_cols = Self::load_columns(pool, &staging_table).await?;
+
+ let mut mismatches = Self::diff_schema(&target_table, &target_cols, &expected_cols);
+ mismatches.extend(Self::diff_schema(
+ &staging_table,
+ &staging_cols,
+ &expected_cols,
+ ));
+
+ tracing::info!("Mismatches: {:?}", mismatches);
+
+ if !mismatches.is_empty() {
+ return Err(Error::InitError(format!(
+ "Schema mismatch detected:\n{}",
+ mismatches.join("\n")
+ )));
+ }
+
+ Ok(())
+ }
+
+ fn diff_schema(
+ table_name: &str,
+ actual_cols: &BTreeMap<String, String>,
+ expected_cols: &BTreeMap<&str, &str>,
+ ) -> Vec<String> {
+ let mut errors = Vec::new();
+
+ for (col_name, expected_type) in expected_cols {
+ match actual_cols.get(*col_name) {
+ None => errors.push(format!(
+ "{table_name}: missing column '{col_name}' (expected {expected_type})"
+ )),
+ Some(actual_type) if !Self::type_matches(actual_type, expected_type) => errors.push(format!(
+ "{table_name}: column '{col_name}' type mismatch — expected {expected_type}, found {actual_type}"
+ )),
+ _ => {}
+ }
+ }
+ errors
+ }
+
+ fn type_matches(actual: &str, expected: &str) -> bool {
+ Self::normalize_type(actual) == Self::normalize_type(expected)
+ }
+
+ async fn load_columns(
+ pool: &sqlx::PgPool,
+ table: &str,
+ ) -> Result<BTreeMap<String, String>, Error> {
+ let query = format!(
+ "SELECT \"column\", type FROM pg_table_def WHERE tablename = '{}'",
+ table.replace('"', "")
+ );
+
+ let rows = sqlx::query(AssertSqlSafe(query))
+ .fetch_all(pool)
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to read schema for '{table}': {e}")))?;
+
+ if rows.is_empty() {
+ return Err(Error::InitError(format!(
+ "Table '{table}' was not found or has no visible columns"
+ )));
+ }
+
+ rows.into_iter()
+ .map(|row| {
+ Ok((
+ row.try_get::<String, _>("column")
+ .map_err(|e| Error::InitError(e.to_string()))?,
+ Self::normalize_type(
+ &row.try_get::<String, _>("type")
+ .map_err(|e| Error::InitError(e.to_string()))?,
+ )
+ .to_string(),
+ ))
+ })
+ .collect()
+ }
+
+ fn normalize_type(t: &str) -> &'static str {
+ let base = t.split('(').next().unwrap_or(t).trim();
+
+ match base.to_ascii_uppercase().as_str() {
+ "INTEGER" | "INT" | "INT4" => "INTEGER",
+ // e.g. "bigint"
+ "INT8" | "BIGINT" => "BIGINT",
+ // e.g "character varying(40)", "character varying(20)", "character varying(256)"
+ "TEXT" | "VARCHAR" | "CHARACTER VARYING" => "VARCHAR",
+ // Having bytea because of the Postgres Test
+ // e.g. "binary varying(64000)"
+ "BYTEA" | "VARBYTE" | "VARBINARY" | "BINARY VARYING" => "VARBYTE",
+ // e.g. "timestamp with time zone"
+ "TIMESTAMPTZ" | "TIMESTAMP WITH TIME ZONE" => "TIMESTAMPTZ",
+ _ => "UNKNOWN",
+ }
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(path) => {
+ // Messages were received and ingested to Redshift
+ if let Some(s3_path) = path {
+ // Truncate the staging table
+ if let Err(e) = self.staging_cleanup().await {
+ tracing::warn!(error = %e, "failed to cleanup staging table");
+ }
+
+ // Handle archiving
+ if let Err(e) = self.archive_parquet(&s3_path).await {
+ tracing::warn!(error = %e, "failed to archive parquet file: {}", s3_path);
+ }
+
+ self.state.lock().await.batches_loaded += 1
+ } else {
+ tracing::info!("Zero messages found for processing");
+ }
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ return Err(e);
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ // This function builds a parquet from messages and metadata
+ // It uploads the parquet to S3 and returns the path
+ // It then copies the parquet to Redshift target table via
+ // a staging table by means of a MERGE statement
+ // This function treats parquet-generation, uploading to s3,
+ // copying to staging and merging to target as atomic
+ // process of focus for this sink connector
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<Option<String>, Error> {
+ if messages.is_empty() {
+ return Ok(None);
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+
+ let schema = record_batch.schema();
+ let cols = schema
+ .fields
+ .iter()
+ .map(|f| f.name().as_str())
+ .collect::<Vec<_>>();
+
+ // Copy the parquet file to Redshift staging
+ // Cleanup
+ tracing::info!("copying parquet to Redshift staging");
+ if let Err(e) = self.copy_parquet(&s3_path, &cols).await {
+ let key = s3_path
+ .strip_prefix(&format!("s3://{}/", self.config.s3_bucket))
+ .ok_or(Error::InvalidConfigValue("Missing Cleanup S3 path".into()))?;
+
+ self.delete_object(key).await?;
+
+ Err(e)?
+ }
+
+ tracing::info!("Redshift staging COPY completed");
+
+ // Do a merge into Redshift target table
+ self.insert_into_target(&cols).await?;
+
+ tracing::info!("Redshift target table merge completed");
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(Some(s3_path))
+ }
+
+ async fn copy_parquet(&self, s3_path: &str, cols: &[&str]) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+
+ let sql = self.build_copy_sql(&staging_table, s3_path, &cols.join(", "))?;
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "Redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(staging_table = staging_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ async fn insert_into_target(&self, cols: &[&str]) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let target_table = quote_identifier(&self.config.target_table)?;
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+ let sql = self.build_insert_sql(cols, &staging_table, &target_table);
+
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, "issuing Redshift MERGE");
+
+ retry_with_backoff(
+ "Redshift INSERT",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(staging_table = %staging_table, target_table = %target_table, "Redshift INSERT completed");
+
+ Ok(())
+ }
+
+ async fn staging_cleanup(&self) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+ let sql = self.build_truncate_sql(&staging_table);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, "issuing Redshift TRUNCATE");
+
+ retry_with_backoff(
+ "Redshift TRUNCATE",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift TRUNCATE completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self, table_name: &str) -> Result<String, Error> {
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut query = format!("CREATE TABLE IF NOT EXISTS {table_name} (id VARCHAR(40)");
+
+ if include_metadata {
+ query.push_str(", iggy_offset VARCHAR(20), iggy_timestamp VARCHAR(20), iggy_stream VARCHAR, iggy_topic VARCHAR, iggy_partition_id BIGINT");
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp VARCHAR(20)");
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at VARCHAR);");
+
+ Ok(query)
+ }
+
+ fn build_copy_sql(
+ &self,
+ staging_table: &str,
+ s3_path: &str,
+ cols: &str,
+ ) -> Result<String, Error> {
+ // Redshift allows this from the docs
+ // https://docs.aws.amazon.com/redshift/latest/dg/r_COPY_command_examples.html
+ let iam_role = quote_identifier(&self.config.aws_iam_role)?.replace('"', "");
+
+ let region = quote_identifier(&self.config.aws_region)?.replace('"', "");
+
+ Ok(format!(
+ "COPY {} ({}) FROM '{}' CREDENTIALS 'aws_iam_role={}' FORMAT AS PARQUET REGION '{}';",
Review Comment:
`REGION` is not supported with parquet COPY. the aws notes on columnar COPY (https://docs.aws.amazon.com/redshift/latest/dg/copy-usage_notes-copy-from-columnar.html) say it fails with "REGION argument is not supported for PARQUET based COPY", and it's missing from their supported parameter list. the mock can't catch this, `execute_copy` only checks for `Parquet` and ignores the rest. drop the clause - COPY defaults to the cluster region and the bucket has to be in that region anyway. also, the thread only has DDL screenshots so far, one pasted real COPY output would settle it.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5335238123
@hubcio I have resolved the conflict that was there on the Cargo.lock
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3675964303
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
Review Comment:
@realonbebeto
it took me a while to actually understand the question. so, thing is the mock isn't really validating those 4 statements. `map_type` rewrites `DECIMAL(_)` to VARCHAR and `VARBYTE` to BYTEA, and between `strip_created_at` and the column list rebuilt from `information_schema`, the COPY that runs is a postgres one with an explicit column list, not the columnless `FORMAT AS PARQUET` you emit. VARCHAR(40) fixes the decimal but not that.
maybe it'd better if you rather `map_type` error on what it can't represent than substitute? plusofc a readme line on what the suite actually covers.
schema mismatch in open(), yeah. postgres_sink isn't precedent, just missing there too, and a bound INSERT names the bad column, whereas a failed COPY just points you at the load-error views.
iam/s3 is environment mostly, but `validate()` requires static keys and `Credentials::new` gets None for both token slots, so temp creds and instance profiles are out. from whatisee aws-creds already falls back to env/profile/container when the access key is None, so making those two fields `Option` and dropping the validate() check is most of it.
nothing per-PR works for a real cluster, fork PRs get no secrets. before merge i'd want you running it once on serverless and pasting it - CREATE TABLE verbatim, one COPY of a parquet the connector made, same file again for MERGE, rows read back. then `#[ignore]`d behind `REDSHIFT_TEST_DSN`.
dropping PKs i agree with, but the mock's postgres was the only uniqueness check in the suite, so MERGE needs its own test.
TLDR:
1. fail loudly, no quiet translation
2. update readme with test suite descr
3. maybe run it once against redshift serverless if you have access
4. make aws keys optional
5. merge dedup test, not sure about that
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611508905
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
+ // uses a Simple Query Handler that doesn't support prepared statements with binds.
+ let credentials = format!(
+ "CREDENTIALS 'ACCESS_KEY_ID={};SECRET_ACCESS_KEY={}'",
+ self.config.aws_access_key_id.expose_secret(),
+ self.config.aws_secret_access_key.expose_secret()
+ );
+
+ format!(
+ "COPY {} FROM '{}' {} FORMAT AS PARQUET REGION '{}';",
+ self.config.target_table, s3_path, credentials, self.config.aws_region
+ )
+ }
+
+ async fn upload_parquet(&self, content: &[u8]) -> Result<String, Error> {
+ let file_id = Uuid::now_v7();
+ let key = build_s3_key(&self.config.s3_prefix, &format!("{file_id}.parquet"));
+ let bucket = self.get_bucket()?;
+
+ tracing::debug!(key = %key, bytes = content.len(), "uploading parquet to S3");
+
+ let response = bucket.put_object(&key, content).await.map_err(|e| {
+ tracing::error!("Failed to upload to S3 key '{key}': {e}");
+ Error::Storage(format!("S3 upload failed: {e}"))
+ })?;
+
+ ensure_s3_status(response.status_code(), 200, "S3 upload")?;
+
+ let path = format!("s3://{}{}", bucket.name(), key);
+ tracing::info!(path = %path, bytes = content.len(), "uploaded parquet to S3");
+
+ Ok(path)
+ }
+
+ async fn archive_parquet(&self, key: &str) -> Result<(), Error> {
+ let old_key = key
+ .strip_prefix(&format!("s3://{}/", self.config.s3_bucket))
+ .unwrap_or(key);
+
+ if !self.get_archive() {
+ self.delete_object(old_key).await?;
+ tracing::info!(key = old_key, "deleted parquet file (archiving disabled)");
+ return Ok(());
+ }
+
+ let bucket = self.get_bucket()?;
+ let prefix = self.config.s3_prefix.trim_matches('/');
+ let archived_key = old_key.replacen(prefix, DEFAULT_ARCHIVE_PREFIX.trim_matches('/'), 1);
+
+ tracing::debug!(from = old_key, to = %archived_key, "archiving parquet file");
+
+ let status_code = bucket
+ .copy_object_internal(old_key, &archived_key)
+ .await
+ .map_err(|e| {
+ tracing::error!(key = old_key, error = %e, "failed to copy object for archiving");
+ Error::Storage(format!("S3 archiving failed: {e}"))
+ })?;
+
+ ensure_s3_status(status_code, 200, "S3 archive copy")?;
+
+ self.delete_object(old_key).await?;
+ tracing::info!(archived_to = %archived_key, "archived parquet file");
+
+ Ok(())
+ }
+
+ async fn delete_object(&self, key: &str) -> Result<(), Error> {
+ let bucket = self.get_bucket()?;
+
+ let response = bucket.delete_object(key).await.map_err(|e| {
+ tracing::error!(key, error = %e, "failed to delete S3 object");
+ Error::Storage(format!("S3 deleting failed: {e}"))
+ })?;
+ ensure_s3_status(response.status_code(), 204, "S3 object deletion")?;
+
+ tracing::debug!(key, "deleted S3 object");
+ Ok(())
+ }
+
+ fn get_pool(&self) -> Result<&Pool<Postgres>, Error> {
+ self.pool
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Database not connected".to_string()))
+ }
+
+ fn get_bucket(&self) -> Result<&Bucket, Error> {
+ let r = self
+ .bucket
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Database not connected".to_string()))?;
+
+ Ok(r)
+ }
+
+ fn payload_format(&self) -> PayloadFormat {
+ PayloadFormat::from_config(self.config.payload_format.as_deref())
+ }
+
+ fn get_max_retries(&self) -> u32 {
+ self.config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES)
+ }
+
+ fn get_retry_delay(&self) -> Duration {
+ self.config
+ .retry_delay
+ .as_deref()
+ .unwrap_or(DEFAULT_RETRY_DELAY)
+ .parse::<HumanDuration>()
+ .map(Into::into)
+ .unwrap_or_else(|_| Duration::from_secs(1))
+ }
+
+ fn get_archive(&self) -> bool {
+ self.config.archive.unwrap_or(false)
+ }
+}
+
+#[derive(Debug, Default)]
+struct State {
+ messages_processed: u64,
+ batches_loaded: u64,
+ insertion_errors: u64,
+}
+
+/// Generic retry helper with linear backoff, used for transient warehouse errors.
+async fn retry_with_backoff<F, Fut, T>(
+ operation: &str,
+ max_retries: u32,
+ base_delay: Duration,
+ is_transient: impl Fn(&sqlx::Error) -> bool,
+ mut op: F,
+) -> Result<T, Error>
+where
+ F: FnMut() -> Fut,
+ Fut: Future<Output = Result<T, sqlx::Error>>,
+{
+ let mut attempts = 0u32;
+
+ loop {
+ match op().await {
+ Ok(value) => return Ok(value),
+ Err(e) => {
+ attempts += 1;
+ let transient = is_transient(&e);
+
+ if !transient || attempts >= max_retries {
+ tracing::error!(operation, attempts, error = %e, "operation failed permanently");
+ return Err(Error::CannotStoreData(format!(
+ "{operation} failed after {attempts} attempts: {e}"
+ )));
+ }
+
+ tracing::warn!(operation, attempts, max_retries, error = %e, "transient error, retrying");
+ tokio::time::sleep(base_delay * attempts).await;
+ }
+ }
+ }
+}
+
+fn encode_parquet(batch: &RecordBatch) -> Result<Vec<u8>, Error> {
+ let mut content = Vec::new();
+ let mut writer = ArrowWriter::try_new(&mut content, batch.schema(), None).map_err(|e| {
+ tracing::error!(error = %e, "failed to create parquet writer");
+ Error::WriteFailure(format!("Failed to create parquet writer: {e}"))
+ })?;
+
+ writer.write(batch).map_err(|e| {
+ tracing::error!(error = %e, "failed to write parquet batch");
+ Error::WriteFailure(format!("Failed to write parquet: {e}"))
+ })?;
+
+ writer.close().map_err(|e| {
+ tracing::error!(error = %e, "failed to close parquet writer");
+ Error::WriteFailure(format!("Failed to close writer: {e}"))
+ })?;
+
+ Ok(content)
+}
+
+fn build_s3_key(prefix: &str, filename: &str) -> String {
+ if prefix.is_empty() {
+ format!("/{filename}")
+ } else {
+ format!("/{}/{filename}", prefix.trim_end_matches('/'))
+ }
+}
+
+fn ensure_s3_status<T: PartialEq + std::fmt::Display>(
+ status: T,
+ expected: T,
+ context: &str,
+) -> Result<(), Error> {
+ if status != expected {
+ tracing::error!(context, %status, "unexpected S3 response status");
+ return Err(Error::Storage(format!(
+ "{context} failed with status {status}"
+ )));
+ }
+ Ok(())
+}
+
+fn create_record_batch(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ include_metadata: bool,
+ include_checksum: bool,
+ include_origin_timestamp: bool,
+ payload_format: PayloadFormat,
+) -> Result<RecordBatch, Error> {
+ let mut fields = vec![Field::new("id", DataType::Decimal256(39, 0), false)];
+ let mut columns: Vec<ArrayRef> = vec![id_column(messages)?];
+
+ if include_metadata {
+ let (mut metadata_fields, mut metadata_columns) =
+ metadata_columns(topic_metadata, messages_metadata, messages);
+ fields.append(&mut metadata_fields);
+ columns.append(&mut metadata_columns);
+ }
+
+ if include_checksum {
+ fields.push(Field::new("iggy_checksum", DataType::Utf8, false));
+ columns.push(checksum_column(messages));
+ }
+
+ if include_origin_timestamp {
+ fields.push(Field::new(
+ "iggy_origin_timestamp",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ false,
+ ));
+ columns.push(origin_timestamp_column(messages));
+ }
+
+ fields.push(Field::new("payload", payload_format.arrow_type(), false));
+ columns.push(payload_column(messages, payload_format)?);
+
+ let schema = Arc::new(Schema::new(fields));
+ let batch =
+ RecordBatch::try_new(schema, columns).map_err(|e| Error::CannotStoreData(e.to_string()))?;
+
+ tracing::debug!(
+ rows = batch.num_rows(),
+ columns = batch.num_columns(),
+ "built record batch"
+ );
+
+ Ok(batch)
+}
+
+fn id_column(messages: &[ConsumedMessage]) -> Result<ArrayRef, Error> {
+ let ids = Decimal256Array::from_iter_values(
+ messages
+ .iter()
+ .map(|v| arrow::datatypes::i256::from_parts(v.id, 0)),
+ )
+ .with_precision_and_scale(39, 0)
+ .map_err(|e| Error::CannotStoreData(e.to_string()))?;
+
+ Ok(Arc::new(ids))
+}
+
+fn metadata_columns(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+) -> (Vec<Field>, Vec<ArrayRef>) {
+ let fields = vec![
+ Field::new("iggy_offset", DataType::Int64, false),
+ Field::new(
+ "iggy_timestamp",
+ DataType::Timestamp(TimeUnit::Microsecond, None),
+ false,
+ ),
+ Field::new("iggy_stream", DataType::Utf8, false),
+ Field::new("iggy_topic", DataType::Utf8, false),
+ Field::new("iggy_partition_id", DataType::Int32, false),
+ ];
+
+ let columns: Vec<ArrayRef> = vec![
+ Arc::new(Int64Array::from_iter_values(
+ messages.iter().map(|v| v.offset as i64),
+ )),
+ Arc::new(TimestampMicrosecondArray::from_iter_values(
+ messages.iter().map(|v| v.timestamp as i64),
+ )),
+ Arc::new(StringArray::from_iter_values(
+ (0..messages.len()).map(|_| topic_metadata.stream.clone()),
+ )),
+ Arc::new(StringArray::from_iter_values(
+ (0..messages.len()).map(|_| topic_metadata.topic.clone()),
+ )),
+ Arc::new(Int32Array::from_iter_values(
+ (0..messages.len()).map(|_| messages_metadata.partition_id as i32),
+ )),
+ ];
+
+ (fields, columns)
+}
+
+fn checksum_column(messages: &[ConsumedMessage]) -> ArrayRef {
+ Arc::new(StringArray::from_iter_values(
+ messages.iter().map(|v| v.checksum.to_string()),
+ ))
+}
+
+fn origin_timestamp_column(messages: &[ConsumedMessage]) -> ArrayRef {
+ Arc::new(TimestampMicrosecondArray::from_iter_values(
+ messages.iter().map(|v| v.origin_timestamp as i64),
+ ))
+}
+
+fn payload_column(messages: &[ConsumedMessage], format: PayloadFormat) -> Result<ArrayRef, Error> {
+ match format {
+ PayloadFormat::Varbyte => {
+ let values: Vec<Vec<u8>> = messages
+ .iter()
+ .map(|v| v.payload.clone().try_to_bytes())
+ .collect::<Result<_, _>>()?;
+ let slices: Vec<&[u8]> = values.iter().map(Vec::as_slice).collect();
+ Ok(Arc::new(BinaryArray::from_vec(slices)))
+ }
+ PayloadFormat::Text => {
+ let values: Vec<String> = messages
+ .iter()
+ .map(|v| {
+ let bytes = v.payload.try_to_bytes()?;
+ String::from_utf8(bytes).map_err(|_| Error::InvalidTextPayload)
+ })
+ .collect::<Result<_, _>>()?;
+ Ok(Arc::new(StringArray::from_iter_values(values.iter())))
+ }
+ PayloadFormat::Json => {
+ let values: Vec<String> = messages
+ .iter()
+ .map(|v| {
+ let bytes = v.payload.try_to_bytes()?;
+
+ Ok(serde_json::from_slice::<serde_json::Value>(&bytes)
+ .map_err(|_| Error::InvalidJsonPayload)?
+ .to_string())
+ })
+ .collect::<Result<_, _>>()?;
+ Ok(Arc::new(StringArray::from_iter_values(values.iter())))
+ }
+ }
+}
+
+fn redact_connection_string(conn_str: &str) -> String {
+ // Guard against very short strings
+ const PREVIEW_LEN: usize = 3;
+
+ if let Some(scheme_end) = conn_str.find("://") {
+ let scheme = &conn_str[..scheme_end + 3];
+ let rest = &conn_str[scheme_end + 3..];
+
+ // Stop preview at the first sensitive boundary
+ let safe_end = rest
+ .find([':', '@', '?', '/'])
+ .unwrap_or(rest.len())
+ .min(PREVIEW_LEN);
+
+ let preview = &rest[..safe_end];
Review Comment:
lib.rs:730 — redact_connection_string slices &rest[..safe_end] on byte index, not char boundary; multibyte DSN userinfo/host → panic across FFI in open(). **Fix:** char-safe slice (no-scheme branch at :734 already safe).
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611511575
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
Review Comment:
lib.rs:69,191-212 — CREATE TABLE IF NOT EXISTS with schema from current include_* flags; flipping a flag between runs keeps old table → positional COPY loads into wrong columns silently. **Fix:** validate live schema on open, error on drift.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611513469
##########
core/connectors/sinks/redshift_sink/README.md:
##########
@@ -0,0 +1,119 @@
+# Redshift Sink Connector
+
+Writes Apache Iggy stream messages into Amazon Redshift via S3-staged Parquet
+files and a `COPY` load.
+
+Each connector batch is serialized to a Parquet file and uploaded to the
+configured S3 bucket/prefix, then loaded into the target Redshift table with a
+`COPY` statement. This makes S3 a staging area rather than a destination in
+its own right — Redshift is the system of record for the data.
+
+Persistent load failures are at-most-once from the runtime's perspective:
+messages may already be committed in Iggy before this connector exhausts its
+write attempts, so failed loads are logged but not redelivered.
+
+## Configuration
+
+```toml
+type = "sink"
+key = "redshift"
+enabled = true
+version = 0
+name = "Redshift sink"
+path = "../../target/release/libiggy_connector_redshift_sink"
+verbose = false
+
+[[streams]]
+stream = "user_events"
+topics = ["users", "orders"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "redshift_sink"
+
+[plugin_config]
+connection_string = "postgresql://user:pass@localhost:5439/database"
+target_table = "iggy_messages"
+batch_size = 100
+max_connections = 10
+include_metadata = true
+include_checksum = true
+include_origin_timestamp = true
+payload_format = "varbyte"
+aws_access_key_id = "admin"
+aws_secret_access_key = "password"
+s3_bucket = "iggystaging"
+s3_prefix = "iggy/messages"
+s3_endpoint = "http://localhost:9000"
+aws_region = "us-east-1"
+archive = true
+```
+
+### Plugin Fields
+
+| Field | Required | Default | Description |
+| --- | --- | --- | --- |
+| `connection_string` | yes | — | Postgres-wire connection string used to reach the Redshift cluster and issue the `COPY` command. |
+| `target_table` | yes | — | Destination Redshift table that batches are copied into. |
+| `batch_size` | no | `100` | Number of messages buffered per Parquet file / `COPY` operation. |
+| `max_connections` | no | `5` | Size of the connection pool used against Redshift. |
+| `include_metadata` | no | `false` | Stores stream/topic/partition/offset/timestamp/schema fields alongside the payload. |
Review Comment:
README.md:60-62 — defaults say false; code defaults true (lib.rs:271-273,334-336). **Fix**: table → true.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5017739993
/author
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611499745
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
+ // uses a Simple Query Handler that doesn't support prepared statements with binds.
+ let credentials = format!(
Review Comment:
**AWS secret inlined in COPY SQL → log-leak + convention break.** lib.rs:369-378 bakes ACCESS_KEY_ID=...;SECRET_ACCESS_KEY=... into the SQL string via format!, run through sqlx::query(AssertSqlSafe(...)). sqlx surfaces statement text on its tracing target (DEBUG default; slow-query WARN — parquet COPY routinely >1s). Sibling postgres_sink never inlines: .bind() everything (postgres_sink/src/lib.rs:381-417). Secret also held as plain String, bypassing secrecy zeroize. **Fix:** use Redshift IAM_ROLE 'arn:...' (no embedded keys) or session-scoped credentials; never format!/log the COPY string; keep creds in SecretString.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5199046487
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.
If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either `@core` on Discord or by mentioning them directly here on the PR.
Thank you for your contribution!
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5207829401
@ryerraguntla I haven't been able to test the S3 → Redshift flow yet due to limitations with setting up a free-tier AWS account. The [fix(redshift): improved correctness and reliability](https://github.com/apache/iggy/pull/3654/commits/3708d133d5c0e2b1f06cf62ca2ba0af42d7b6d7c) is in place, but real Redshift testing is pending.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5207804287
/ready
--
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(connectors): add Redshift sink connector [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4993170653
@realonbebeto I can see that changes unrelated to the pr have also been committed. can you correct this please?
--
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(connectors): add Redshift sink connector [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4995888691
@realonbebeto the CI failed due to clippy.
We suggest you to setup prek locally, so that easier CI failures are caught locally.
Setup instructions here: https://github.com/apache/iggy/blob/c0218f1c7af0613b915ae70a7f5e1bab887c285e/CONTRIBUTING.md?plain=1#L80
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5415014967
/ready
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3649213930
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
Review Comment:
I've been thinking about this with the proposed design where id is VARCHAR(40), PKs are dropped, and uniqueness is enforced via MERGE from staging into the target.
The sink only emits a small, fixed set of Redshift queries. SELECT 1, CREATE TABLE, COPY, and MERGE are the only statements requiring dynamic column lists, which seems reasonable to validate with the pgwire mock.
The main gaps I see are:
- Malformed Parquet – I don't think this can be meaningfully tested without a real Redshift instance.
- Schema mismatch – This can be validated during open(), as pointed elsewhere in this review. And help clarify why the sibling postgres_sink does not enforce this? My assumption is the auto_create flag transfers the burden to the user to ensure there's no schema mismatch(by means of migrations)
- IAM/S3 permissions – This is more of an user environment/configuration issue than connector behaviour unless we want to check for S3 and Redshift permissions from the iam_role, access_key and secret_key config provide.
Given that, I'm wondering how the smoke test against a real Redshift instance would be to cover the 4 queries semantics.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3602682279
##########
Cargo.lock:
##########
Review Comment:
what is reason for so many changes in this file?
--
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(connectors): add Redshift sink connector [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3839729538
##########
core/connectors/sinks/redshift_sink/src/config.rs:
##########
@@ -0,0 +1,146 @@
+// 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 arrow::datatypes::DataType;
+use iggy_connector_sdk::Error;
+use secrecy::{ExposeSecret, SecretString};
+
+/// Configuration for the Redshift Sink
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RedshiftSinkConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+ pub target_table: String,
+ pub batch_size: Option<u32>,
+ pub max_connections: Option<u32>,
+ pub include_metadata: Option<bool>,
+ pub include_checksum: Option<bool>,
+ pub include_origin_timestamp: Option<bool>,
+ pub payload_format: Option<String>,
+ pub verbose_logging: Option<bool>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ /// aws_access_key_id and aws_secret_access_key MUST be provided
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")]
+ pub aws_access_key_id: Option<SecretString>,
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")]
+ pub aws_secret_access_key: Option<SecretString>,
+ pub aws_iam_role: String,
+ pub s3_bucket: String,
+ pub s3_prefix: String,
+ pub s3_endpoint: Option<String>,
+ pub aws_region: String,
+ /// Offers the option to archive staged S3 files after COPY
+ /// Defaults to deletion once COPY completes
+ /// Files are moved to different prefix within the same bucket
+ pub archive: Option<bool>,
+}
+
+impl RedshiftSinkConfig {
+ pub fn validate(&self) -> Result<(), Error> {
+ let mut errors = String::new();
+
+ if self.connection_string.expose_secret().is_empty() {
+ errors.push_str("connection_string is empty\n");
+ }
+
+ if self.target_table.is_empty() {
+ errors.push_str(", target_table is empty\n");
+ }
+
+ if self.s3_bucket.is_empty() {
+ errors.push_str(", s3_bucket is empty\n");
+ }
+
+ if self.aws_region.is_empty() {
+ errors.push_str(", aws_region is empty\n");
+ }
+
+ if self.aws_iam_role.is_empty() {
+ errors.push_str(", aws_iam_role is empty\n");
+ }
+
+ if let (Some(access), Some(secret)) = (&self.aws_access_key_id, &self.aws_secret_access_key)
Review Comment:
Static S3 credentials need to be validated as a pair. This branch only checks them when both options are present, so an access key without a secret, or vice versa, is accepted. `Credentials::new` then treats the access key as explicit credentials and does not fall back to the environment or instance-role chain; the first S3 upload fails after startup. Reject partial credentials while preserving the both-absent provider-chain case.
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1482 @@
+// 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.
+
+mod config;
+
+use std::{collections::BTreeMap, str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{ArrayRef, BinaryBuilder, Int64Array, RecordBatch, StringArray, StringBuilder},
+ datatypes::{DataType, Field, Schema},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::{
+ arrow::ArrowWriter,
+ basic::{Compression, ZstdLevel},
+ file::properties::WriterProperties,
+};
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, Row, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ // Ensuring tables exist
+ self.ensure_tables_exist().await?;
+ // Checking for schema drift
+ self.ensure_schema_match().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ self.config
+ .aws_access_key_id
+ .as_ref()
+ .map(|v| v.expose_secret()),
+ self.config
+ .aws_secret_access_key
+ .as_ref()
+ .map(|v| v.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_tables_exist(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let target_table = quote_identifier(&self.config.target_table)?;
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+
+ let payload_type = self.payload_format().sql_type();
+
+ let target_query = self.build_create_table_sql(&target_table)?;
+
+ let staging_query = self.build_create_table_sql(&staging_table)?;
+
+ tracing::debug!("ensuring staging and target tables exist");
+
+ sqlx::query(AssertSqlSafe(staging_query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{staging_table}': {e}"))
+ })?;
+
+ tracing::debug!("Staging table created");
+
+ sqlx::query(AssertSqlSafe(target_query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{target_table}': {e}"))
+ })?;
+
+ tracing::info!(
+ staging_table = staging_table,
+ target_table = target_table,
+ payload_type,
+ "staging and target tables ready"
+ );
+
+ Ok(())
+ }
+
+ // This method ensures that the target table schema matches the expected schema.
+ // it also verifies there is a created_at column
+ async fn ensure_schema_match(&self) -> Result<(), Error> {
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let target_table = quote_identifier(&self.config.target_table)?;
+ let staging_table = quote_identifier(&format!("staging_{}", self.config.target_table))?;
+ let payload_type = self.payload_format().sql_type();
+ let pool = self.get_pool()?;
+
+ let mut expected_cols: BTreeMap<&str, &str> = BTreeMap::new();
+ expected_cols.insert("id", "VARCHAR");
+ if include_metadata {
+ expected_cols.insert("iggy_offset", "VARCHAR");
+ expected_cols.insert("iggy_timestamp", "VARCHAR");
+ expected_cols.insert("iggy_stream", "VARCHAR");
+ expected_cols.insert("iggy_topic", "VARCHAR");
+ expected_cols.insert("iggy_partition_id", "BIGINT");
+ }
+ if include_checksum {
+ expected_cols.insert("iggy_checksum", "VARCHAR");
+ }
+ if include_origin_timestamp {
+ expected_cols.insert("iggy_origin_timestamp", "VARCHAR");
+ }
+ expected_cols.insert("payload", payload_type);
+ expected_cols.insert("created_at", "VARCHAR");
+
+ let target_cols = Self::load_columns(pool, &target_table).await?;
+ let staging_cols = Self::load_columns(pool, &staging_table).await?;
+
+ let mut mismatches = Self::diff_schema(&target_table, &target_cols, &expected_cols);
+ mismatches.extend(Self::diff_schema(
+ &staging_table,
+ &staging_cols,
+ &expected_cols,
+ ));
+
+ tracing::info!("Mismatches: {:?}", mismatches);
+
+ if !mismatches.is_empty() {
+ return Err(Error::InitError(format!(
+ "Schema mismatch detected:\n{}",
+ mismatches.join("\n")
+ )));
+ }
+
+ Ok(())
+ }
+
+ fn diff_schema(
+ table_name: &str,
+ actual_cols: &BTreeMap<String, String>,
+ expected_cols: &BTreeMap<&str, &str>,
+ ) -> Vec<String> {
+ let mut errors = Vec::new();
+
+ for (col_name, expected_type) in expected_cols {
+ match actual_cols.get(*col_name) {
+ None => errors.push(format!(
+ "{table_name}: missing column '{col_name}' (expected {expected_type})"
+ )),
+ Some(actual_type) if !Self::type_matches(actual_type, expected_type) => errors.push(format!(
+ "{table_name}: column '{col_name}' type mismatch — expected {expected_type}, found {actual_type}"
+ )),
+ _ => {}
+ }
+ }
+ errors
+ }
+
+ fn type_matches(actual: &str, expected: &str) -> bool {
+ Self::normalize_type(actual) == Self::normalize_type(expected)
+ }
+
+ async fn load_columns(
+ pool: &sqlx::PgPool,
+ table: &str,
+ ) -> Result<BTreeMap<String, String>, Error> {
+ let query = format!(
+ "SELECT \"column\", type FROM pg_table_def WHERE tablename = '{}'",
+ table.replace('"', "")
+ );
+
+ let rows = sqlx::query(AssertSqlSafe(query))
+ .fetch_all(pool)
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to read schema for '{table}': {e}")))?;
+
+ if rows.is_empty() {
+ return Err(Error::InitError(format!(
+ "Table '{table}' was not found or has no visible columns"
+ )));
+ }
+
+ rows.into_iter()
+ .map(|row| {
+ Ok((
+ row.try_get::<String, _>("column")
+ .map_err(|e| Error::InitError(e.to_string()))?,
+ Self::normalize_type(
+ &row.try_get::<String, _>("type")
+ .map_err(|e| Error::InitError(e.to_string()))?,
+ )
+ .to_string(),
+ ))
+ })
+ .collect()
+ }
+
+ fn normalize_type(t: &str) -> &'static str {
+ let base = t.split('(').next().unwrap_or(t).trim();
+
+ match base.to_ascii_uppercase().as_str() {
+ "INTEGER" | "INT" | "INT4" => "INTEGER",
+ // e.g. "bigint"
+ "INT8" | "BIGINT" => "BIGINT",
+ // e.g "character varying(40)", "character varying(20)", "character varying(256)"
+ "TEXT" | "VARCHAR" | "CHARACTER VARYING" => "VARCHAR",
+ // Having bytea because of the Postgres Test
+ // e.g. "binary varying(64000)"
+ "BYTEA" | "VARBYTE" | "VARBINARY" | "BINARY VARYING" => "VARBYTE",
+ // e.g. "timestamp with time zone"
+ "TIMESTAMPTZ" | "TIMESTAMP WITH TIME ZONE" => "TIMESTAMPTZ",
+ _ => "UNKNOWN",
+ }
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
Review Comment:
batch_size = 0 is accepted and reaches `messages.chunks(batch_size)`, which panics. Because this is invoked through the plugin’s extern "C" boundary, a bad operator config can abort the connectors process instead of
returning a configuration error. Please reject zero during startup or clamp the stored value to at least one
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1482 @@
+// 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.
+
+mod config;
+
+use std::{collections::BTreeMap, str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{ArrayRef, BinaryBuilder, Int64Array, RecordBatch, StringArray, StringBuilder},
+ datatypes::{DataType, Field, Schema},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::{
+ arrow::ArrowWriter,
+ basic::{Compression, ZstdLevel},
+ file::properties::WriterProperties,
+};
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, Row, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ // Ensuring tables exist
+ self.ensure_tables_exist().await?;
+ // Checking for schema drift
+ self.ensure_schema_match().await?;
+ Ok(())
Review Comment:
`RedshiftSinkConfig::validate()` is never called. An empty `aws_iam_role` therefore lets the connector start, create tables, and upload the Parquet file before failing while building the later `COPY`. Both shipped examples omit this field despite documenting it as required. Please call `self.config.validate()?` before connecting and add `aws_iam_role` to `config.toml` and the README example.
##########
core/connectors/sinks/redshift_sink/src/config.rs:
##########
@@ -0,0 +1,146 @@
+// 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 arrow::datatypes::DataType;
+use iggy_connector_sdk::Error;
+use secrecy::{ExposeSecret, SecretString};
+
+/// Configuration for the Redshift Sink
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RedshiftSinkConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+ pub target_table: String,
+ pub batch_size: Option<u32>,
+ pub max_connections: Option<u32>,
+ pub include_metadata: Option<bool>,
+ pub include_checksum: Option<bool>,
+ pub include_origin_timestamp: Option<bool>,
+ pub payload_format: Option<String>,
+ pub verbose_logging: Option<bool>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ /// aws_access_key_id and aws_secret_access_key MUST be provided
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")]
+ pub aws_access_key_id: Option<SecretString>,
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")]
+ pub aws_secret_access_key: Option<SecretString>,
+ pub aws_iam_role: String,
+ pub s3_bucket: String,
+ pub s3_prefix: String,
+ pub s3_endpoint: Option<String>,
+ pub aws_region: String,
+ /// Offers the option to archive staged S3 files after COPY
+ /// Defaults to deletion once COPY completes
+ /// Files are moved to different prefix within the same bucket
+ pub archive: Option<bool>,
+}
+
+impl RedshiftSinkConfig {
+ pub fn validate(&self) -> Result<(), Error> {
+ let mut errors = String::new();
+
+ if self.connection_string.expose_secret().is_empty() {
+ errors.push_str("connection_string is empty\n");
+ }
+
+ if self.target_table.is_empty() {
+ errors.push_str(", target_table is empty\n");
+ }
+
+ if self.s3_bucket.is_empty() {
+ errors.push_str(", s3_bucket is empty\n");
+ }
+
+ if self.aws_region.is_empty() {
+ errors.push_str(", aws_region is empty\n");
+ }
+
+ if self.aws_iam_role.is_empty() {
+ errors.push_str(", aws_iam_role is empty\n");
+ }
+
+ if let (Some(access), Some(secret)) = (&self.aws_access_key_id, &self.aws_secret_access_key)
+ {
+ // Validate AWS credentials: access keys must be provided
+ let has_access_key = !access.expose_secret().is_empty();
+
+ let has_secret_key = !secret.expose_secret().is_empty();
+
+ if !(has_access_key && has_secret_key) {
+ errors.push_str(", aws_access_key_id and aws_secret_access_key are empty\n");
+ }
+ }
+
+ if !errors.is_empty() {
+ Err(Error::InvalidConfigValue(errors))
+ } else {
+ Ok(())
+ }
+ }
+}
+
+/// This connector supports:
+/// 1. Byte -> which has VARBYTE as the Redshift equivalent
+/// 2. Text -> which has VARCHAR as the Redshift equivalent
+///
+/// We dont have Json because we are using parquet as a means to sink ingestion
+/// As at the development of this connector there's no direct parquet type that matches JSON
+/// For JSON needs Redshift has SUPER(VARCHAR can be parsed by JSON_PARSE)ß
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+ Text,
+ #[default]
+ Varbyte,
+}
+
+impl PayloadFormat {
+ pub fn from_config(s: Option<&str>) -> Self {
+ match s.map(|s| s.to_lowercase()).as_deref() {
+ Some("text") => PayloadFormat::Text,
+ Some("json") => {
+ tracing::warn!("Json is not supported, falling back to Text");
+ PayloadFormat::Text
+ }
+
+ other => {
+ if other.is_some() {
+ tracing::warn!(
+ "Unrecognized payload_format {:?}, falling back to VARBYTE",
+ other
+ );
+ }
+
+ PayloadFormat::Varbyte
+ }
+ }
+ }
+
+ pub fn sql_type(&self) -> &'static str {
+ match self {
+ PayloadFormat::Varbyte => "VARBYTE",
+ PayloadFormat::Text => "VARCHAR",
+ }
+ }
Review Comment:
The generated payload types are too small for general Iggy messages: bare `VARCHAR` becomes `VARCHAR(256)` in Redshift, and bare `VARBYTE` defaults to 64,000 bytes. Text/JSON payloads over 256 bytes, and binary payloads over 64 KB, fail during `COPY`. Please declare explicit supported maxima, such as `VARCHAR(MAX)` and a suitable `VARBYTE(n)`, or validate and document a lower connector payload limit.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4974044697
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3654?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 72.70%. Comparing base ([`8e12709`](https://app.codecov.io/gh/apache/iggy/commit/8e12709c5f8ab3cb02653d16d27b6e52bb5f553d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`753da90`](https://app.codecov.io/gh/apache/iggy/commit/753da90ebca4e7c9382b4740e62af650e21b6365?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 14 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3654 +/- ##
============================================
- Coverage 73.43% 72.70% -0.74%
Complexity 937 937
============================================
Files 1286 1223 -63
Lines 141019 136332 -4687
Branches 116908 116820 -88
============================================
- Hits 103563 99125 -4438
+ Misses 34226 33974 -252
- Partials 3230 3233 +3
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.65% <ø> (ø)` | |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.44% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.12% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.65% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.29% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `87.27% <ø> (-4.08%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3654/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `42.28% <ø> (ø)` | |
[see 81 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3654/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4976823592
@hubcio resolved the CI failures.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4949987233
<!-- iggy-pr-triage-welcome -->
Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
Slash commands (own line, regular comment) move it around the queue:
- `/ready` - back to `S-waiting-on-review` after addressing feedback
- `/author` - flip to `S-waiting-on-author` while you finish changes
- `/request-review @user-or-team` - request a reviewer
See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-4994230151
@slbotbm made the unrelated changes correction
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611492231
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
Review Comment:
lib.rs:366-379 emits COPY t FROM ... with no (cols); table always carries created_at DEFAULT GETDATE() (:360) that parquet never writes. Real Redshift COPY-parquet maps by position → count/shape mismatch. Mock hides it: explicit COPY t (col_list) FROM STDIN BINARY + strip_created_at (rs_mock_load.rs:283,486-490). **Fix:** emit an explicit column list matching parquet fields exactly; verify created_at DEFAULT against real Redshift.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3602705240
##########
Cargo.lock:
##########
Review Comment:
I did a cargo update to resolve staleness. This is the context: https://github.com/apache/iggy/pull/3654#issuecomment-4989491660
--
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(connectors): add Redshift sink connector [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3602682279
##########
Cargo.lock:
##########
Review Comment:
what is reason for changes in this file?
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611488880
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
Review Comment:
lib.rs:342. Redshift caps NUMERIC precision at 38; ensure_table_exists fails. Mock hides it: map_type maps numeric/Decimal→VARCHAR (mock_load.rs:426,446-447). Also u128 id (39 digits) fits no legal DECIMAL(≤38) even after cap. **Fix:** store id as VARCHAR(40) (or split hi/lo BIGINT); update Arrow schema + mock to match
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611494106
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
Review Comment:
lib.rs:222-258: process_messages catches insert_batch Err, bumps insertion_errors, continues, returns Ok(()). Runtime auto-commits offsets (AutoCommitWhen::PollingMessages) → failed batches dropped permanently, no redelivery. README:11-13 admits at-most-once, but this is silent partial loss inside a multi-batch consume. **Fix:** return Err on first batch failure and stop the pipeline; pair with a runtime offset-commit-on-error fix or a persisted failed-batch manifest.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5017739488
Bunch of nits, but needs to be looked into and fixed
1. messages_processed counts failed batches
- yet README:118 claims loaded-only (lib.rs:237-238) ·
- upload_parquet has no retry while COPY does (:381-399) ·
- naive-micros→TIMESTAMPTZ UTC assumption (:596-598) ·
- lossy as i32/as i64 casts (:649,661) ·
- empty s3_prefix → malformed archive key (:413-414) ·
2. dead params_per_row return (:330-363) ·
3. invalid crates.io category "warehouse" → "database" (Cargo.toml:25) ·
4. const ICEBERG_SINK_KEY = "redshift" copy-paste (int.rs:36) ·
5. "IcebergFixture" error strings + "bytea" fixture doc mislabels (fixtures sink.rs:393,429,198,245,292) ·
6. typos "Reshift"/"vis S3"/"Reshshift"/"We dont" ·
7. payload_column Vec churn + uncompressed parquet writer
Did you check "typos" as part of precheck?
Good effort. Thanks for chiming in to contribute.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611497528
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
Review Comment:
lib.rs:342,311-323,740. id PRIMARY KEY advertises idempotency Redshift does not enforce (PK informational). Transient Io/08006/57P01 after a server-side commit re-runs the same COPY on the same S3 file → dup rows. Mock hides it: backing Postgres enforces PK (create.rs:111-112) → unique-violation, opposite of Redshift. **Fix:** COPY into a staging table + MERGE/dedup on id, or manifest-idempotent load; stop relying on PK.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3611514717
##########
core/connectors/sinks/redshift_sink/src/config.rs:
##########
@@ -0,0 +1,127 @@
+// 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 arrow::datatypes::DataType;
+use iggy_connector_sdk::Error;
+use secrecy::{ExposeSecret, SecretString};
+
+/// Configuration for the Redshift Sink
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct RedshiftSinkConfig {
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub connection_string: SecretString,
+ pub target_table: String,
+ pub batch_size: Option<u32>,
+ pub max_connections: Option<u32>,
+ pub include_metadata: Option<bool>,
+ pub include_checksum: Option<bool>,
+ pub include_origin_timestamp: Option<bool>,
+ pub payload_format: Option<String>,
+ pub verbose_logging: Option<bool>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ /// aws_access_key_id and aws_secret_access_key MUST be provided
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub aws_access_key_id: SecretString,
+ #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")]
+ pub aws_secret_access_key: SecretString,
+ pub s3_bucket: String,
+ pub s3_prefix: String,
+ pub s3_endpoint: Option<String>,
+ pub aws_region: String,
+ /// Offers the option to archive staged S3 files after COPY
+ /// Defaults to deletion once COPY completes
+ /// Files are moved to different prefix within the same bucket
+ pub archive: Option<bool>,
+}
+
+impl RedshiftSinkConfig {
+ pub fn validate(&self) -> Result<(), Error> {
+ let mut errors = String::new();
+
+ if self.connection_string.expose_secret().is_empty() {
+ errors.push_str("connection_string is empty\n");
+ }
+
+ if self.target_table.is_empty() {
+ errors.push_str(", target_table is empty\n");
+ }
+
+ if self.s3_bucket.is_empty() {
+ errors.push_str(", s3_bucket is empty\n");
+ }
+
+ if self.aws_region.is_empty() {
+ errors.push_str(", aws_region is empty\n");
+ }
+
+ // Validate AWS credentials: access keys must be provided
+ let has_access_key = !self.aws_access_key_id.expose_secret().is_empty();
+
+ let has_secret_key = !self.aws_secret_access_key.expose_secret().is_empty();
+
+ if !(has_access_key && has_secret_key) {
+ errors.push_str(", aws_access_key_id and aws_secret_access_key are empty\n");
+ }
+
+ if !errors.is_empty() {
+ Err(Error::InvalidConfigValue(errors))
+ } else {
+ Ok(())
+ }
+ }
+}
+
+/// This connector supports:
+/// 1. Byte -> which has VARBYTE as the Redshift equivalent
+/// 2. Text -> which has VARCHAR as the Redshift equivalent
+///
+/// We dont have Json because we are using parquet as a means to sink ingestion
+/// As at the development of this connector there's no direct parquet type that matches JSON
+/// For JSON needs Reshshift has SUPER(VARCHAR can be parsed by JSON_PARSE)
+#[allow(unused)]
Review Comment:
config.rs:96-103 — dead PayloadFormat::Json masked by #[allow(unused)]; from_config never yields it; "json" silently → Text. **Fix**: delete variant + attr, or log the alias.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on PR #3654:
URL: https://github.com/apache/iggy/pull/3654#issuecomment-5036293633
> Bunch of nits, but needs to be looked into and fixed
>
> 1. messages_processed counts failed batches
>
> * yet README:118 claims loaded-only (lib.rs:237-238) ·
> * upload_parquet has no retry while COPY does (:381-399) ·
> * naive-micros→TIMESTAMPTZ UTC assumption (:596-598) ·
> * lossy as i32/as i64 casts (:649,661) ·
> * empty s3_prefix → malformed archive key (:413-414) ·
> 2. dead params_per_row return (:330-363) ·
> 3. invalid crates.io category "warehouse" → "database" (Cargo.toml:25) ·
> 4. const ICEBERG_SINK_KEY = "redshift" copy-paste (int.rs:36) ·
> 5. "IcebergFixture" error strings + "bytea" fixture doc mislabels (fixtures sink.rs:393,429,198,245,292) ·
> 6. typos "Reshift"/"vis S3"/"Reshshift"/"We dont" ·
> 7. payload_column Vec churn + uncompressed parquet writer
>
> Did you check "typos" as part of precheck?
>
> Good effort. Thanks for chiming in to contribute.
Thanks for the review. I might have missed running 'typos'.
--
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(connectors): add Redshift sink connector [iggy]
Posted by "realonbebeto (via GitHub)" <gi...@apache.org>.
realonbebeto commented on code in PR #3654:
URL: https://github.com/apache/iggy/pull/3654#discussion_r3649213930
##########
core/connectors/sinks/redshift_sink/src/lib.rs:
##########
@@ -0,0 +1,1182 @@
+// 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.
+
+mod config;
+
+use std::{str::FromStr, sync::Arc, time::Duration};
+
+use arrow::{
+ array::{
+ ArrayRef, BinaryArray, Decimal256Array, Int32Array, Int64Array, RecordBatch, StringArray,
+ TimestampMicrosecondArray,
+ },
+ datatypes::{DataType, Field, Schema, TimeUnit},
+};
+use async_trait::async_trait;
+use humantime::Duration as HumanDuration;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use parquet::arrow::ArrowWriter;
+use s3::{Bucket, Region, creds::Credentials};
+use secrecy::ExposeSecret;
+use sqlx::{AssertSqlSafe, Pool, Postgres, postgres::PgPoolOptions};
+use tokio::sync::Mutex;
+use uuid::Uuid;
+
+use crate::config::{PayloadFormat, RedshiftSinkConfig};
+
+sink_connector!(RedshiftSink);
+
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "1s";
+const DEFAULT_MAX_CONNECTIONS: u32 = 5;
+const DEFAULT_ARCHIVE_PREFIX: &str = "archive/messages";
+
+#[derive(Debug)]
+pub struct RedshiftSink {
+ pub id: u32,
+ config: RedshiftSinkConfig,
+ pool: Option<Pool<Postgres>>,
+ state: Mutex<State>,
+ verbose: bool,
+ bucket: Option<Box<Bucket>>,
+}
+
+#[async_trait]
+impl Sink for RedshiftSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ tracing::info!(
+ sink_id = self.id,
+ table = %self.config.target_table, "opening Redshift sink connector"
+ );
+
+ self.connect().await?;
+ self.ensure_table_exists().await?;
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ tracing::debug!(
+ sink_id = self.id,
+ count = messages.len(),
+ "consuming messages"
+ );
+ self.process_messages(topic_metadata, &messages_metadata, &messages)
+ .await
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ tracing::info!(sink_id = self.id, "closing Redshift sink connector");
+
+ if let Some(pool) = self.pool.take() {
+ pool.close().await;
+
+ tracing::debug!(sink_id = self.id, "database pool closed");
+ }
+
+ let state = self.state.lock().await;
+
+ tracing::info!(
+ sink_id = self.id,
+ messages_processed = state.messages_processed,
+ batches_loaded = state.batches_loaded,
+ insertion_errors = state.insertion_errors,
+ "Redshift sink connector closed",
+ );
+
+ Ok(())
+ }
+}
+
+impl RedshiftSink {
+ pub fn new(id: u32, config: RedshiftSinkConfig) -> Self {
+ let verbose = config.verbose_logging.unwrap_or(false);
+
+ Self {
+ id,
+ config,
+ pool: None,
+ state: Mutex::new(State::default()),
+ verbose,
+ bucket: None,
+ }
+ }
+
+ async fn connect(&mut self) -> Result<(), Error> {
+ let max_connections = self
+ .config
+ .max_connections
+ .unwrap_or(DEFAULT_MAX_CONNECTIONS);
+
+ let redacted = redact_connection_string(self.config.connection_string.expose_secret());
+
+ tracing::info!(max_connections, dsn = %redacted, "connecting to Redshift");
+
+ let pool = PgPoolOptions::new()
+ .max_connections(max_connections)
+ .connect(self.config.connection_string.expose_secret())
+ .await
+ .map_err(|e| Error::InitError(format!("Failed to connect to Redshift: {e}")))?;
+
+ sqlx::query("SELECT 1").execute(&pool).await.map_err(|e| {
+ tracing::error!("Tracing failed: {:#?}", e);
+ Error::InitError(format!("Warehouse connectivity test failed: {e}"))
+ })?;
+
+ self.pool = Some(pool);
+ tracing::debug!("Redshift connection pool established");
+
+ let region = self.build_region()?;
+
+ let credentials = Credentials::new(
+ Some(self.config.aws_access_key_id.expose_secret()),
+ Some(self.config.aws_secret_access_key.expose_secret()),
+ None,
+ None,
+ None,
+ )
+ .map_err(|e| {
+ tracing::error!("Failed to create S3 credentials: {e}");
+ Error::InvalidConfig
+ })?;
+
+ let mut bucket = Bucket::new(&self.config.s3_bucket, region, credentials).map_err(|e| {
+ tracing::error!("Failed to create S3 bucket client: {e}");
+ Error::InvalidConfig
+ })?;
+
+ if self.config.s3_endpoint.is_some() {
+ bucket = bucket.with_path_style();
+ }
+
+ self.bucket = Some(bucket);
+
+ tracing::info!("Redshift sink connector ready");
+
+ Ok(())
+ }
+
+ fn build_region(&self) -> Result<Region, Error> {
+ if let Some(endpoint) = &self.config.s3_endpoint {
+ tracing::debug!(endpoint = %endpoint, "using custom S3 endpoint");
+ Ok(Region::Custom {
+ region: self.config.aws_region.clone(),
+ endpoint: endpoint.clone(),
+ })
+ } else {
+ Region::from_str(&self.config.aws_region).map_err(|_| Error::InvalidConfig)
+ }
+ }
+
+ async fn ensure_table_exists(&self) -> Result<(), Error> {
+ let pool = self.get_pool()?;
+
+ let table_name = &self.config.target_table;
+ let payload_type = self.payload_format().sql_type();
+
+ let (query, _) = self.build_create_table_sql()?;
+
+ tracing::debug!("ensuring target table exists");
+
+ sqlx::query(AssertSqlSafe(query))
+ .execute(pool)
+ .await
+ .map_err(|e| {
+ tracing::error!(error = %e);
+ Error::InitError(format!("Failed to create table '{table_name}': {e}"))
+ })?;
+
+ tracing::info!(table = %table_name, payload_type, "target table ready");
+
+ Ok(())
+ }
+
+ async fn process_messages(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<(), Error> {
+ let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+ for batch in messages.chunks(batch_size) {
+ match self
+ .insert_batch(batch, topic_metadata, messages_metadata)
+ .await
+ {
+ Ok(()) => {
+ self.state.lock().await.batches_loaded += 1;
+ }
+ Err(e) => {
+ self.state.lock().await.insertion_errors += batch.len() as u64;
+ tracing::error!(error = %e, batch_size = batch.len(), "failed to insert batch");
+ }
+ }
+ }
+
+ let mut state = self.state.lock().await;
+ state.messages_processed += messages.len() as u64;
+
+ if self.verbose {
+ tracing::info!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ batch_received = messages.len(),
+ table = %self.config.target_table,
+ batches_loaded = state.batches_loaded,
+ "processed message batch"
+ );
+ } else {
+ tracing::debug!(
+ sink_id = self.id,
+ total_processed = state.messages_processed,
+ table = %self.config.target_table,
+ "processed message batch"
+ );
+ }
+
+ Ok(())
+ }
+
+ async fn insert_batch(
+ &self,
+ messages: &[ConsumedMessage],
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ ) -> Result<(), Error> {
+ if messages.is_empty() {
+ return Ok(());
+ }
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_format = self.payload_format();
+
+ let record_batch = create_record_batch(
+ topic_metadata,
+ messages_metadata,
+ messages,
+ include_metadata,
+ include_checksum,
+ include_origin_timestamp,
+ payload_format,
+ )?;
+
+ let content = encode_parquet(&record_batch)?;
+
+ tracing::debug!(
+ bytes = content.len(),
+ rows = record_batch.num_rows(),
+ "encoded parquet batch"
+ );
+
+ let s3_path = self.upload_parquet(&content).await?;
+ self.copy_parquet(&s3_path).await?;
+ self.archive_parquet(&s3_path).await?;
+
+ tracing::info!(count = messages.len(), path = %s3_path, "batch inserted into Redshift");
+
+ Ok(())
+ }
+
+ async fn copy_parquet(&self, s3_path: &str) -> Result<(), Error> {
+ let max_retries = self.get_max_retries();
+ let retry_delay = self.get_retry_delay();
+ let sql = self.build_copy_sql(s3_path);
+ let pool = self.get_pool()?;
+
+ tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+ retry_with_backoff(
+ "redshift COPY",
+ max_retries,
+ retry_delay,
+ is_transient_error,
+ || async {
+ sqlx::query(AssertSqlSafe(sql.as_str()))
+ .execute(pool)
+ .await
+ .map(|_| ())
+ },
+ )
+ .await?;
+
+ tracing::debug!(table = %self.config.target_table, "Redshift COPY completed");
+
+ Ok(())
+ }
+
+ fn build_create_table_sql(&self) -> Result<(String, u32), Error> {
+ let table_name = &self.config.target_table;
+ let quoted_table = quote_identifier(table_name)?;
+
+ let include_metadata = self.config.include_metadata.unwrap_or(true);
+ let include_checksum = self.config.include_checksum.unwrap_or(true);
+ let include_origin_timestamp = self.config.include_origin_timestamp.unwrap_or(true);
+ let payload_type = self.payload_format().sql_type();
+
+ let mut params_per_row: u32 = 1; // id
+
+ let mut query =
+ format!("CREATE TABLE IF NOT EXISTS {quoted_table} (id DECIMAL(39, 0) PRIMARY KEY");
+
+ if include_metadata {
+ query.push_str(", iggy_offset BIGINT, iggy_timestamp TIMESTAMPTZ, iggy_stream TEXT, iggy_topic TEXT, iggy_partition_id INTEGER");
+ params_per_row += 5;
+ }
+
+ if include_checksum {
+ query.push_str(", iggy_checksum VARCHAR");
+ params_per_row += 1;
+ }
+
+ if include_origin_timestamp {
+ query.push_str(", iggy_origin_timestamp TIMESTAMPTZ");
+ params_per_row += 1;
+ }
+
+ query.push_str(&format!(", payload {payload_type}"));
+ query.push_str(", created_at TIMESTAMPTZ DEFAULT GETDATE());");
+ params_per_row += 2;
+
+ Ok((query, params_per_row))
+ }
+
+ fn build_copy_sql(&self, s3_path: &str) -> String {
+ // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here
Review Comment:
@ryerraguntla @hubcio
I've been thinking about this with the proposed design where id is VARCHAR(40), PKs are dropped, and uniqueness is enforced via MERGE from staging into the target.
The sink only emits a small, fixed set of Redshift queries. SELECT 1, CREATE TABLE, COPY, and MERGE are the only statements requiring dynamic column lists, which seems reasonable to validate with the pgwire mock.
The main gaps I see are:
- Malformed Parquet – I don't think this can be meaningfully tested without a real Redshift instance.
- Schema mismatch – This can be validated during open(), as pointed elsewhere in this review. And help clarify why the sibling postgres_sink does not enforce this? My assumption is the auto_create flag transfers the burden to the user to ensure there's no schema mismatch(by means of migrations)
- IAM/S3 permissions – This is more of an user environment/configuration issue than connector behaviour unless we want to check for S3 and Redshift permissions from the iam_role, access_key and secret_key config provide.
Given that, I'm wondering how the smoke test against a real Redshift instance would be to cover the 4 queries semantics.
--
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