feat(connectors): add Redshift sink connector by realonbebeto · Pull Request #3654 · apache/iggy

realonbebeto · 2026-07-12T04:44:00Z

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

pgwire, sqlparser, and tokio-postgres are introduced as workspace dependencies solely to support the RedshiftFixture used by the integration test suite.
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

github-actions · 2026-07-12T04:44:13Z

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 for details.

codecov · 2026-07-14T21:12:48Z

Codecov Report

❌ Patch coverage is 84.07960% with 128 lines in your changes missing coverage. Please review.
✅ Project coverage is 40.38%. Comparing base (6f1d548) to head (6ec283d).

Files with missing lines	Patch %	Lines
core/connectors/sinks/redshift_sink/src/lib.rs	83.39%	84 Missing and 43 partials ⚠️
core/connectors/sinks/redshift_sink/src/config.rs	97.43%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3654       +/-   ##
=============================================
- Coverage     73.99%   40.38%   -33.61%     
  Complexity      937      937               
=============================================
  Files          1301     1301               
  Lines        147393   129167    -18226     
  Branches     122945   104795    -18150     
=============================================
- Hits         109063    52170    -56893     
- Misses        34858    74142    +39284     
+ Partials       3472     2855      -617

Components	Coverage Δ
Rust Core	`32.56% <84.07%> (-41.74%)`	⬇️
Java SDK	`62.44% <ø> (ø)`
C# SDK	`71.04% <ø> (-1.14%)`	⬇️
Python SDK	`92.17% <ø> (ø)`
PHP SDK	`84.29% <ø> (ø)`
Node SDK	`91.35% <ø> (+0.09%)`	⬆️
Go SDK	`42.87% <ø> (ø)`

Files with missing lines	Coverage Δ
core/connectors/sinks/redshift_sink/src/config.rs	`97.43% <97.43%> (ø)`
core/connectors/sinks/redshift_sink/src/lib.rs	`83.39% <83.39%> (ø)`

... and 444 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

krishvishal · 2026-07-15T03:50:17Z

@realonbebeto can you please fix the CI failures? After that we can review the code.

realonbebeto · 2026-07-15T04:20:03Z

@hubcio resolved the CI failures.

slbotbm · 2026-07-16T07:54:00Z

@realonbebeto looks like your lockfile is stale, and that is why a lot of the tests are failing

realonbebeto · 2026-07-16T14:18:54Z

@slbotbm I made a cargo update on the PR. I think this resolves the staleness.

slbotbm · 2026-07-16T14:38:59Z

@realonbebeto I can see that changes unrelated to the pr have also been committed. can you correct this please?

realonbebeto · 2026-07-16T16:27:13Z

@slbotbm made the unrelated changes correction

krishvishal · 2026-07-16T19:39:56Z

@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:

iggy/CONTRIBUTING.md

Line 80 in c0218f1

We use [prek](https://github.com/j178/prek):

realonbebeto · 2026-07-16T21:47:28Z

@slbotbm @krishvishal prek passed by allowing the commit. Thanks for the hint.

hubcio · 2026-07-17T11:15:45Z

what is reason for so many changes in this file?

I did a cargo update to resolve staleness. This is the context: #3654 (comment)

This PR introduces the Redshift sink connector. P.S. a forced redo and push of an earlier commit on the same branch to resolve a divergence that was affecting cli tests(help command)

ryerraguntla · 2026-07-19T22:40:32Z

+        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");


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

ryerraguntla · 2026-07-19T22:43:04Z

+        Ok((query, params_per_row))
+    }
+
+    fn build_copy_sql(&self, s3_path: &str) -> String {


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.

ryerraguntla · 2026-07-19T22:44:23Z

+    ) -> Result<(), Error> {
+        let batch_size = self.config.batch_size.unwrap_or(100) as usize;
+
+        for batch in messages.chunks(batch_size) {


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.

ryerraguntla · 2026-07-19T22:46:58Z

+
+        tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");
+
+        retry_with_backoff(


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.

ryerraguntla · 2026-07-19T22:48:29Z

+    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!(


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.

ryerraguntla · 2026-07-19T22:51:36Z

+    }
+
+    fn build_copy_sql(&self, s3_path: &str) -> String {
+        // Built via format! (not sqlx binds) because the Redshift/Pgwire endpoint here


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."

ryerraguntla · 2026-07-19T22:52:27Z

+            self.config.aws_secret_access_key.expose_secret()
+        );
+
+        format!(


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.

ryerraguntla · 2026-07-19T22:53:13Z

+            "encoded parquet batch"
+        );
+
+        let s3_path = self.upload_parquet(&content).await?;


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.

ryerraguntla · 2026-07-19T22:54:16Z

+            rows = record_batch.num_rows(),
+            "encoded parquet batch"
+        );
+


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.

ryerraguntla · 2026-07-19T22:55:19Z

+            .unwrap_or(rest.len())
+            .min(PREVIEW_LEN);
+
+        let preview = &rest[..safe_end];


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).

ryerraguntla · 2026-07-19T22:56:25Z

+        );
+
+        self.connect().await?;
+        self.ensure_table_exists().await?;


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.

