feat(connectors): add Redshift sink connector

Cargo.toml

@@ -43,6 +43,7 @@ members = [
     "core/connectors/sinks/mongodb_sink",
     "core/connectors/sinks/postgres_sink",
     "core/connectors/sinks/quickwit_sink",
+    "core/connectors/sinks/redshift_sink",
     "core/connectors/sinks/s3_sink",
     "core/connectors/sinks/stdout_sink",
     "core/connectors/sinks/surrealdb_sink",
@@ -247,6 +248,7 @@ parquet = "57.3.1"
 partitions = { path = "core/partitions" }
 passterm = "2.0.6"
 paste = "1.0"
+pgwire = "0.40.4"
 postcard = { version = "1.1.3", features = ["alloc"] }
 predicates = "3.1.4"
 proc-macro2 = "1"
@@ -293,6 +295,7 @@ simd-json = { version = "0.17.0", features = ["serde_impl"] }
 slab = "0.4.12"
 smallvec = "1.15"
 socket2 = "0.6.4"
+sqlparser = "0.62.0"
 sqlx = { version = "0.9.0", features = [
     "runtime-tokio",
     "tls-rustls",
@@ -318,6 +321,7 @@ testcontainers = { version = "0.27.3", features = ["reusable-containers"] }
 testcontainers-modules = { version = "0.15.0", features = ["postgres", "http_wait"] }
 thiserror = "2.0.18"
 tokio = { version = "1.52.3", features = ["full"] }
+tokio-postgres = "0.7.18"
 tokio-rustls = "0.26.4"
 tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] }
 tokio-util = { version = "0.7.18", features = ["compat"] }

core/connectors/README.md

@@ -85,6 +85,7 @@ Each sink should have its own, custom configuration, which is passed along with
 - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
 - **PostgreSQL Sink** - stores messages in PostgreSQL database tables
 - **Quickwit Sink** - indexes messages in Quickwit search engine
+- **Reshift Sink** - stores messages in Redshift warehouse tables via S3 as staging
 - **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces)
 - **Stdout Sink** - prints messages to standard output (useful for debugging/development)
 - **SurrealDB Sink** - writes messages into SurrealDB with deterministic record IDs for idempotent replay

core/connectors/sinks/README.md

@@ -14,6 +14,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | **influxdb_sink** | Writes messages to InfluxDB as line-protocol points; supports both V2 (org/bucket, Flux) and V3 (db, SQL) |
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
 | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics |
+| **redshift_sink** | Stores messages in Redshift warehouse tables with configurable schemas vis S3 as staging |
 | **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) |
 | **stdout_sink** | Prints messages to standard output (useful for debugging and development) |
 | **surrealdb_sink** | Writes messages into SurrealDB with deterministic record IDs for idempotent replay |

core/connectors/sinks/redshift_sink/Cargo.toml

