[Feature][CDC] Expose binlog file/position/row as metadata fields for deterministic downstream deduplication

[ricky2129]

Description

Currently the Metadata transform exposes EventTime and Delay for CDC sources.
For downstream deduplication (e.g. CDC → S3 Parquet → StarRocks/Athena), users need
a monotonically increasing, per-row unique key to identify the latest state of a record.

EventTime is insufficient for this purpose because:

1. Multiple CDC events for the same primary key can share the same millisecond timestamp
   when a MySQL statement affects multiple rows or two statements execute within 1ms
2. c_event_time - c_delay (a common pattern) collapses ordering further through
   arithmetic coincidence — different delay values can produce identical results

Example from production:

{"c_operation_type": "UPDATE_AFTER", "c_event_time": 1774342441039, "c_delay": 39, "c_db_change_time": 1774342441000}
{"c_operation_type": "UPDATE_AFTER", "c_event_time": 1774342441040, "c_delay": 40, "c_db_change_time": 1774342441000}

Both rows produce the same c_db_change_time despite being distinct CDC events.

The correct solution — is to expose the binlog filename + position + row index, which form a strict total order guaranteed by MySQL's binary log protocol. These are already present in the Debezium SourceRecord source struct (file, pos, row) but are never surfaced to the SeaTunnel metadata layer.

Usage Scenario :

Any CDC pipeline that writes to a data lake (S3 Parquet, Iceberg) and needs to deduplicate or find the latest state of a record in downstream queries:

Metadata {
metadata_fields {
BinlogFile = c_binlog_file // e.g. "mysql-bin.000123"
BinlogPos = c_binlog_pos // e.g. 4857392 (BIGINT)
BinlogRow = c_binlog_row // e.g. 0 (INT, row index within event)
}
}

Downstream dedup key in StarRocks/Athena:
CONCAT(
COALESCE(c_binlog_file, ''), ':',
LPAD(CAST(COALESCE(c_binlog_pos, 0) AS STRING), 20, '0'), ':',
LPAD(CAST(COALESCE(c_binlog_row, 0) AS STRING), 10, '0')
) AS c_change_seq

[ricky2129]

@davidzollo can you help?? need your POV on this

[dybyte]

This looks good to me overall.
It might be helpful to have a design doc before submitting a PR, so we can align on the scope and approach.

[ricky2129]

@dybyte

Design Doc: [Feature][CDC] Expose binlog file/position/row as metadata fields for deterministic downstream deduplication

---

Problem Statement

The CDC → data lake pattern (MySQL-CDC → S3 Parquet / Iceberg / Delta) is one of the most common SeaTunnel deployment topologies. In this pattern the sink is append-only: every INSERT, UPDATE_AFTER, and DELETE from the source lands as a new row in the lake. Downstream queries reconstruct the latest state of each record using a deduplication query:

SELECT * FROM cdc_table
QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY <ordering_key> DESC) = 1

The ordering key is the crux of the problem. It must satisfy two properties:

1. Uniqueness — no two distinct CDC events can have the same value
2. Total order — the value must reflect the true sequence of changes in the source database

The SeaTunnel Metadata transform (v2.3.12) exposes the following fields for CDC sources:

Field	Type	Description
Database	STRING	Database name containing the row
Table	STRING	Table name containing the row
RowKind	STRING	Operation type (INSERT, UPDATE_BEFORE, etc.)
EventTime	BIGINT	Time at which the connector processed the event (ms)
Delay	BIGINT	Difference between connector processing time and database change time (ms)
Partition	STRING	Partition field information

None of these fields is suitable as a unique ordering key for deduplication.

---

Why existing fields fail as an ordering key

Failure mode 1: Same-statement multi-row events

A single UPDATE statement matching N rows produces N CDC events, all with identical EventTime — they are all processed by the connector at the same instant. There is no way to distinguish their order using timestamp alone.

-- One statement, two rows touched → two CDC events, same EventTime
UPDATE payments SET status = 'settled' WHERE batch_id = 42;

After deduplication on EventTime, one row is dropped arbitrarily. The lake has an inconsistent view.

Failure mode 2: Sub-millisecond transactions

Two separate transactions committing within the same millisecond share the same EventTime. High-throughput OLTP workloads (payment processing, order management) routinely produce thousands of transactions per second — sub-millisecond commit intervals are not edge cases.

Failure mode 3: The c_db_change_time collapse

A common pattern in SeaTunnel CDC jobs is to approximate the source database change time by subtracting lag:

(c_event_time - c_delay) AS c_db_change_time

This can make the collision worse. Two events with slightly different EventTime and Delay values produce identical c_db_change_time:

{"c_event_time": 1774342441039, "c_delay": 39, "c_db_change_time": 1774342441000}
{"c_event_time": 1774342441040, "c_delay": 40, "c_db_change_time": 1774342441000}

