ashokkumar-allu opened a new issue, #19281:
URL: https://github.com/apache/hudi/issues/19281
### Feature Description
**What the feature achieves:**
Add first-class support for reading **multiple Hudi datasets incrementally
in a single Hudi
Streamer (DeltaStreamer) job** and merging them with a single SQL
transformation — atomically,
under one checkpoint and one commit.
Concretely this introduces:
- **`HoodieIncrMultiSource`** — a new Streamer source that reads N Hudi
tables incrementally in
one batch, each table with its own independent checkpoint,
`numInstantsPerFetch`, missing-
checkpoint strategy, hollow-commit handling and timeline management
(reusing `HoodieIncrSource`
semantics per table).
- **`MultiDatasetTransformer`** — a transformer interface that accepts
`List<Dataset<Row>>`;
`SqlFileBasedTransformer` implements it and lets a SQL file reference each
source by index as
`<SRC_0>`, `<SRC_1>`, … (with `<SRC>` kept as an alias for the first, for
backward compatibility).
- **`MultiTableCheckpointManager`** — a per-table checkpoint format
(`table1=ckpt1,table2=ckpt2`) stored in `deltastreamer.checkpoint.key`,
fully backward
compatible with the legacy single-timestamp format.
- **Multi-dataset `InputBatch`** — carries a `List<Dataset<Row>>` via
`getBatches()`; `getBatch()`
is unchanged for single-dataset sources.
**Why this feature is needed:**
Today Hudi has no single-query mechanism for incremental reads across
multiple tables. Users must
either run **one DeltaStreamer job per source table and merge downstream**,
or fall back to
**full snapshot reads** for all-but-one table. Both are problematic:
- **Data inconsistency / silent failures** — separate jobs drift in time
(Table A at T+10, Table B
at T+7), producing incomplete/misleading joined output that only backfills
can fix.
- **No atomicity** — there is no transaction boundary across N jobs; a
partial success leaves the
unified table half-updated.
- **Operational overhead** — N watermarks to reconcile (watermark drift
makes replay/debugging
hard), and N pipelines to monitor and alert on.
- **Snapshot-read workaround is expensive** — re-reading the last X days of
every other table on
each run burns far more compute than a true incremental read.
**Expected improvement:**
- Fewer data-inconsistency-driven backfill incidents on multi-source
pipelines.
- Lower end-to-end data lag for multi-source ingestion.
- Lower total compute vs. the snapshot-read workaround.
**Non-goals:**
- Does **not** auto-resolve join/merge semantics — users still own join keys
and conflict
resolution (outer joins, merge payloads).
- Does **not** change Hudi's write/commit mechanism — this is confined to
source reading and
transformation inside Hudi Streamer.
### User Experience
## **How users will use this feature:**
Users opt in purely through Streamer configuration + a SQL file. No code
changes required.
- Configuration changes needed
### 1. Point the Streamer at the new source
```properties
hoodie.deltastreamer.source.class=org.apache.hudi.utilities.sources.HoodieIncrMultiSource
```
### 2. Declare the source tables and per-table settings
```properties
# Ordered list of source tables (order maps to <SRC_0>, <SRC_1>, … in SQL)
hoodie.streamer.source.multi.hudi.tables=db.orders,db.users
# Per-table config. Dots in a table name become underscores in the property
key;
# the original dotted name is preserved inside the checkpoint.
hoodie.streamer.source.hudi.db_orders.path=/hudi/orders
hoodie.streamer.source.hudi.db_orders.num_instants=5
hoodie.streamer.source.hudi.db_orders.missing_checkpoint_strategy=READ_LATEST
hoodie.streamer.source.hudi.db_users.path=/hudi/users
hoodie.streamer.source.hudi.db_users.num_instants=3
hoodie.streamer.source.hudi.db_users.missing_checkpoint_strategy=READ_UPTO_LATEST_COMMIT
# Multi-table checkpointing (default true). false => legacy
single-checkpoint behavior.
hoodie.streamer.source.multi.hudi.checkpoint.enable=true
```
### 3. Merge the datasets with a SQL file
```properties
hoodie.deltastreamer.transformer.class=org.apache.hudi.utilities.transform.SqlFileBasedTransformer
hoodie.deltastreamer.transformer.sql.file=/path/to/multi_table_transform.sql
```
```sql
-- multi_table_transform.sql : <SRC_0> = db.orders, <SRC_1> = db.users
-- OUTER JOIN is recommended so a table with no new data in this batch never
drops rows.
SELECT
o.order_id, o.amount, o.user_id,
u.user_name, u.email
FROM <SRC_0> o
FULL OUTER JOIN <SRC_1> u
ON o.user_id = u.user_id;
```
### Checkpoint (stored in `deltastreamer.checkpoint.key`, one atomic commit)
```
Multi-table : "db.orders=20250606182826197,db.users=20250606182830145"
Legacy/single: "20250606182826197" (still read/written when multi-table
checkpoint is disabled)
```
## API changes
API changes are all additive / backward compatible
- `InputBatch#getBatches(): List<T>` — new; `getBatch()` unchanged for
single-dataset sources
(it now fails loud if called on a multi-dataset batch, rather than
silently dropping datasets).
- `MultiDatasetTransformer` — new interface: `apply(jsc, spark,
List<Dataset<Row>>, props)`.
- `SqlFileBasedTransformer` — now implements `MultiDatasetTransformer`;
adds `<SRC_0>`,`<SRC_1>`,…
placeholders alongside the existing `<SRC>`.
## Usage examples
- Provided in config section
## Guardrails
- Startup validation: `HoodieIncrMultiSource` **requires** a
`MultiDatasetTransformer` (e.g.
`SqlFileBasedTransformer`, directly or inside a `ChainedTransformer`).
Misconfiguration fails
the job at startup instead of silently writing only the last table.
- Every configured table always yields a dataset each batch — an empty,
correctly-schema'd
dataset when there's no new data — so `LEFT/FULL OUTER JOIN`s always have
a valid relation.
## Guidance
- Prefer **OUTER JOIN** on incremental sources (inner join can silently miss
delayed data).
- For overlapping columns / partial-field updates, use a **merge payload**
(e.g. a
`HoodieGenericMergePayload` or a custom payload) rather than
last-writer-wins overwrite.
### Hudi RFC Requirements
**RFC PR link:** (if applicable)
Work in progress
**Why RFC is/isn't needed:**
- Does this change public interfaces/APIs? (Yes)
- Does this change storage format? (No)
- Justification:
This introduces a **new Hudi Streamer source** and **new public interfaces**
(`HoodieIncrMultiSource`,
`MultiDatasetTransformer`, `InputBatch#getBatches()`) plus a new multi-table
checkpoint encoding —
per the RFC process, new Streamer sources and changes to public interfaces
warrant an RFC.
- It does **not** change the on-disk table/storage format. The only
serialized change is the
contents of `deltastreamer.checkpoint.key` in the Streamer's own `.commit`
metadata, which
remains backward compatible: legacy single-value checkpoints are read and
honored, and the
multi-table format degrades to the legacy format when
`...checkpoint.enable=false`.
- All Java API changes are additive; existing single-source pipelines and
non-SQL transformers
are unaffected.
--
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]
