docs: RFC-108 - Multi-dataset incremental reads in Hudi Streamer#19299
docs: RFC-108 - Multi-dataset incremental reads in Hudi Streamer#19299ashokkumar-allu wants to merge 1 commit into
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the RFC! This proposes native multi-table incremental reads in Hudi Streamer via a new HoodieIncrMultiSource, a multi-dataset InputBatch/MultiDatasetTransformer contract, and a backward-compatible multi-table checkpoint format. The design is well-organized and the backward-compatibility story is thoughtfully considered; a few areas — the lenient checkpoint parser's failure semantics, cross-table schema evolution, scalability of N reads per batch, and a missing Alternatives section — could use more detail (see inline comments). Once those are addressed, a Hudi committer or PMC member can take it from here for a deeper design review.
| ``` | ||
|
|
||
| - Any value containing `=` is parsed as multi-table via a lenient parser that skips malformed entries | ||
| while preserving valid ones (so a single corrupt entry never causes the whole string to be |
There was a problem hiding this comment.
🤖 The lenient parser silently skipping a malformed entry seems to interact dangerously with the per-table missing-checkpoint strategy. If a table's entry is dropped, that table has no checkpoint and falls back to its missing_checkpoint_strategy — and the default READ_LATEST would skip all unprocessed data (silent data loss), which contradicts the fail-loud philosophy used elsewhere in this design. Could you spell out exactly what happens to a table whose entry is skipped? @nsivabalan could you weigh in on whether silently dropping a checkpoint entry is safe here versus failing the batch?
| | `hoodie.streamer.source.hudi.{table}.num_instants` | `HoodieIncrSourceConfig.NUM_INSTANTS_PER_FETCH` default | Per-table instants per fetch | | ||
| | `hoodie.streamer.source.hudi.{table}.missing_checkpoint_strategy` | `READ_LATEST` | `READ_LATEST` \| `READ_UPTO_LATEST_COMMIT` | | ||
| | `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` | `false` | On placeholder-schema resolution failure: `false` fails loud (checkpoint not advanced past a real failure); `true` returns an empty dataset and emits a `<table>.schema_resolution_failure` metric | | ||
| | `hoodie.streamer.source.hudi.{table}.partition.fields` / `.partition.extractor.class` | — | Optional per-table partitioning | |
There was a problem hiding this comment.
🤖 How does this handle schema evolution across the N source tables over time? With a single SchemaProvider feeding one commit, it isn't clear what happens when one source table evolves its schema mid-stream while others don't, or how the empty-but-schema'd placeholder dataset stays in sync with an evolved source schema. It might be worth a dedicated subsection on per-table schema resolution and target-schema reconciliation.
| 2. For each table, performs a `HoodieIncrSource`-equivalent incremental read using that table's own | ||
| `numInstantsPerFetch`, missing-checkpoint strategy, hollow-commit handling, timeline management and | ||
| "already caught up" detection. Tables that ingest at different rates never block one another. | ||
| 3. Returns a dataset for **every** configured table, in declaration order — an empty dataset carrying |
There was a problem hiding this comment.
🤖 "Tables ... never block one another" reads as a concurrency claim, but the sequence suggests the N incremental reads happen within a single batch on the driver. Are the per-table reads issued in parallel or sequentially, and is there an intended upper bound on N? A short note on the performance/resource profile as N grows (and whether one slow/large source stretches the whole batch) would strengthen the scalability story.
|
|
||
| - **Impact on existing users.** None by default. The feature is inert unless a user sets | ||
| `source.class=HoodieIncrMultiSource` and declares source tables. Existing single-source jobs, | ||
| transformers, and checkpoints continue to work byte-for-byte. |
There was a problem hiding this comment.
🤖 The RFC doesn't include an Alternatives Considered section. It would help to briefly document approaches that were evaluated and rejected — e.g. orchestrating multiple single-source jobs with multi-writer/coordinated commits, a dedicated join/union source, or handling this at the transformer layer only — and why a new source plus InputBatch change is the preferred path.
|
|
||
| A new source extending `Source<Dataset<Row>>` (SourceType `ROW`) — it extends `Source` directly | ||
| rather than `RowSource` so it can return a multi-dataset `InputBatch` without per-row sanitization | ||
| side effects. Per batch it: |
There was a problem hiding this comment.
🤖 Extending Source directly to skip per-row sanitization is a notable departure from RowSource. What sanitization is being bypassed (e.g. invalid/duplicate field-name handling), and could skipping it produce rows that later fail on write or violate downstream assumptions? It might be worth clarifying whether sanitization is re-applied after the transformer merge instead of dropped entirely.
| - **Scope.** No change to Hudi's write/commit path or on-disk storage format; the only serialized | ||
| change is the contents of the Streamer's own `deltastreamer.checkpoint.key`. | ||
|
|
||
| ## Rollout/Adoption Plan |
There was a problem hiding this comment.
🤖 Could you expand on the failure/atomicity semantics within a batch? Specifically: if the incremental read of table K succeeds but table K+1's read (or the merge/write) fails, is the entire batch aborted with no checkpoints advanced, and are all per-table checkpoints persisted atomically together with the single commit? Making the all-or-nothing boundary explicit would clarify the core atomicity guarantee this RFC is built on.
|
|
||
| ### Motivating use cases | ||
|
|
||
| - **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence |
There was a problem hiding this comment.
🤖 The motivation is well argued conceptually, but the Motivating use cases section is a single fairly abstract bullet. Could you add one or two concrete scenarios (real tables/joins) and, if available, any evidence of demand — issue reports, scale figures, or the compute cost of the current snapshot-read workaround — to strengthen the case that this is worth the added surface area?
github-actions
Bot
added
size:L
1fb4a24 to
15fee89
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the RFC! This proposes native multi-table incremental reads in Hudi Streamer via a new HoodieIncrMultiSource, a MultiDatasetTransformer contract, and a backward-compatible multi-table checkpoint format. The design is well developed and the recent round of expansion addressed the earlier feedback nicely — a few areas could still use more detail around what the cross-source consistency guarantee actually is, how <SRC_N> placeholders bind when the table list changes, and operational checkpoint-override / table-set-lifecycle semantics (see inline comments). Once those are addressed, a Hudi committer or PMC member can take it from here for a deeper design review.
| downstream merge, the merge step periodically observes fact rows whose dimension row has not yet | ||
| been ingested and either drops them (inner join) or emits rows with null dimension fields (outer | ||
| join) that must be repaired on a later run. A native multi-source incremental read advances both | ||
| tables under one commit, so the join always sees a consistent slice. |
There was a problem hiding this comment.
🤖 Atomic checkpoint advancement isn't the same as temporal alignment of the data: each table reads its own window sized by its own num_instants from its own checkpoint, so two sources ingesting at different rates can still land at very different commit/event times within the same batch — the same drift a downstream OUTER JOIN sees today. Could you clarify what "consistent slice" actually guarantees, and whether per-table read windows should be bounded to a shared commit-time ceiling to deliver true cross-source temporal consistency?
| `SqlFileBasedTransformer` implements it: | ||
|
|
||
| - Registers each input dataset as a unique Spark temp view (`HOODIE_SRC_TMP_TABLE_<uuid>`). | ||
| - SQL references datasets by index — `<SRC_0>`, `<SRC_1>`, … — with `<SRC>` kept as an alias for the |
There was a problem hiding this comment.
🤖 The <SRC_N> placeholders bind positionally to the order of ...multi.hudi.tables (line 208), while checkpoints bind by table name (line 235). If an operator reorders or edits the table list, the SQL's <SRC_0>/<SRC_1> silently rebind to different tables and the join becomes wrong — yet checkpoints keep resuming correctly by name, so nothing surfaces the mistake. Have you considered binding placeholders by table name/alias (e.g. <SRC:db.orders>) instead of by position?
| - Any value containing `=` is parsed as multi-table via a lenient parser that skips malformed entries | ||
| while preserving valid ones (so a single corrupt entry never causes the whole string to be | ||
| misclassified as single-table and collapsed under a default key). | ||
| - A bare value with no `=` is treated as a legacy single checkpoint (mapped to a default table), |
There was a problem hiding this comment.
🤖 How does the multi-table checkpoint interact with Streamer's operational checkpoint override (--checkpoint) and reset (hoodie.deltastreamer.checkpoint.reset_key)? A bare-timestamp override would parse as the legacy single form here and map to a single default table, which looks like it would drop the other tables' cursors on the next run. It'd help to specify override/reset semantics (and the expected value format) in multi-table mode.
| misclassified as single-table and collapsed under a default key). | ||
| - A bare value with no `=` is treated as a legacy single checkpoint (mapped to a default table), | ||
| giving automatic backward compatibility and a clean upgrade path from an existing single-source job. | ||
| - The formatted output uses a deterministic key order (backed by a `TreeMap`) so the same set of |
There was a problem hiding this comment.
🤖 The migration section covers single→multi upgrade, but not changing the table set over the life of a running job. Adding a table gives it no checkpoint entry (first-run → READ_LATEST skips history, or READ_UPTO_LATEST_COMMIT re-reads everything), and removing a table leaves a stale table=ts entry serialized in the TreeMap-backed string indefinitely. Could you document the intended behavior when ...multi.hudi.tables evolves — including whether stale entries are pruned and how a newly-added table should be seeded?
| - **Routing.** In `fetchNextBatchFromSource`, when a multi-dataset transformer is present the full | ||
| `getBatches()` list is passed to `MultiDatasetTransformer#apply`; otherwise the existing | ||
| single-dataset path (`getBatch().map(...)`) is used unchanged. An empty-batch guard skips the | ||
| transformer when the source produced no data (rather than passing an empty list that would throw). |
There was a problem hiding this comment.
🤖 There's some tension between step 3 (line 139), where the source always returns a dataset for every configured table, and this empty-batch guard that "skips the transformer when the source produced no data." If N datasets are always present, when does the guard actually fire — only when all N are empty? And in that all-caught-up case, does the batch still produce a commit and persist checkpoints, or is it a full no-op? Worth spelling out.
|
@ashok-allu, can you please claim the RFC first like so in a separate PR: #19303 Thanks! |
Propose native multi-table incremental reads in Hudi Streamer via a new HoodieIncrMultiSource that reads N Hudi tables incrementally in a single batch with independent per-table checkpoints, a MultiDatasetTransformer contract implemented by SqlFileBasedTransformer for merging datasets with a single SQL file referencing <SRC_0>, <SRC_1>, ..., and a backward-compatible multi-table checkpoint format managed by MultiTableCheckpointManager. Enables a single, atomic ingestion job - one commit, one checkpoint - for related datasets. The RFC-108 entry in rfc/README.md is claimed separately in the split-out claim PR to keep this content PR reviewable in isolation.
15fee89 to
63c738a
Compare
|
@voonhous thanks for the pointer! I've split this into two PRs following the pattern of #19303:
Happy to hold this content PR until the claim PR is merged. |
5 participants
Describe the issue this Pull Request addresses
This PR adds a new RFC — RFC-108: Multi-dataset incremental reads in Hudi Streamer — proposing native multi-table incremental reads in Hudi Streamer so that N Hudi tables can be read incrementally into a single, atomic ingestion job (one commit, one checkpoint).
Related feature issue: Closes #19281
Summary and Changelog
Hudi Streamer (DeltaStreamer) can currently read incrementally from exactly one source per job. Pipelines that need to combine incremental changes from several Hudi tables into one target must run one Streamer job per source and merge downstream (drift, no cross-source atomicity, extra watermark/monitoring overhead) or fall back to expensive full-snapshot reads for all but one table.
This RFC proposes:
HoodieIncrMultiSource— a new source that reads N Hudi tables incrementally in a single batch with independent per-table checkpoints (each table advances on its ownnumInstantsPerFetch/ missing-checkpoint strategy).InputBatch— additiveList<T> batchesalongside the existingOption<T> batch;getBatch()now fails loud if a multi-dataset batch would be silently narrowed, single-source callers are unaffected.MultiDatasetTransformercontract, implemented bySqlFileBasedTransformer— merges the per-source datasets with a single SQL file referencing<SRC_0>,<SRC_1>, …;<SRC>is kept as an alias for the first dataset for backward compatibility; supports multi-statement SQL and CTE-styleCREATE [OR REPLACE] TEMPORARY VIEW, with temp-view cleanup on both success and failure.MultiTableCheckpointManager— persists per-table checkpoints astable1=ts1,table2=ts2indeltastreamer.checkpoint.key; a bare timestamp continues to be honored as a legacy single-source checkpoint, so existing jobs migrate without checkpoint surgery.StreamSync/SourceFormatAdapterwiring — startup validation thatHoodieIncrMultiSourcerequires aMultiDatasetTransformer(resolved even inside aChainedTransformer) to prevent a silent-data-loss path where only the last source would be written; per-dataset error-event handling on the multi-dataset row path.Changelog:
rfc/rfc-108/rfc-108.md— new RFC document (Abstract, Background, Implementation, Rollout/Adoption Plan, Test Plan).rfc/README.md— add RFC-108 entry with statusUNDER REVIEW.No code changes in this PR — this is the RFC document only.
Impact
source.class=HoodieIncrMultiSourceand does not alter behavior of existing single-source Streamer jobs, transformers, or checkpoints. API changes are additive; the checkpoint format degrades cleanly to the legacy single-timestamp form.Risk Level
none — RFC documentation only; no code or config changes.
Documentation Update
This PR is the design documentation for the proposed feature. No website updates are required until the implementation lands; user-facing configuration and the recommended OUTER JOIN / merge-payload patterns will be documented on the Hudi website when the feature is shipped.
Contributor's checklist