[source-postgres] CTID cursor resets to (0,0) when incremental sync returns 0 records
- Lenguaje dominante
- Python
- Estrellas
- 22.1k
- Forks
- 5.4k
- Merge medio
- 5 h
- PR fusionados (30 d)
- 671
Descripción
### Connector Name
source-postgres
### Connector Version
3.7.0
### What step the error happened?
During the sync
### Relevant information
## Summary
When using CTID-based incremental sync (standard cursor mode, not CDC), if a sync returns 0 records for a stream, the cursor state is saved as `(0,0)` instead of preserving the previous cursor position. This causes subsequent syncs to perform a full table re-read for those streams.
## Source Connector
source-postgres
## Sync Mode
Incremental Append (standard cursor / CTID-based, **not** CDC)
## Destination
Kafka
## Steps to Reproduce
1. Set up a Postgres source using Xmin mode with multiple tables
2. Configure connection with Full Refresh Append mode
3. Run initial sync - all data syncs successfully
4. Switch all streams to Incremental Append mode (standard cursor)
5. Modify rows in **some** tables (not all)
6. Run first incremental sync - works correctly, only changed rows sync
7. Modify rows in a **different subset** of tables
8. Run second incremental sync - **tables that had 0 records in step 6 do a full resync**
## Expected Behavior
Tables with 0 new records should preserve their cursor position and continue returning 0 records on subsequent syncs.
## Actual Behavior
Tables with 0 new records have their CTID cursor reset to `(0,0)`, causing a full table scan on the next sync.
## Root Cause Analysis
The bug is in `CtidStateManager.java` in the `generateCtidStatusForState` method:
```java
protected CtidStatus generateCtidStatusForState(final AirbyteStreamNameNamespacePair pair) {
final Long fileNode = fileNodeHandler.getFileNode(pair);
assert fileNode != null;
final String lastCtid = pairToLastCtid.get(pair);
// If the table is empty, lastCtid will be set to zero for the final state message.
final String lastCtidInState = (Objects.nonNull(lastCtid)
&& StringUtils.isNotBlank(lastCtid)) ? lastCtid : Ctid.ZERO.toString(); // <-- BUG HERE
return new CtidStatus()
.withVersion(CTID_STATUS_VERSION)
.withStateType(StateType.CTID)
.withCtid(lastCtidInState)
...
}
```
**The Problem:**
1. `pairToLastCtid` is only populated in `processRecordMessage()` when records are processed
2. When a stream has 0 records, `processRecordMessage()` is never called
3. `pairToLastCtid.get(pair)` returns `null`
4. The code falls back to `Ctid.ZERO.toString()` which is `"(0,0)"`
5. State is saved with cursor `(0,0)`
6. Next sync reads from the beginning of the table
**The Fix:**
When no records are processed, the code should fall back to the **previous cursor position** from `pairToCtidStatus` (which contains the state from the previous sync), not `Ctid.ZERO`.
## Proposed Fix
```java
final String lastCtidInState;
if (Objects.nonNull(lastCtid) && StringUtils.isNotBlank(lastCtid)) {
// Use the CTID from records processed in this sync
lastCtidInState = lastCtid;
} else {
// No records processed - preserve the previous cursor position if available
final CtidStatus previousStatus = pairToCtidStatus.get(pair);
if (previousStatus != null && StringUtils.isNotBlank(previousStatus.getCtid())) {
lastCtidInState = previousStatus.getCtid();
} else {
// Truly empty table or first sync - use zero
lastCtidInState = Ctid.ZERO.toString();
}
}
```
## Impact
This bug affects all users using CTID-based incremental sync (non-CDC mode) with Postgres. It causes:
- Unexpected full table resyncs
- Increased load on source databases
- Increased data transfer to destinations
- Potential duplicate data in append-mode destinations
### Relevant log output
```shell
=== FIRST INCREMENTAL SYNC (correct behavior) ===
2026-02-06 18:19:11 Executing query for table orders: SELECT ... FROM "cdc_testing"."orders" WHERE ctid > ?::tid with binding (0,18)
2026-02-06 18:19:11 sending final state message, with count per stream: {cdc_testing_orders=1}
2026-02-06 18:19:11 Executing query for table categories: SELECT ... FROM "cdc_testing"."categories" WHERE ctid > ?::tid with binding (0,5)
2026-02-06 18:19:11 sending final state message, with count per stream: {}
2026-02-06 18:19:11 Executing query for table inventory: SELECT ... FROM "cdc_testing"."inventory" WHERE ctid > ?::tid with binding (0,13)
2026-02-06 18:19:11 sending final state message, with count per stream: {cdc_testing_inventory=1}
2026-02-06 18:19:11 Executing query for table products: SELECT ... FROM "cdc_testing"."products" WHERE ctid > ?::tid with binding (0,11)
2026-02-06 18:19:11 sending final state message, with count per stream: {}
2026-02-06 18:19:11 Executing query for table customers: SELECT ... FROM "cdc_testing"."customers" WHERE ctid > ?::tid with binding (0,13)
2026-02-06 18:19:11 sending final state message, with count per stream: {cdc_testing_customers=1}
2026-02-06 18:19:11 Executing query for table information: SELECT ... FROM "cdc_testing"."information" WHERE ctid > ?::tid with binding (0,51)
2026-02-06 18:19:11 sending final state message, with count per stream: {}
=== SECOND INCREMENTAL SYNC (bug - cursors reset to 0,0) ===
2026-02-06 18:19:44 Executing query for table orders: SELECT ... FROM "cdc_testing"."orders" WHERE ctid > ?::tid with binding (0,19)
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_orders=1}
2026-02-06 18:19:44 Executing query for table categories: SELECT ... FROM "cdc_testing"."categories" WHERE ctid > ?::tid with binding (0,0) <-- RESET!
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_categories=5} <-- FULL RESYNC
2026-02-06 18:19:44 Executing query for table inventory: SELECT ... FROM "cdc_testing"."inventory" WHERE ctid > ?::tid with binding (0,14)
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_inventory=2}
2026-02-06 18:19:44 Executing query for table products: SELECT ... FROM "cdc_testing"."products" WHERE ctid > ?::tid with binding (0,0) <-- RESET!
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_products=11} <-- FULL RESYNC
2026-02-06 18:19:44 Executing query for table customers: SELECT ... FROM "cdc_testing"."customers" WHERE ctid > ?::tid with binding (0,14)
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_customers=1}
2026-02-06 18:19:44 Executing query for table information: SELECT ... FROM "cdc_testing"."information" WHERE ctid > ?::tid with binding (0,0) <-- RESET!
2026-02-06 18:19:44 sending final state message, with count per stream: {cdc_testing_information=7} <-- FULL RESYNC
=== PATTERN ===
Tables with 0 records in first sync (categories, products, information) had cursor reset to (0,0) in second sync.
Tables with >0 records in first sync (orders, inventory, customers) progressed correctly.
```
### Contribute
- [x] Yes, I want to contribute
---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/11202
Guía de contribución
Evaluación
Este issue todavía no se ha evaluado.