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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