These are two distinct committed changes. They are indistinguishable in the lake.

Failure mode 4: UPDATE_BEFORE and UPDATE_AFTER have the same EventTime

For every UPDATE, SeaTunnel emits two rows: UPDATE_BEFORE and UPDATE_AFTER. Both carry the same EventTime because they originate from the same Debezium SourceRecord. A deduplication query using only timestamp cannot determine which one represents the final state — it can and does pick UPDATE_BEFORE as the current row.

---

The correct solution: MySQL binlog coordinates

MySQL's binary log protocol (row-based replication, required for CDC) provides an exact, stable, globally ordered position for every committed row change.

For a committed transaction, the binlog guarantees:

• Each DML statement produces a separate ROWS_EVENT at a unique byte offset (pos) in the binlog file (file)
• If one statement matches multiple rows, those rows share the same pos but each gets a distinct row index (0, 1, 2…)

Therefore (file, pos, row) is unique per committed row-change, across all cases:

Scenario	pos	row
Two UPDATE statements, 1 row each	different	both 0
One UPDATE matching 2 rows	same	0 and 1
Two UPDATEs on the same row in one transaction	different	both 0
UPDATE_BEFORE vs UPDATE_AFTER	same	same — both from the same Debezium SourceRecord

The last row is important: UPDATE_BEFORE and UPDATE_AFTER intentionally share (file, pos, row). The downstream query should retain UPDATE_AFTER using RowKind as a tiebreaker — not a timestamp coin flip.

These three values are already present in every Debezium SourceRecord for MySQL-CDC, inside the source struct:

source=Struct{
  version=1.9.8.Final,
  connector=mysql,
  file=mysql-bin-changelog.309222,
  pos=54106803,
  row=0,
  ...
}

They are not surfaced to SeaTunnel's metadata layer today. This proposal adds them as opt-in fields in the Metadata transform with a minimal, fully backward-compatible change.

Non-MySQL connectors (PostgreSQL, Oracle, SQL Server) do not have a file field in their Debezium source struct. All three extractors guard on source.schema().field("file") == null, making the fields null for non-MySQL sources with no behavioral change to those connectors.

---

Design

Files changed

File	Change
seatunnel-api/.../CommonOptions.java	Declare BINLOG_FILE, BINLOG_POS, BINLOG_ROW enum values
seatunnel-api/.../MetadataUtil.java	Add setBinlogFile, setBinlogPos, setBinlogRow setters
.../cdc/base/utils/SourceRecordUtils.java	Add getBinlogFile, getBinlogPos, getBinlogRow extractors
.../debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java	Populate fields on every emitted SeaTunnelRow
.../cdc/base/source/IncrementalSource.java	Register as advertised metadata columns

CommonOptions.java

Three values added after DELAY with supportMetadataTrans = true, which is the flag the Metadata transform checks to include a field in its switch statement:

BINLOG_FILE("BinlogFile", true),
BINLOG_POS("BinlogPos",   true),
BINLOG_ROW("BinlogRow",   true);

SourceRecordUtils.java

Three extractor methods, each null-safe for null value, null source struct, and missing schema field:

public static String getBinlogFile(SourceRecord record) {
    Struct value = (Struct) record.value();
    if (value == null) return null;
    Struct source = value.getStruct(Envelope.FieldName.SOURCE);
    if (source == null || source.schema().field("file") == null) return null;
    return source.getString("file");
}

public static Long getBinlogPos(SourceRecord record) {
    Struct value = (Struct) record.value();
    if (value == null) return null;
    Struct source = value.getStruct(Envelope.FieldName.SOURCE);
    if (source == null || source.schema().field("pos") == null) return null;
    return source.getInt64("pos");
}

public static Integer getBinlogRow(SourceRecord record) {
    Struct value = (Struct) record.value();
    if (value == null) return null;
    Struct source = value.getStruct(Envelope.FieldName.SOURCE);
    if (source == null || source.schema().field("row") == null) return null;
    return source.getInt32("row");
}

SeaTunnelRowDebeziumDeserializeSchema.java

In deserializeDataChangeRecord(), binlog coordinates are extracted once per SourceRecord (all rows within the same binlog event share the same source struct), then set on every emitted row (INSERT, DELETE, UPDATE_BEFORE, UPDATE_AFTER):

// Extract binlog position once per record — same source struct for all rows in the event
String binlogFile = SourceRecordUtils.getBinlogFile(record);
Long binlogPos = SourceRecordUtils.getBinlogPos(record);
Integer binlogRow = SourceRecordUtils.getBinlogRow(record);

// set on each emitted SeaTunnelRow:
MetadataUtil.setBinlogFile(row, binlogFile);
MetadataUtil.setBinlogPos(row, binlogPos);
MetadataUtil.setBinlogRow(row, binlogRow);