@@ -0,0 +1,52 @@
+# 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.
+
+[package]
+name = "iggy_connector_redshift_sink"
+version = "0.4.1-edge.1"
+description = "Iggy Redshift sink connector for storing stream messages into Redshift warehouse via S3"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "redshift", "sink"]
+categories = ["command-line-utilities", "warehouse", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+arrow = { workspace = true }
+async-trait = { workspace = true }
+humantime = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+parquet = { workspace = true }
+rust-s3 = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sqlx = { workspace = true, features = ["runtime-tokio", "tls-rustls", "postgres", "chrono"] }
+tokio = { workspace = true }
+tracing = { workspace = true }
+uuid = { workspace = true, features = ["v7"] }
+
+[dev-dependencies]
+simd-json = { workspace = true }

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. |
+| `include_checksum` | no | `false` | Stores the Iggy message checksum. |
+| `include_origin_timestamp` | no | `false` | Stores the original Iggy origin timestamp. |
+| `payload_format` | no | `varbyte` | Encoding used for the payload column in the Parquet file. See **Payload Format** below. |
+| `verbose_logging` | no | `false` | Enables verbose logging for debugging purposes. |
+| `max_retries` | no | `3` | Maximum number of retries for failed `COPY` operations. `0` disables retries (only one attempt will be made) |
+| `retry_delay` | no | `1s` | Delay in seconds between retry attempts. |
+| `aws_access_key_id` | yes | — | AWS access key used for S3 staging. |
+| `aws_secret_access_key` | yes | — | AWS secret key used for S3 staging. |
+| `s3_bucket` | yes | — | S3 bucket that Parquet batch files are staged into before the Redshift `COPY`. |
+| `s3_prefix` | yes | — | Key prefix under which staged Parquet files are written, e.g. `iggy/messages`. |
+| `s3_endpoint` | no | — | Override endpoint for S3-compatible stores (e.g. MinIO). Omit for AWS S3 itself. |
+| `aws_region` | yes | — | AWS region for the S3 bucket. |
+| `archive` | no | `false` | See **Archiving Staged Files** below. |
+
+## Staging via S3
+
+Redshift's `COPY` command loads from files, not from a live stream, so every
+batch is first written out as a Parquet file and uploaded to
+`s3://<s3_bucket>/<s3_prefix>/...` before the `COPY` into `target_table` runs.
+S3 is purely a staging area in this flow — it is not queried directly by
+consumers of the data, and its cost is the price of getting bulk data into
+Redshift efficiently rather than row-by-row.
+
+## Archiving Staged Files
+
+The `archive` field controls what happens to a batch's Parquet file **after**
+it has been successfully loaded into Redshift:
+
+- `archive = true` — the file is kept, moved under an `archive` prefix
+  (i.e. `s3://<s3_bucket>/archive/...`) instead of being deleted.
+  Useful for replay, auditing, or downstream batch jobs that read Parquet
+  directly.
+- `archive = false` — the file is deleted from S3 once the `COPY` succeeds,
+  since Redshift itself is now the source of truth for that data and the
+  staged copy has no further purpose.
+
+## Payload Format
+
+`payload_format` controls how the payload column is written in the staged
+Parquet file, which in turn determines its type once loaded into Redshift:
+
+- Parquet has no dedicated JSON logical type, so a `payload_format = "json"`
+  payload is written as a Parquet `VARCHAR` (string), not a structured type.
+- As a result, the column lands in Redshift as `VARCHAR`, not `SUPER`.
+- To query the payload as structured data downstream, use Redshift's
+  `JSON_PARSE()` (or equivalent JSON functions) on the `VARCHAR` column at
+  query time rather than expecting a native `SUPER` column out of the box.
+
+## Stored Shape
+
+With metadata enabled, records contain:
+
+- `id`: original Iggy message id as numeric
+- `iggy_stream`, `iggy_topic`, `iggy_partition_id`, `iggy_offset`
+- `iggy_timestamp`, `iggy_origin_timestamp`, `iggy_checksum`,
+- `payload`: encoded per `payload_format` (see above)
+
+The `messages_processed` counter reports valid records submitted to Redshift
+via `COPY`.

core/connectors/sinks/redshift_sink/config.toml

@@ -0,0 +1,49 @@
+# 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.
+
+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

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)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum PayloadFormat {
+    Json,
+    Text,
+    #[default]
+    Varbyte,
+}
+
+impl PayloadFormat {
+    pub fn from_config(s: Option<&str>) -> Self {
+        match s.map(|s| s.to_lowercase()).as_deref() {
+            Some("text") | Some("json") => PayloadFormat::Text,
+            _ => PayloadFormat::Varbyte,
+        }
+    }
+
+    pub fn sql_type(&self) -> &'static str {
+        match self {
+            PayloadFormat::Varbyte => "VARBYTE",
+            PayloadFormat::Text | PayloadFormat::Json => "VARCHAR",
+        }
+    }
+
+    pub fn arrow_type(&self) -> DataType {
+        match self {
+            PayloadFormat::Varbyte => DataType::Binary,
+            PayloadFormat::Text => DataType::Utf8,
+            PayloadFormat::Json => DataType::Utf8,
+        }
+    }
+}