apache / apache/hudi

Multi-dataset incremental reads in Hudi Streamer

Open
#19,281 1 comment 0 reactions 0 assignees View on GitHub
type:feature
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.