ryerraguntla · 2026-07-19T22:57:07Z

+| `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. |


README.md:60-62 — defaults say false; code defaults true (lib.rs:271-273,334-336). Fix: table → true.

ryerraguntla · 2026-07-19T22:58:02Z

+/// 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)]


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.

ryerraguntla · 2026-07-19T23:01:57Z

+
+fn payload_column(messages: &[ConsumedMessage], format: PayloadFormat) -> Result<ArrayRef, Error> {
+    match format {
+        PayloadFormat::Varbyte => {


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.

ryerraguntla · 2026-07-19T23:12:59Z

Bunch of nits, but needs to be looked into and fixed

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) ·
dead params_per_row return (:330-363) ·
invalid crates.io category "warehouse" → "database" (Cargo.toml:25) ·
const ICEBERG_SINK_KEY = "redshift" copy-paste (int.rs:36) ·
"IcebergFixture" error strings + "bytea" fixture doc mislabels (fixtures sink.rs:393,429,198,245,292) ·
typos "Reshift"/"vis S3"/"Reshshift"/"We dont" ·
payload_column Vec churn + uncompressed parquet writer

Did you check "typos" as part of precheck?

Good effort. Thanks for chiming in to contribute.

ryerraguntla · 2026-07-19T23:13:10Z

/author

realonbebeto · 2026-07-21T16:08:06Z

Bunch of nits, but needs to be looked into and fixed

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) ·

dead params_per_row return (:330-363) ·

invalid crates.io category "warehouse" → "database" (Cargo.toml:25) ·

const ICEBERG_SINK_KEY = "redshift" copy-paste (int.rs:36) ·

"IcebergFixture" error strings + "bytea" fixture doc mislabels (fixtures sink.rs:393,429,198,245,292) ·

typos "Reshift"/"vis S3"/"Reshshift"/"We dont" ·

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'.

github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Jul 12, 2026

realonbebeto force-pushed the feat/redshift-sink-connector branch from 3ea9216 to efa4ffb Compare July 16, 2026 14:54

hubcio reviewed Jul 17, 2026

View reviewed changes

feat(connectors): add Redshift sink connector

6ec283d

This PR introduces the Redshift sink connector. P.S. a forced redo and push of an earlier commit on the same branch to resolve a divergence that was affecting cli tests(help command)

realonbebeto force-pushed the feat/redshift-sink-connector branch from 9c1f696 to 6ec283d Compare July 18, 2026 02:15

ryerraguntla reviewed Jul 19, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Jul 19, 2026


		tracing::debug!(table = %self.config.target_table, s3_path, "issuing Redshift COPY");

		retry_with_backoff(

Uh oh!

Conversation

realonbebeto commented Jul 12, 2026

Summary

Disclosures

AI Usage

Uh oh!

github-actions Bot commented Jul 12, 2026

Uh oh!

codecov Bot commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

krishvishal commented Jul 15, 2026

Uh oh!

realonbebeto commented Jul 15, 2026

Uh oh!

slbotbm commented Jul 16, 2026

Uh oh!

realonbebeto commented Jul 16, 2026

Uh oh!

slbotbm commented Jul 16, 2026

Uh oh!

realonbebeto commented Jul 16, 2026

Uh oh!

krishvishal commented Jul 16, 2026

Uh oh!

realonbebeto commented Jul 16, 2026

Uh oh!

hubcio Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryerraguntla commented Jul 19, 2026

Uh oh!

ryerraguntla commented Jul 19, 2026

Uh oh!

realonbebeto commented Jul 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

codecov Bot commented Jul 14, 2026 •

edited

Loading

hubcio Jul 17, 2026 •

edited

Loading