fightBoxing opened a new pull request, #4487:
URL: https://github.com/apache/flink-cdc/pull/4487
## Purpose
Add support for extracting Oracle **ROWID pseudo-column** as a metadata
column in Oracle CDC. Users can then access it in Flink SQL via:
```sql
CREATE TABLE oracle_source (
ID INT,
NAME STRING,
row_id STRING METADATA FROM 'row_id' VIRTUAL -- new
) WITH (
'connector' = 'oracle-cdc',
'scan.incremental.snapshot.enabled' = 'true',
...
);
```
Typical use cases: data lineage, exact-row identification, idempotent
downstream writes, systems that rely on ROWID for reconciliation.
## What is changed
### 1. `OracleReadableMetaData` — new `ROW_ID` enum
Reads the ROWID value from the `SourceRecord` headers (key = `ROWID`).
### 2. `OracleScanFetchTask` (snapshot phase)
Oracle's `SELECT *` does **not** include ROWID. So:
- Rewrite the split-scan SQL: `SELECT * FROM tab` → `SELECT T0.*, ROWID FROM
tab T0` (ROWID placed at the tail keeps physical column positions unchanged).
- Manually construct `ColumnArray` from the table's own columns to bypass
Debezium's `ColumnUtils.toArray(rs, table)` strict validation, which would
otherwise reject the extra ROWID column.
- Extract ROWID via `((OracleResultSet) rs).getROWID(N+1)` and inject it
into `SourceRecord` headers by overriding
`SnapshotChangeRecordEmitter#getEmitConnectHeaders`.
Streaming phase already carries ROWID in headers through Debezium's
`LogMinerChangeRecordEmitter` (from `V$LOGMNR_CONTENTS.ROW_ID`), so no changes
are needed there.
### 3. `JdbcSourceFetchTaskContext` — headers-loss bug fix ⭐
While implementing the above, we discovered a **latent bug affecting all
JDBC-based CDC connectors** (MySQL, PostgreSQL, SQL Server, Db2, Oracle):
Both `rewriteOutputBuffer()` (used for PK-update path) and
`formatMessageTimestamp()` (used for snapshot record normalization) were
constructing new `SourceRecord`s using the 8-argument constructor, silently
discarding the **timestamp** and **headers** from the original record.
Any header-based metadata — including the new Oracle `row_id`, and any
future header-based columns that may be added — would be silently dropped by
these two rewrite paths.
Fix: switch both call sites to the 10-argument `SourceRecord` constructor so
`timestamp` and `headers` are preserved.
## Verification
Compilation:
```
mvn compile -pl
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc -am
-DskipTests
```
✅ BUILD SUCCESS
Runtime (Oracle Database 12c Standard Edition 12.1.0.2.0, back-ported to
release-2.4 and verified end-to-end):
```
+I[2, bbb, AABDuHAAGAAAAF3AAA]
+I[3, ccc, AABDuHAAGAAAAF0AAA]
+I[4, ddd, AABDuHAAGAAAAF0AAB]
-D[2, bbb, AABDuHAAGAAAAF3AAA]
```
Both snapshot (READ) and streaming (INSERT/UPDATE/DELETE) rows carry a valid
ROWID.
## Data flow
```
Snapshot phase:
OracleScanFetchTask (rs.getROWID)
→ SnapshotChangeRecordEmitter.getEmitConnectHeaders (inject ROWID)
→ BufferingSnapshotChangeRecordReceiver (new SourceRecord with headers)
→ ChangeEventQueue
Streaming phase (already works):
LogMinerChangeRecordEmitter.getEmitConnectHeaders (inject ROWID)
→ ChangeEventQueue
Reader side (bug fix here):
IncrementalSourceScanFetcher.pollSplitRecords
→ JdbcSourceFetchTaskContext.formatMessageTimestamp/rewriteOutputBuffer
(preserve headers) ⭐
→ OracleReadableMetaData.ROW_ID.read(record.headers())
```
## User requirements
- `scan.incremental.snapshot.enabled=true` (this PR does not touch
Debezium's native snapshot path).
- Table supplemental logging enabled for streaming capture:
```
ALTER TABLE <schema>.<table> ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
```
## Compatibility
- No breaking API changes.
- No new configuration options.
- Users who don't declare `row_id` metadata column see zero behavioral
change (SQL rewrite still runs but only adds a single pseudo-column to the JDBC
query, which is negligible).
- The `JdbcSourceFetchTaskContext` fix is transparent — records that
previously had `null`/absent headers still have `null`/absent headers
afterwards.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
