hudi-agent commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3589165479
########## 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: 🤖 How does this handle schema evolution across the N source tables over time? With a single SchemaProvider feeding one commit, it isn't clear what happens when one source table evolves its schema mid-stream while others don't, or how the empty-but-schema'd placeholder dataset stays in sync with an evolved source schema. It might be worth a dedicated subsection on per-table schema resolution and target-schema reconciliation. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 The lenient parser silently skipping a malformed entry seems to interact dangerously with the per-table missing-checkpoint strategy. If a table's entry is dropped, that table has no checkpoint and falls back to its `missing_checkpoint_strategy` — and the default `READ_LATEST` would skip all unprocessed data (silent data loss), which contradicts the fail-loud philosophy used elsewhere in this design. Could you spell out exactly what happens to a table whose entry is skipped? @nsivabalan could you weigh in on whether silently dropping a checkpoint entry is safe here versus failing the batch? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 "Tables ... never block one another" reads as a concurrency claim, but the sequence suggests the N incremental reads happen within a single batch on the driver. Are the per-table reads issued in parallel or sequentially, and is there an intended upper bound on N? A short note on the performance/resource profile as N grows (and whether one slow/large source stretches the whole batch) would strengthen the scalability story. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 The RFC doesn't include an Alternatives Considered section. It would help to briefly document approaches that were evaluated and rejected — e.g. orchestrating multiple single-source jobs with multi-writer/coordinated commits, a dedicated join/union source, or handling this at the transformer layer only — and why a new source plus InputBatch change is the preferred path. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 Extending `Source` directly to skip per-row sanitization is a notable departure from `RowSource`. What sanitization is being bypassed (e.g. invalid/duplicate field-name handling), and could skipping it produce rows that later fail on write or violate downstream assumptions? It might be worth clarifying whether sanitization is re-applied after the transformer merge instead of dropped entirely. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 The motivation is well argued conceptually, but the Motivating use cases section is a single fairly abstract bullet. Could you add one or two concrete scenarios (real tables/joins) and, if available, any evidence of demand — issue reports, scale figures, or the compute cost of the current snapshot-read workaround — to strengthen the case that this is worth the added surface area? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## 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: 🤖 Could you expand on the failure/atomicity semantics within a batch? Specifically: if the incremental read of table K succeeds but table K+1's read (or the merge/write) fails, is the entire batch aborted with no checkpoints advanced, and are all per-table checkpoints persisted atomically together with the single commit? Making the all-or-nothing boundary explicit would clarify the core atomicity guarantee this RFC is built on. <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]
