airbytehq / airbytehq/airbyte

[source-postgres] CTID cursor resets to (0,0) when incremental sync returns 0 records

Abierto
#72,936 3 comentarios 0 reacciones 0 asignados Ver en GitHub
area/connectors autoteam community connectors/source/postgres needs-triage team/extensibility type/bug
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

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.