IncrementalSource.java

metadata.add(MetadataColumn.of(
    CommonOptions.BINLOG_FILE.getName(), BasicType.STRING_TYPE, (String) null, true, null, null));
metadata.add(MetadataColumn.of(
    CommonOptions.BINLOG_POS.getName(), BasicType.LONG_TYPE, (Long) null, true, null, null));
metadata.add(MetadataColumn.of(
    CommonOptions.BINLOG_ROW.getName(), BasicType.INT_TYPE, (Integer) null, true, null, null));

---

Parallelism and checkpoint safety

SeaTunnelRow.options is populated inside deserializeDataChangeRecord(), which runs within the source TaskGroup. All transforms are co-located in the same TaskGroup via IntermediateBlockingQueue (in-memory, no serialization). SeaTunnelRow.options is not part of the RecordSerializer wire format — only tableId, rowKind, arity, and fields[] are serialized — so this change has no impact on checkpoint state or cross-task-boundary data movement at any parallelism level.

On checkpoint restore, the source re-emits from its BinlogOffset and binlog coordinates are repopulated fresh from the new Debezium records. No state migration is required.

---

User-facing configuration

Using the same Metadata transform syntax as existing fields:

Metadata {
  plugin_input = "cdc_stream"
  plugin_output = "cdc_with_meta"
  metadata_fields {
    BinlogFile = c_binlog_file   # STRING — e.g. "mysql-bin-changelog.309222"
    BinlogPos  = c_binlog_pos    # BIGINT — e.g. 54106803
    BinlogRow  = c_binlog_row    # INT    — e.g. 0
  }
}

Downstream dedup key — lexicographically sortable:

CONCAT(
  COALESCE(c_binlog_file, ''), ':',
  LPAD(CAST(COALESCE(c_binlog_pos, 0) AS VARCHAR), 20, '0'), ':',
  LPAD(CAST(COALESCE(c_binlog_row, 0) AS VARCHAR), 10, '0')
) AS c_change_seq

Latest-state deduplication query:

SELECT * EXCEPT(c_row_kind, c_change_seq)
FROM cdc_table
WHERE c_row_kind != 'DELETE'
QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY c_change_seq DESC) = 1

---

Alternatives considered

Alternative	Why rejected
Single pre-formatted ChangeSeq STRING field	Loses type information; BinlogPos as BIGINT is independently useful as a numeric sort key
EventTime alone	Not unique — same value for multi-row events, and identical for UPDATE_BEFORE/UPDATE_AFTER pairs
Expose PostgreSQL LSN / Oracle SCN in the same PR	Different offset semantics per connector — better addressed per-connector in a follow-up

---

Compatibility

• Existing jobs: no impact — options entries are populated on every row but only materialise as output columns when explicitly mapped via the Metadata transform
• Non-MySQL connectors: all three fields return null, guarded by schema field presence check
• Snapshot phase: fields are null for snapshot rows (Debezium source struct has no file field during snapshot); users with startup.mode = latest are unaffected
• Checkpoint format: unchanged

[davidzollo]

Hi @ricky2129 , thanks for the incredibly detailed design doc. This is indeed a critical pain point for the Enterprise CDC-to-Data-Lake pipeline, and relying on strict transaction log coordinates instead of EventTime is the absolute right way to go.

Overall, your design looks very elegant and safe:

1. Putting these extra descriptors in SeaTunnelRow.options without interfering with core layout and serialization logic is correct and protects the checkpoint safety.
2. Generating null gracefully for non-MySQL sources and keeping backward compatibility is fully acknowledged.

A few points/suggestions before we proceed with the PR:

1. About Field Naming Generalization: Since different databases have different position terminologies (e.g., LSN for PostgreSQL, SCN for Oracle), maybe we could think abstractly whether we should use generalized names like LogFile, LogPos, or LogSequence if we want to make the Metadata transform generic across JDBC sources down the road. If you prefer sticking with explicit MySQL BinlogXX terminologies for this PR due to exactness, that is perfectly fine with me. We can document it specifically.
2. Snapshot Phase Null Handling: As you mentioned, startup.mode for historical snapshots will produce null for these binlog positions. Just make sure we can provide a small best-practice note in the config documentation on how users should write their COALESCE query in downstream engines to correctly sequence snapshot data vs incremental binlog data.
3. E2E Testing: Please ensure that we cover this new metadata extraction via both Unit Tests and a test case in our E2E module verifying both the value population and the null fallback logic.

The design doc looks great to me. Feel free to assign this to yourself and open a PR !

[ricky2129]

@davidzollo thanks for the thorough review — addressing each point:

1. Field naming generalization

