Multi-dataset incremental reads in Hudi Streamer
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### 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>`;
`SqlFileBasedTransformer` implements it and lets a SQL file reference each source by index as
``, ``, … (with `` 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>` 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 , , … 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 : = db.orders, = 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 o
FULL OUTER JOIN 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` — 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>, props)`.
- `SqlFileBasedTransformer` — now implements `MultiDatasetTransformer`; adds ``,``,…
placeholders alongside the existing ``.
## 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.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reviewing the named HoodieIncrMultiSource, MultiDatasetTransformer, InputBatch#getBatches(), SqlFileBasedTransformer, and MultiTableCheckpointManager APIs. Trace the existing HoodieIncrSource and single-dataset transformer behavior before defining how per-table checkpoints, empty batches, validation, and legacy formats interact. Done requires the additive API and configuration behavior, atomic multi-table checkpointing, SQL placeholders, backward compatibility, and the described guardrails to be covered by the project’s tests.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark, sql
- Domain
- backend-api-design, data-engineering, stream-processing
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100