hudi-agent commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3591845135
########## 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. Review 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? <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 + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `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 | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`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 Review 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? <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 + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `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 | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`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 + first dataset (backward compatible). +- Supports multi-statement SQL files and `CREATE [OR REPLACE] TEMPORARY VIEW` (e.g. CTE-style setup); + user-created views are tracked and dropped alongside the source temp views in a `finally` block. +- Validates inputs: rejects null/empty dataset lists and any dangling `<SRC_N>` placeholder that has + no corresponding dataset (typo guard), with clear error messages. + +Example SQL (`<SRC_0>` = orders, `<SRC_1>` = users): + +```sql +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; +``` + +### 4. `MultiTableCheckpointManager` + +Parses, formats, and manipulates per-table checkpoints stored in `deltastreamer.checkpoint.key`: + +``` +Multi-table : "table1=20250606182826197,table2=20250606182830145,table3=20250606182828456" +Single/legacy: "20250606182826197" +``` + +- 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), + 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 Review 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? <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 + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `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 | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`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 + first dataset (backward compatible). +- Supports multi-statement SQL files and `CREATE [OR REPLACE] TEMPORARY VIEW` (e.g. CTE-style setup); + user-created views are tracked and dropped alongside the source temp views in a `finally` block. +- Validates inputs: rejects null/empty dataset lists and any dangling `<SRC_N>` placeholder that has + no corresponding dataset (typo guard), with clear error messages. + +Example SQL (`<SRC_0>` = orders, `<SRC_1>` = users): + +```sql +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; +``` + +### 4. `MultiTableCheckpointManager` + +Parses, formats, and manipulates per-table checkpoints stored in `deltastreamer.checkpoint.key`: + +``` +Multi-table : "table1=20250606182826197,table2=20250606182830145,table3=20250606182828456" +Single/legacy: "20250606182826197" +``` + +- 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), Review 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. <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 + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `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 | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`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 + first dataset (backward compatible). +- Supports multi-statement SQL files and `CREATE [OR REPLACE] TEMPORARY VIEW` (e.g. CTE-style setup); + user-created views are tracked and dropped alongside the source temp views in a `finally` block. +- Validates inputs: rejects null/empty dataset lists and any dangling `<SRC_N>` placeholder that has + no corresponding dataset (typo guard), with clear error messages. + +Example SQL (`<SRC_0>` = orders, `<SRC_1>` = users): + +```sql +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; +``` + +### 4. `MultiTableCheckpointManager` + +Parses, formats, and manipulates per-table checkpoints stored in `deltastreamer.checkpoint.key`: + +``` +Multi-table : "table1=20250606182826197,table2=20250606182830145,table3=20250606182828456" +Single/legacy: "20250606182826197" +``` + +- 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), + 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 + per-table checkpoints always serializes to the same string, making commits and diffs reproducible. + +#### Behavior when the parser skips a malformed entry + +The lenient parser is designed to keep a single corrupt `table=ts` entry from taking down the whole +job. That behavior has a data-correctness consequence that the operator must understand: + +- A skipped entry is logged as `WARN`, and the source emits a + `multi_table_checkpoint.malformed_entries` counter metric with the number of dropped entries so the + drop is observable in monitoring, not silent. +- The affected table then has **no checkpoint** for the next batch, so `HoodieIncrMultiSource` treats + it as first-run and applies its `missing_checkpoint_strategy`. With the default `READ_LATEST` this + means the table jumps to the latest instant and **skips any data between the lost checkpoint and + latest** — a silent data-loss window sized by however long the corruption went unnoticed. +- Operators who cannot tolerate that failure mode should set + `hoodie.streamer.source.hudi.{table}.missing_checkpoint_strategy=READ_UPTO_LATEST_COMMIT`, which + reads the whole table history up to latest instead — safer default for reprocessable / dedup-keyed + writes, but heavier. +- The malformed-entries metric should be alerted on with threshold `> 0` in any pipeline that ingests + event streams where re-reading history is not equivalent to a resume. + +### 5. Wiring into `StreamSync` and `SourceFormatAdapter` + +- **Startup validation.** If the source is `HoodieIncrMultiSource`, `StreamSync` requires a + `MultiDatasetTransformer` (resolved via `UtilHelpers.resolveMultiDatasetTransformer`, so it is found + even when wrapped in a `ChainedTransformer`). If none is configured, the job throws + `HoodieStreamerException` at startup — preventing the silent-data-loss failure mode where only the + last source table would be written. +- **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). Review 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. <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]
