ashokkumar-allu commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3650976347
########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +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 + 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). + +**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 Review Comment: Addressed via the new **checkpoint carry-forward invariant** in the *Per-source schema resolution* section (L249–256) — an index whose entry was dropped by the lenient parser is rebuilt from the previous committed checkpoint rather than falling through to `missing_checkpoint_strategy`. Silent drop is also observable as a `multi_source_checkpoint.malformed_entries` counter metric (Section 7, L383–392), documented for hard-alerting. ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +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 + 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). + +**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 | Review Comment: Addressed via the new **Section 4: Per-source schema resolution** (L220–289). Each per-source `HoodieIncrSource` resolves its read schema independently against its own Hudi table's metadata — heterogeneous or independently-evolving schemas are supported directly. The empty-vs-non-empty schema-shape identity guarantee (L226–231, incl. `_hoodie_*` meta columns and `drop.all.meta.fields` honoring) prevents `<SRC_K>` from flipping columns batch-to-batch. Merged-output schema story is spelled out in the schema-provider paragraph (L266–289). ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +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 Review Comment: Addressed in new **Section 5: Scale and parallelism** (L291–306). Per-source reads run sequentially on the driver — wall-clock is the **sum** of per-source read times, not the max. Expected N is small (2–5), one slow source stretches the batch (per-source `num_instants` guidance), large-N or truly-parallel use cases redirected to separate Streamer jobs + downstream merge. Parallel per-source reads deferred as opt-in future work behind a config, without a public-contract change (also documented in Alternatives Considered L629–633). ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +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 + 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). + +**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. + +### 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). +- **`SourceFormatAdapter`.** Processes error events per-dataset on the multi-dataset row path. + +### Sequence (per batch) + +``` +StreamSync.sync() + └─ HoodieIncrMultiSource.fetchNext() + ├─ read incrementally: db.orders (own ckpt), db.users (own ckpt), … → List<Dataset<Row>> + └─ InputBatch(batches=[...], checkpoint="db.orders=ts1,db.users=ts2") + └─ SqlFileBasedTransformer.apply(jsc, spark, getBatches(), props) // <SRC_0>,<SRC_1> → merged DF + └─ write merged DF + commit (single atomic commit; multi-table checkpoint persisted) +``` + +### Design decisions & risks + +- **User-owned merge semantics (non-goal to automate).** Delayed data in a source means **inner joins + can silently miss rows**; the recommended pattern is **OUTER JOIN** on incremental sources. Overlapping + columns and partial/nested-field updates are resolved with a **merge payload** (e.g. a generic merge + payload, or a custom `HoodieRecordPayload` that merges non-null fields) rather than last-writer-wins + overwrite — otherwise fields from an earlier upsert are lost. These are documented, not enforced. +- **Backward compatibility.** All Java API changes are additive; `getBatch()`, single-source sources, + and non-SQL transformers are unaffected. The checkpoint format degrades cleanly to legacy. +- **Checkpoint correctness across failures/re-runs** is the main implementation risk and is covered by + the lenient multi-table parser and by the test plan below. +- **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 + +- **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. Review Comment: Addressed as a full new **Alternatives Considered** section (L582–633) covering 7 options: N single-source jobs + downstream merge (status quo), snapshot-scan workaround, `InputBatch#getBatches()` + `HoodieIncrMultiSource` Source subclass (rejected in favor of the composition design — with three-breakage-point rationale), dedicated join/union source, transformer-only approach, extending `HoodieIncrSource` in place, parallel per-source reads on the driver (deferred). ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: Review Comment: The `extends Source` concern no longer applies — the composition design does not introduce a new `Source` subclass at all; each per-source read is a stock `HoodieIncrSource` inside its own ordinary single-source `SourceFormatAdapter`. The sanitization question is answered separately in Section 4 **Sanitization at write time** (L258–264): per-row sanitization is bypassed at the per-source read (rows are already Avro-conformant and it would otherwise run twice), but write-time sanitization on the merged output continues to flow through Streamer's normal write path when enabled — confirmed by a new integration test in the Test Plan. ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence + of a multi-source incremental read in Hudi Streamer. + +## 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` so it can return a multi-dataset `InputBatch` without per-row sanitization +side effects. Per batch it: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +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 + 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). + +**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. + +### 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). +- **`SourceFormatAdapter`.** Processes error events per-dataset on the multi-dataset row path. + +### Sequence (per batch) + +``` +StreamSync.sync() + └─ HoodieIncrMultiSource.fetchNext() + ├─ read incrementally: db.orders (own ckpt), db.users (own ckpt), … → List<Dataset<Row>> + └─ InputBatch(batches=[...], checkpoint="db.orders=ts1,db.users=ts2") + └─ SqlFileBasedTransformer.apply(jsc, spark, getBatches(), props) // <SRC_0>,<SRC_1> → merged DF + └─ write merged DF + commit (single atomic commit; multi-table checkpoint persisted) +``` + +### Design decisions & risks + +- **User-owned merge semantics (non-goal to automate).** Delayed data in a source means **inner joins + can silently miss rows**; the recommended pattern is **OUTER JOIN** on incremental sources. Overlapping + columns and partial/nested-field updates are resolved with a **merge payload** (e.g. a generic merge + payload, or a custom `HoodieRecordPayload` that merges non-null fields) rather than last-writer-wins + overwrite — otherwise fields from an earlier upsert are lost. These are documented, not enforced. +- **Backward compatibility.** All Java API changes are additive; `getBatch()`, single-source sources, + and non-SQL transformers are unaffected. The checkpoint format degrades cleanly to legacy. +- **Checkpoint correctness across failures/re-runs** is the main implementation risk and is covered by + the lenient multi-table parser and by the test plan below. +- **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 Review Comment: Addressed by the new **Failure and atomicity contract** section (L533–563). All-or-nothing boundary is explicit: any per-source read failure aborts the batch before the write, transformer exception aborts the batch, write is a single Hudi commit, and the composite checkpoint is written into that same commit's `deltastreamer.checkpoint.key` extra-metadata so per-source cursors advance atomically with the data. No partial-success mode by design. ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,250 @@ +<!-- + 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 + +- **Incremental-transformation ETLs** for data-freshness initiatives, currently blocked by the absence Review Comment: Motivating use cases expanded from one bullet to four concrete scenarios (L83–109): cross-entity enrichment (fact + dimension foreign-key join), consolidating sharded/regional upstreams into a unified table, SCD-style pipelines combining change stream with reference tables, and the compute-cost argument for the snapshot-scan workaround (which grows with retention window rather than with new data). ########## 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: Same as `@nsivabalan`'s ask #1. Motivating use cases now caveat with a pointer to the *Failure and atomicity contract* section (L92–93), which explicitly states the guarantee is transactional not distributed-snapshot and directs users needing row-level cross-source point-in-time consistency to layer their own as-of-time reads on top (L537–543). -- 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]
