hudi-agent commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3624641379
########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,385 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-108: Multi-dataset incremental reads in Hudi Streamer + +## Proposers + +- @ashokkumar-allu + +## Approvers + +- @nsivabalan + +## Status + +Issue: https://github.com/apache/hudi/issues/19281 + +> Please keep the status updated in `rfc/README.md`. + +## Abstract + +Apache Hudi Streamer (DeltaStreamer) can read incrementally from exactly one source per job. Pipelines +that need to combine incremental changes from **several Hudi tables** into one target table must +therefore run one Streamer job per source and merge the results downstream, or fall back to expensive +full-snapshot reads for all-but-one table. The multi-job approach has no cross-source transaction +boundary, so the tables it stitches together drift in time, produce incomplete/inconsistent joined +output, and require backfills to repair; it also multiplies watermark and monitoring overhead. + +This RFC proposes **native multi-table incremental reads** in Hudi Streamer: a new source +(`HoodieIncrMultiSource`) that reads N Hudi tables incrementally in a single batch with **independent +per-table checkpoints**, a **multi-dataset transformer** contract (`MultiDatasetTransformer`, +implemented by `SqlFileBasedTransformer`) that merges the datasets with a single SQL file referencing +`<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible multi-table checkpoint format** managed by +`MultiTableCheckpointManager`. The result is a single, atomic ingestion job — one commit, one +checkpoint — for related datasets. + +## Background + +### Current single-source flow + +``` +Streamer CLI + └─ StreamSync.sync() + └─ SourceFormatAdapter (adapts ROW / JSON / AVRO to a common shape) + └─ HoodieIncrSource (incremental read of ONE Hudi table via + hoodie.deltastreamer.source.hoodieincr.path) + └─ Transformer (single Dataset<Row> in → Dataset<Row> out) + └─ write + commit (checkpoint = single timestamp, e.g. "20250606182826197") +``` + +`InputBatch` carries a single `Option<T>` payload and one checkpoint string. Checkpoints are opaque +single timestamps stored in `deltastreamer.checkpoint.key` in the Streamer commit metadata. + +### Why this is insufficient for multi-source pipelines + +- **Data inconsistency / silent failures.** N independent jobs advance at different rates. A join of + Table A (at T+10) and Table B (at T+7) yields incomplete or misleading data with no error raised — + discovered only later, and repaired only with backfills. +- **No atomicity across sources.** There is no shared transaction boundary. If one job fails after + others commit, the unified downstream table is left partially updated. +- **Operational overhead.** Each job maintains its own watermark (watermark drift complicates replay + and debugging) and its own monitoring/alerting surface. +- **Expensive workaround.** Reading one table incrementally while snapshot-reading the last X days of + every other table consumes far more compute than a true incremental read. + +### Motivating use cases + +- **Cross-entity enrichment for analytics or serving.** A downstream table is populated by joining + two Hudi tables — e.g. a fact table (transactions, events, clicks, orders) with a dimension table + (accounts, users, products) on a foreign key. Both sources ingest continuously from different + upstreams and each is already a Hudi table. When run as two independent Streamer jobs plus a + 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. +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). Four building blocks plus wiring. + +### 1. Multi-dataset `InputBatch` + +`InputBatch<T>` gains a `List<T> batches` alongside the existing `Option<T> batch`: + +- New constructor `InputBatch(List<T> batches, String checkpointForNextBatch, SchemaProvider)`. +- `getBatches(): List<T>` returns all datasets (defensive copy). +- `getBatch(): Option<T>` is retained for single-dataset sources and now **fails loud** — it throws + via `ValidationUtils.checkState` if more than one batch is present, so a multi-dataset batch can + never be silently narrowed to one dataset. For single-dataset construction, `getBatches()` returns + a singleton and `getBatch()` returns that element (unchanged behavior). + +This keeps every existing single-source source and transformer working unchanged. + +### 2. `HoodieIncrMultiSource` + +A new source extending `Source<Dataset<Row>>` (SourceType `ROW`). It extends `Source` directly rather +than `RowSource` because `RowSource#fetchNextBatch` calls `SanitizationUtils` on the returned dataset +(masking Avro-invalid field names) before wrapping the result in an `InputBatch` — that per-row scan +would (a) run twice, once here and again after the transformer merges the datasets, and (b) only +supports a single `Option<Dataset<Row>>` payload, not the multi-dataset list this source needs. +Rows read from a Hudi table are already conformant with that table's Avro schema, so this bypass is +safe; sanitization still applies at write time to the merged transformer output when the user has +enabled it. Per batch the source: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +2. Iterates the parsed tables **sequentially on the driver** and performs a `HoodieIncrSource`-equivalent + incremental read for each — its 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; see *Scale and parallelism* below for the wall-clock implications. +3. Returns a dataset for **every** configured table, in declaration order — an empty dataset carrying + the Hudi table's schema when a table has no new incremental data — so downstream `LEFT/FULL OUTER + JOIN`s always resolve against a valid relation. +4. Packages all datasets into one `InputBatch` via the multi-dataset constructor, with the combined + multi-table checkpoint (or a plain timestamp when multi-table checkpointing is disabled). The + `InputBatch` carries a `NullSchemaProvider` — Streamer infers the write schema from the merged + transformer output rather than from any single input source (see *Per-table schema resolution*). + +#### Per-table schema resolution + +Each source table's read schema is resolved independently against that table's own Hudi metadata, +so heterogeneous or independently-evolving source schemas are supported directly: + +- A table with new commits: the read produces rows conformant with the current commit's schema. +- A table that is caught up or empty: the source emits an empty `Dataset<Row>` built from that table's Review Comment: 🤖 When a source has no new data, the placeholder is built from "the table's current Hudi schema" — will that be schema-identical to a non-empty incremental read of the same table, including the `_hoodie_*` meta columns and `drop.all.meta.fields` handling? If the empty vs. non-empty shapes differ, `<SRC_K>` changes columns depending on whether that source advanced, so the merged output (and the write schema derived from it via `NullSchemaProvider`) can flip batch-to-batch. @yihua does this line up with `HoodieIncrSource`'s actual output schema? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,385 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-108: Multi-dataset incremental reads in Hudi Streamer + +## Proposers + +- @ashokkumar-allu + +## Approvers + +- @nsivabalan + +## Status + +Issue: https://github.com/apache/hudi/issues/19281 + +> Please keep the status updated in `rfc/README.md`. + +## Abstract + +Apache Hudi Streamer (DeltaStreamer) can read incrementally from exactly one source per job. Pipelines +that need to combine incremental changes from **several Hudi tables** into one target table must +therefore run one Streamer job per source and merge the results downstream, or fall back to expensive +full-snapshot reads for all-but-one table. The multi-job approach has no cross-source transaction +boundary, so the tables it stitches together drift in time, produce incomplete/inconsistent joined +output, and require backfills to repair; it also multiplies watermark and monitoring overhead. + +This RFC proposes **native multi-table incremental reads** in Hudi Streamer: a new source +(`HoodieIncrMultiSource`) that reads N Hudi tables incrementally in a single batch with **independent +per-table checkpoints**, a **multi-dataset transformer** contract (`MultiDatasetTransformer`, +implemented by `SqlFileBasedTransformer`) that merges the datasets with a single SQL file referencing +`<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible multi-table checkpoint format** managed by +`MultiTableCheckpointManager`. The result is a single, atomic ingestion job — one commit, one +checkpoint — for related datasets. + +## Background + +### Current single-source flow + +``` +Streamer CLI + └─ StreamSync.sync() + └─ SourceFormatAdapter (adapts ROW / JSON / AVRO to a common shape) + └─ HoodieIncrSource (incremental read of ONE Hudi table via + hoodie.deltastreamer.source.hoodieincr.path) + └─ Transformer (single Dataset<Row> in → Dataset<Row> out) + └─ write + commit (checkpoint = single timestamp, e.g. "20250606182826197") +``` + +`InputBatch` carries a single `Option<T>` payload and one checkpoint string. Checkpoints are opaque +single timestamps stored in `deltastreamer.checkpoint.key` in the Streamer commit metadata. + +### Why this is insufficient for multi-source pipelines + +- **Data inconsistency / silent failures.** N independent jobs advance at different rates. A join of + Table A (at T+10) and Table B (at T+7) yields incomplete or misleading data with no error raised — + discovered only later, and repaired only with backfills. +- **No atomicity across sources.** There is no shared transaction boundary. If one job fails after + others commit, the unified downstream table is left partially updated. +- **Operational overhead.** Each job maintains its own watermark (watermark drift complicates replay + and debugging) and its own monitoring/alerting surface. +- **Expensive workaround.** Reading one table incrementally while snapshot-reading the last X days of + every other table consumes far more compute than a true incremental read. + +### Motivating use cases + +- **Cross-entity enrichment for analytics or serving.** A downstream table is populated by joining + two Hudi tables — e.g. a fact table (transactions, events, clicks, orders) with a dimension table + (accounts, users, products) on a foreign key. Both sources ingest continuously from different + upstreams and each is already a Hudi table. When run as two independent Streamer jobs plus a + 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. +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). Four building blocks plus wiring. + +### 1. Multi-dataset `InputBatch` + +`InputBatch<T>` gains a `List<T> batches` alongside the existing `Option<T> batch`: + +- New constructor `InputBatch(List<T> batches, String checkpointForNextBatch, SchemaProvider)`. +- `getBatches(): List<T>` returns all datasets (defensive copy). +- `getBatch(): Option<T>` is retained for single-dataset sources and now **fails loud** — it throws + via `ValidationUtils.checkState` if more than one batch is present, so a multi-dataset batch can + never be silently narrowed to one dataset. For single-dataset construction, `getBatches()` returns + a singleton and `getBatch()` returns that element (unchanged behavior). + +This keeps every existing single-source source and transformer working unchanged. + +### 2. `HoodieIncrMultiSource` + +A new source extending `Source<Dataset<Row>>` (SourceType `ROW`). It extends `Source` directly rather +than `RowSource` because `RowSource#fetchNextBatch` calls `SanitizationUtils` on the returned dataset +(masking Avro-invalid field names) before wrapping the result in an `InputBatch` — that per-row scan +would (a) run twice, once here and again after the transformer merges the datasets, and (b) only +supports a single `Option<Dataset<Row>>` payload, not the multi-dataset list this source needs. +Rows read from a Hudi table are already conformant with that table's Avro schema, so this bypass is +safe; sanitization still applies at write time to the merged transformer output when the user has +enabled it. Per batch the source: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +2. Iterates the parsed tables **sequentially on the driver** and performs a `HoodieIncrSource`-equivalent + incremental read for each — its 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; see *Scale and parallelism* below for the wall-clock implications. +3. Returns a dataset for **every** configured table, in declaration order — an empty dataset carrying + the Hudi table's schema when a table has no new incremental data — so downstream `LEFT/FULL OUTER + JOIN`s always resolve against a valid relation. +4. Packages all datasets into one `InputBatch` via the multi-dataset constructor, with the combined + multi-table checkpoint (or a plain timestamp when multi-table checkpointing is disabled). The + `InputBatch` carries a `NullSchemaProvider` — Streamer infers the write schema from the merged + transformer output rather than from any single input source (see *Per-table schema resolution*). + +#### Per-table schema resolution + +Each source table's read schema is resolved independently against that table's own Hudi metadata, +so heterogeneous or independently-evolving source schemas are supported directly: + +- A table with new commits: the read produces rows conformant with the current commit's schema. +- A table that is caught up or empty: the source emits an empty `Dataset<Row>` built from that table's + current Hudi schema, so downstream joins and column references remain valid rather than exploding + on an "unresolved column" error. +- Placeholder-schema resolution failure for any table: controlled by + `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` (default `false` — fail loud so a real + failure does not silently advance the checkpoint; `true` returns an empty dataset for that table Review Comment: 🤖 With `skip_on_schema_failure=true` the batch still commits even though this table produced no rows — what checkpoint does it get written into the committed multi-table map? If the map is rebuilt only from the tables that were successfully read (dropping this table's prior value rather than carrying it forward), the next batch sees no checkpoint for it, treats it as first-run, and with the default `READ_LATEST` would skip its backlog — a silent data-loss window. Could you specify the invariant that every configured table's checkpoint is carried forward whenever it doesn't advance? @nsivabalan <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