We're happy to go either way, but our preference is to keep the explicit BinlogXX prefix for this PR for precision — BinlogPos is unambiguously a byte offset, while LogPos could mean different things across connectors (PostgreSQL LSN is a String like 0/16B3748, Oracle SCN is a BIGINT). Sharing a name across connectors with different types and semantics would complicate downstream SQL. A connector-specific follow-up for PostgresLsn, OracleScn etc. seems cleaner.

2. GTID as an additional field

While implementing this we realized binlog file/pos has one remaining failure mode: on RDS MySQL failover (replica promoted to primary), the new master starts a fresh binlog sequence. mysql-bin.000001:154 post-failover can numerically collide or sort before pre-failover positions — breaking the dedup key.

GTID (server_uuid:transaction_id) solves this: it is globally unique across the entire replication topology and survives failover because old transactions retain the original primary's server_uuid. MySQL 8.0+ recommends GTID-based replication specifically for this reason.

We're planning to add Gtid as a fourth field in the same PR:

GTID("Gtid", true) // STRING, e.g. "3E11FA47-71CA-11E1-9E33-C80AA9429562:23"

Null when GTID is disabled or for snapshot rows. Users on GTID-enabled RDS can use Gtid alone as the dedup identity key; users on binlog-only setups use BinlogFile/Pos/Row. Both are exposed; user picks what applies.

3. Source timestamp (source.ts_ms)

One more field worth considering: the existing EventTime metadata field maps to the envelope-level ts_ms — the time the connector processed the event. The Debezium source struct also carries a separate source.ts_ms — the time the database committed the transaction.

Should we expose this as a SourceTimestamp (or CommitTime) metadata field? It would be useful for:

• Data freshness SLAs (how old is the data at the lake)
• Time-partitioned lake layouts aligned to when the data actually changed in the source
• Human-readable ordering alongside GTID coordinates

Unlike binlog coords and GTID, source.ts_ms is available for all connectors (MySQL, PostgreSQL, Oracle, SQL Server) — making it a good candidate for a generic cross-connector metadata field. Happy to include it in this PR if the scope is acceptable, or defer to a follow-up.

4. Snapshot null handling + E2E tests

Will add best-practice notes in the config doc for COALESCE handling of snapshot rows, and will add unit + E2E test coverage for both populated and null fallback cases.

[julianbradford19-png]

Thanks for the feedback! I agree on all points. For this PR, I’ll keep the MySQL-specific field names for accuracy, add a note about COALESCE usage for snapshots in the docs, and ensure both unit and E2E tests cover the null fallback logic. Will proceed with opening the PR

…

On Sat, Mar 28, 2026, 10:13 AM David Zollo ***@***.***> wrote: *davidzollo* left a comment (apache/seatunnel#10641) <#10641 (comment)> Hi @ricky2129 <https://github.com/ricky2129> , thanks for the incredibly detailed design doc. This is indeed a critical pain point for the Enterprise CDC-to-Data-Lake pipeline, and relying on strict transaction log coordinates instead of EventTime is the absolute right way to go. Overall, your design looks very elegant and safe: 1. Putting these extra descriptors in SeaTunnelRow.options without interfering with core layout and serialization logic is correct and protects the checkpoint safety. 2. Generating null gracefully for non-MySQL sources and keeping backward compatibility is fully acknowledged. A few points/suggestions before we proceed with the PR: 1. *About Field Naming Generalization:* Since different databases have different position terminologies (e.g., LSN for PostgreSQL, SCN for Oracle), maybe we could think abstractly whether we should use generalized names like LogFile, LogPos, or LogSequence if we want to make the Metadata transform generic across JDBC sources down the road. If you prefer sticking with explicit MySQL BinlogXX terminologies for this PR due to exactness, that is perfectly fine with me. We can document it specifically. 2. *Snapshot Phase Null Handling:* As you mentioned, startup.mode for historical snapshots will produce null for these binlog positions. Just make sure we can provide a small best-practice note in the config documentation on how users should write their COALESCE query in downstream engines to correctly sequence snapshot data vs incremental binlog data. 3. *E2E Testing:* Please ensure that we cover this new metadata extraction via both Unit Tests and a test case in our E2E module verifying both the value population and the null fallback logic. The design doc looks great to me. Feel free to assign this to yourself and open a PR ! — Reply to this email directly, view it on GitHub <#10641?email_source=notifications&email_token=B74KUGZDM5YQMGKPG3NIQG34S7T2PA5CNFSNUABFM5UWIORPF5TWS5BNNB2WEL2JONZXKZKDN5WW2ZLOOQXTIMJUHAZDIMBXGQZKM4TFMFZW63VKON2WE43DOJUWEZLEUVSXMZLOOSWGM33PORSXEX3DNRUWG2Y#issuecomment-4148240742>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/B74KUG3FHB3MMUHUVXT53234S7T2PAVCNFSM6AAAAACW5RGIASVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHM2DCNBYGI2DANZUGI> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[ricky2129]

@julianbradford19-png added the same in the PR #10667