source-postgres: introduce cdc on bulk cdk #70278

mwbayley · 2025-12-01T22:41:33Z

What

Introduce basic CDC functionality for source-postgres on the bulk CDK. Does not include replication slot validation or management.

Can this PR be safely reverted and rolled back?

YES 💚
NO ❌

github-actions · 2025-12-01T22:42:01Z

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

Helpful Resources

Breaking Changes Guide - Breaking changes, migration guides, and upgrade deadlines
Developing Connectors Locally
Managing Connector Secrets
On-Demand Live Tests
On-Demand Regression Tests
#connector-ci-issues
#connector-publish-updates
#connector-build-statuses

PR Slash Commands

Airbyte Maintainers (that's you!) can execute the following slash commands on your PR:

/format-fix - Fixes most formatting issues.
/bump-version - Bumps connector versions.
- You can specify a custom changelog by passing changelog. Example: /bump-version changelog="My cool update"
- Leaving the changelog arg blank will auto-populate the changelog from the PR title.
/run-cat-tests - Runs legacy CAT tests (Connector Acceptance Tests)
/build-connector-images - Builds and publishes a pre-release docker image for the modified connector(s).
JVM connectors:
- /update-connector-cdk-version connector=<CONNECTOR_NAME> - Updates the specified connector to the latest CDK version.
  Example: /update-connector-cdk-version connector=destination-bigquery
- /bump-bulk-cdk-version bump=patch changelog='foo' - Bump the Bulk CDK's version. bump can be major/minor/patch.
Python connectors:
- /poe connector source-example lock - Run the Poe lock task on the source-example connector, committing the results back to the branch.
- /poe source example lock - Alias for /poe connector source-example lock.
- /poe source example use-cdk-branch my/branch - Pin the source-example CDK reference to the branch name specified.
- /poe source example use-cdk-latest - Update the source-example CDK dependency to the latest available version.

📝 Edit this welcome message.

mwbayley · 2025-12-01T22:42:47Z

...main/java/io/airbyte/integrations/source/postgres/cdc/PostgresDebeziumDatetimeConverter.java

This was taken verbatim from the old connector, with some inlining of constants from otherwise unneeded objects and classes. It's a plugin for Debezium at runtime, and my understanding is that it shouldn't need to be changed substantially.

Convert to kotlin - you can do one click machine conversion from the IDE

I assumed that it had to be Java, but Claude thinks Kotlin will work too. I'll give it a try.

The generated code isn't compiling, and it looks like it'll take more than a few minutes to fix. I'm going to leave it as-is for the moment. We can revisit this later, if you'd like.

github-actions · 2025-12-01T22:45:11Z

`source-postgres` Connector Test Results

0 tests ±0 0 ✅ ±0 0s ⏱️ ±0s
0 suites ±0 0 💤 ±0
0 files ±0 0 ❌ ±0

Results for commit a7c2ccd. ± Comparison against base commit cb7cc14.

♻️ This comment has been updated with latest results.

mwbayley · 2025-12-02T18:47:58Z

...ce-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/DateTimeConverter.kt

+            // In snapshot mode, debezium produces a java.sql.Timestamp object for the TIMESTAMPTZ
+            // type. Conceptually, a timestamp with timezone is an Instant. But t.toInstant()
+            // actually mangles the value for ancient dates, because leap years weren't applied
+            // consistently in ye olden days. Additionally, toInstant() (and toLocalDateTime())
+            // actually lose the era indicator, so we can't rely on their getEra() methods.
+            // So we have special handling for this case, which sidesteps the toInstant conversion.


This comes from the previous implementation, including the comment. The bit about snapshot mode confuses me a bit - are there any cases where we let Debezium take snapshots? Or is this a relic of a bygone era?

This code carries a lot of legacy. For example snapshot mode is no longer valid as we took over snapshotting.
Doesn't need to be on first pass, but we might want to clean up this file

TODO re old comment on types in dbz snapshot

mwbayley · 2025-12-02T18:51:11Z

...ce-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/DateTimeConverter.kt

This also comes from the old implementation. The only addition here is the @JvmField val ... bits from line 40-49. These came from another constants class which was otherwise unused.

rodireich · 2025-12-02T17:08:58Z

...main/java/io/airbyte/integrations/source/postgres/cdc/PostgresDebeziumDatetimeConverter.java

Convert to kotlin - you can do one click machine conversion from the IDE

rodireich · 2025-12-03T17:03:39Z

...ce-postgres/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/DateTimeConverter.kt

+            // In snapshot mode, debezium produces a java.sql.Timestamp object for the TIMESTAMPTZ
+            // type. Conceptually, a timestamp with timezone is an Instant. But t.toInstant()
+            // actually mangles the value for ancient dates, because leap years weren't applied
+            // consistently in ye olden days. Additionally, toInstant() (and toLocalDateTime())
+            // actually lose the era indicator, so we can't rely on their getEra() methods.
+            // So we have special handling for this case, which sidesteps the toInstant conversion.


This code carries a lot of legacy. For example snapshot mode is no longer valid as we took over snapshotting.
Doesn't need to be on first pass, but we might want to clean up this file

rodireich · 2025-12-03T17:05:42Z

...res/src/main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceCdcPosition.kt

+    val lsn: Lsn,
+    val lsnProc: Lsn,
+) : Comparable<PostgresSourceCdcPosition> {
+    override fun compareTo(other: PostgresSourceCdcPosition): Int =


Is this the equivalent of what we do in the legacy connector?
#35939

fix LSN handling

.../main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceDebeziumOperations.kt

rodireich · 2025-12-03T17:08:27Z

.../main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceDebeziumOperations.kt

+        val txId: Long
+        val lsn: Long
+        // TODO: Does it matter if we take timestamp before or after fetching lsn and txid?
+        val time: Instant = Instant.now()


Maybe that's what you refer here in the // TODO above:
Since startState is lazy the time will only be set when lazily wired. is that expected?

Yeah, that's expected. I meant within this block. I don't think this is the case, but if Debezium filtered out records with a lower timestamp than the state, for example, we could lose records if we took the timestamp after the current LSN. They're not in the same transaction, you know? Now that I think about it, we should probably take the timestamp from the DB rather than the application. We weren't doing that previously, though.

rodireich · 2025-12-03T17:09:50Z

.../main/kotlin/io/airbyte/integrations/source/postgres/cdc/PostgresSourceDebeziumOperations.kt

+        val resultRow: NativeRecordPayload = mutableMapOf()
+        for (field in stream.schema) {
+            data[field.id] ?: continue
+            when (data[field.id]) {


Make sure types received from debezium are working as expected.
There were some rough edged here that needed minor conversions in mysql CDC

This is probably easier handled with an integration test. We can use all the same test cases we already defined for the JDBC type support. I filed a sub-issue for this: https://github.com/airbytehq/airbyte-internal-issues/issues/13822?issue=airbytehq%7Cairbyte-internal-issues%7C15470

rodireich · 2025-12-03T17:10:49Z

...rc/main/kotlin/io/airbyte/integrations/source/postgres/config/PostgresSourceConfiguration.kt

@@ -93,6 +95,13 @@ data class CdcIncrementalConfiguration(
    val initialLoadTimeout: Duration,
    val invalidCdcCursorPositionBehavior: InvalidCdcCursorPositionBehavior,
    val shutdownTimeout: Duration,
+    val replicationSlot: String,
+    val publication: String,


publication is optional I think in legacy
String? ?

Looks like it is/was required when using CDC:

Debezium has a default value (dbz_publication), but we don't use it. Our docs describe it as required, as well.

...n/io/airbyte/integrations/source/postgres/config/PostgresSourceConfigurationSpecification.kt

rodireich · 2025-12-03T17:15:02Z

...n/io/airbyte/integrations/source/postgres/config/PostgresSourceConfigurationSpecification.kt

+        "A Postgres publication used for consuming changes. Read about <a href=\\\"https://docs.airbyte.com/integrations/sources/postgres#step-4-create-publications-and-replication-identities-for-tables\\\">publications and replication identities</a>."
+    )
+    @JsonSchemaInject(json = """{"order":5,"always_show":true, "minLength":1}""")
+    lateinit var publication: String


I don't think publication is mandatory.
It's a logical filter on top of a replication slot. No publication is equivalent to all tables publication.

A publication is necessary for CDC. That can be for all tables, or a subset. Debezium has the feature to provision one for you if it doesn't exist, but this requires superuser permissions, so most users provision it themselves. We don't use it either.

…cdc-on-bulk-cdk

octavia-squidington-iii added the connectors/source/postgres label Dec 1, 2025

mwbayley commented Dec 1, 2025

View reviewed changes

github-actions bot added the dependencies-change label Dec 1, 2025

mwbayley force-pushed the mbayley/source-postgres/cdc-on-bulk-cdk branch 3 times, most recently from 2fb6bba to b0ba5d3 Compare December 1, 2025 23:38

basic CDC functionality for Postgres

c20d070

mwbayley force-pushed the mbayley/source-postgres/cdc-on-bulk-cdk branch from b0ba5d3 to c20d070 Compare December 1, 2025 23:40

github-actions bot removed the dependencies-change label Dec 1, 2025

mwbayley commented Dec 2, 2025

View reviewed changes

mwbayley added 5 commits December 2, 2025 10:56

fix state json example comments

4698eea

clean up dbz ops

be05d98

fix missing cdc fields

885ef75

suppress unchecked casts

574f47c

gitignore LocalRunners

b1d727f

rodireich approved these changes Dec 3, 2025

View reviewed changes

mwbayley added 4 commits December 3, 2025 14:11

fix LSN handling

af6f0f5

update comments

b7e6f1e

remove resolved comment

e91f1b7

TODO re old comment on types in dbz snapshot

a7c2ccd

mwbayley marked this pull request as ready for review December 3, 2025 23:48

Merge branch 'source-postgres/bulk-cdk' into mbayley/source-postgres/…

bd506c2

…cdc-on-bulk-cdk

mwbayley merged commit fd5bfb5 into source-postgres/bulk-cdk Dec 3, 2025
25 of 29 checks passed

mwbayley deleted the mbayley/source-postgres/cdc-on-bulk-cdk branch December 3, 2025 23:54

source-postgres: introduce cdc on bulk cdk #70278

source-postgres: introduce cdc on bulk cdk #70278

Uh oh!

Conversation

mwbayley commented Dec 1, 2025

What

Can this PR be safely reverted and rolled back?

Uh oh!

github-actions bot commented Dec 1, 2025

👋 Greetings, Airbyte Team Member!

Helpful Resources

PR Slash Commands

Uh oh!

mwbayley Dec 1, 2025 • 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!

github-actions bot commented Dec 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

source-postgres Connector Test Results

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mwbayley Dec 3, 2025 • 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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mwbayley Dec 1, 2025 •

edited

Loading

github-actions bot commented Dec 1, 2025 •

edited

Loading

`source-postgres` Connector Test Results

mwbayley Dec 3, 2025 •

edited

Loading