hudi-agent commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3653916976
########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,686 @@ +<!-- + 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-source incremental reads** in Hudi Streamer via a **composition** +design: `InputBatch`, `Source`, and `SourceFormatAdapter` remain unchanged, and a new +`MultiSourceFormatAdapter` composes N ordinary per-source `SourceFormatAdapter`s (one per configured +Hudi table). `StreamSync.fetchNextBatchFromSource` branches once on adapter type at the top of the +method; the rest of the pipeline — schema deduction, error-event processing, write, commit — is +byte-for-byte identical for the single-source and multi-source paths. A new `MultiDatasetTransformer` +contract (implemented by `SqlFileBasedTransformer`) merges the per-source datasets with a single SQL +file referencing `<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible indexed, path-fingerprinted +checkpoint format** managed by `MultiTableCheckpointManager` records the per-source cursor set in a +single commit. The result is a single, atomic ingestion job — one commit, one checkpoint — for +related datasets, added without a public-API change to `InputBatch` or the `Source` hierarchy. + +## 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 (subject to the atomicity + contract in *Failure and atomicity* — this is transactional atomicity, not a distributed snapshot). +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Scope + +**v1 scope: Hudi-to-Hudi only.** `MultiSourceFormatAdapter`'s v1 implementation reads exclusively +from Hudi tables (per-source adapters wrap `HoodieIncrSource`). Extending multi-source semantics +to Kafka, S3/GCS events, JSON/DFS, JDBC, or other non-Hudi sources is out of scope for this RFC +and requires its own design work — specifically, per-source-type checkpoint serialization (Kafka +per-partition offsets, S3 event-queue positions, JDBC watermark columns) that does not fit the +`index=pathHash:instantTime` format used by `MultiTableCheckpointManager`. + +The composition design does not preclude future extension. `MultiSourceFormatAdapter` treats each +per-source `InputBatch<Dataset<Row>>` opaquely — every `SourceFormatAdapter` already normalizes +ROW / AVRO / JSON / PROTO to `Dataset<Row>` in `fetchNewDataInRowFormat` — so **homogeneous +multi-source of any type** (N Kafka, N JSON, N JDBC …) and **heterogeneous multi-source** (mixed +Hudi + Kafka + JSON in one pipeline) both fit the design architecturally. The outstanding work +per new source type is checkpoint encoding and its resume semantics, not new adapter plumbing. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). The design is **composition-first**: +`InputBatch`, `Source`, and `SourceFormatAdapter` are all unchanged; a new `MultiSourceFormatAdapter` +composes N per-source `SourceFormatAdapter`s (one per configured Hudi table); `StreamSync` branches +once on adapter type at the top of `fetchNextBatchFromSource`; the rest of the pipeline is identical +for single-source and multi-source. + +### 1. `MultiSourceFormatAdapter` (composition) + +A new class. **Composition, not inheritance** — `MultiSourceFormatAdapter` is *not* a subclass of +`SourceFormatAdapter`; it is a peer class that composes N single-source `SourceFormatAdapter` +instances. + +- Constructor takes `List<SourceFormatAdapter> perSourceAdapters` (one per configured index), a + `MultiTableCheckpointManager`, and the source-identifier registry (index → path, optional label). +- `fetchNewDataInRowFormat(Option<Checkpoint>, long)` → `List<InputBatch<Dataset<Row>>>`: + 1. Parses the composite `Checkpoint` via `MultiTableCheckpointManager` into `Map<Integer, Checkpoint>`. + 2. For each configured index `i`, dispatches the corresponding per-source `Checkpoint` + (or `Option.empty()` if new) to `perSourceAdapters.get(i).fetchNewDataInRowFormat(...)` — + sequentially on the driver. + 3. Collects the per-source `InputBatch<Dataset<Row>>` results in index order and returns the list. +- `fetchNewDataInAvroFormat(...)` throws `UnsupportedOperationException`. v1 is row-only. Non-ROW + source types produce `Dataset<Row>` inside their own `SourceFormatAdapter.fetchNewDataInRowFormat` + today; the multi adapter is source-type-agnostic on the row path. + +Per-source error events are handled inside each underlying `SourceFormatAdapter` — the existing +single-source code path, invoked N times. No branching inside `SourceFormatAdapter` is required. + +### 2. Multi-source configuration and identifiers + +Each source is identified by a **zero-based index** — the *durable* identifier used everywhere the +checkpoint, config keys, and SQL bindings reference a source. An optional human-readable `label` +exists purely for logs, metrics, and error messages and is **not** part of the checkpoint. + +``` +# How many Hudi sources are declared in this pipeline (triggers multi-source mode). +hoodie.streamer.source.multi.hudi.count=2 + +# Per-index configuration +hoodie.streamer.source.multi.hudi.0.path=/hudi/orders +hoodie.streamer.source.multi.hudi.0.num_instants=5 +hoodie.streamer.source.multi.hudi.0.missing_checkpoint_strategy=READ_LATEST +hoodie.streamer.source.multi.hudi.0.skip_on_schema_failure=false + +hoodie.streamer.source.multi.hudi.1.path=/hudi/users +hoodie.streamer.source.multi.hudi.1.num_instants=3 +hoodie.streamer.source.multi.hudi.1.missing_checkpoint_strategy=READ_LATEST + +# Optional non-durable label — used only in logs / metrics / error messages +hoodie.streamer.source.multi.hudi.0.label=orders +hoodie.streamer.source.multi.hudi.1.label=users +``` + +Design rationale for indexed identifiers (vs. dotted-name aliases with property-key mangling): + +- **No property-key collision by construction.** With a dotted-name scheme, `tables=db.orders,db_orders` + would mangle to the same property-key prefix (`...source.hudi.db_orders.*`) and one entry would + silently overwrite the other. Indices remove that failure mode entirely. +- **`label` is decoupled from durability.** Operators can rename freely for log/metric clarity + without touching the checkpoint or invalidating a running pipeline. +- **SQL bindings stay indexed** — `<SRC_0>`, `<SRC_1>`, … with `<SRC>` aliasing `<SRC_0>` for + backward compatibility with single-source `SqlFileBasedTransformer` SQL files. Silent misbinding + of `<SRC_N>` on a source reorder is caught at startup by the **checkpoint path-hash fingerprint** + (see *MultiTableCheckpointManager*), not by binding SQL by label. + +**Bootstrap trigger.** Presence of `hoodie.streamer.source.multi.hudi.count` with a value `> 0` +triggers multi-source mode: Streamer constructs a `MultiSourceFormatAdapter` wrapping N +`SourceFormatAdapter(HoodieIncrSource)` instances — one per index — instead of the ordinary +single-source `SourceFormatAdapter`. If `source.class` is also set, it must be a recognized +multi-source marker; otherwise the job fails at startup with a clear error, so operators are never +in a state where they think they set `source.class` but the count field was ignored (or vice versa). + +**Config compatibility.** All keys carry `hoodie.deltastreamer.*` alternatives for consistency with +the existing Streamer config surface. + +### 3. Per-source incremental reads + +Each per-source `SourceFormatAdapter` wraps a stock `HoodieIncrSource` configured from that index's +keys. That source runs the same per-table logic — `numInstantsPerFetch`, `missing_checkpoint_strategy`, +hollow-commit handling, timeline management, and "already caught up" detection — that a single-source +Streamer pipeline runs today. **No new `Source` subclass is introduced.** + +Per batch, `MultiSourceFormatAdapter`: + +1. Parses the composite checkpoint and dispatches per-source checkpoints in index order. +2. Reads sources **sequentially on the driver** (see *Scale and parallelism* below). +3. Returns the per-source `InputBatch<Dataset<Row>>` list in index order. Each entry is either: + - A dataset with new incremental rows for that source, or + - An **empty-but-schema'd** dataset (the source had no new instants in this window) — so + downstream `LEFT/FULL OUTER JOIN`s in the transformer SQL always resolve against a valid + relation and column references never explode on "unresolved column." + +### 4. Per-source schema resolution + +Each per-source `HoodieIncrSource` resolves its read schema independently against its own Hudi +table's metadata — heterogeneous or independently-evolving source schemas are supported directly. + +- A source with new commits: the read produces rows conformant with the current commit's schema. +- A source that is caught up: the read emits an empty `Dataset<Row>` built from that table's + current Hudi schema — **including the `_hoodie_*` meta columns and honoring + `hoodie.datasource.write.drop.all.meta.fields`** so the empty and non-empty shapes for the same + source are always identical. If they differed, `<SRC_K>` would change columns depending on + whether that source advanced in a given batch, and the merged output (plus the deduced write + schema) could flip batch-to-batch — a data-correctness hazard. +- Placeholder-schema resolution failure for any source: controlled by + `hoodie.streamer.source.multi.hudi.{index}.skip_on_schema_failure` (default `false` — fail loud + so a real failure does not silently advance the checkpoint). When set `true`, the per-source + read returns an empty dataset for that batch and emits a `<label-or-index>.schema_resolution_failure` + counter metric. + +**All-sources-failed under `skip_on_schema_failure=true`.** When every configured source has +`skip_on_schema_failure=true` and every source fails schema resolution in the same batch, the +merged transformer output is empty and the batch **does not commit** by default (respecting +Streamer's existing "commit on no-op batch" configuration) — no checkpoint advances, no timeline +noise, the schema-failure metric fires N times, and operator alerting on that metric is the +surface for detection. Users who have explicitly opted into committing empty batches (e.g. for +downstream freshness heartbeats) get an empty commit at the previous composite checkpoint. Either +way, no data is silently written and no cursor silently advances. The +`<label-or-index>.schema_resolution_failure` metric should be treated as a **hard-alert** signal, +not a soft one — a persistent non-zero value indicates a source is silently no-op'ing. + +**Checkpoint carry-forward invariant.** Whenever a source does not advance in a batch — whether +because it had no new data, because `skip_on_schema_failure=true` swallowed a real failure, or +because its previous entry was dropped by the lenient parser — its **previous checkpoint value Review Comment: 🤖 The invariant lists "its previous entry was dropped by the lenient parser" as a carry-forward case, but the committed checkpoint string is the only durable copy of that entry — and that's exactly what got corrupted/dropped. On resume there's nothing intact to carry forward, so this case still falls through to `missing_checkpoint_strategy` (READ_LATEST → history skip), unlike the "no new data" / `skip_on_schema_failure` cases where the value survives in the string. Is there a fallback here (e.g. walking back to an earlier commit's `deltastreamer.checkpoint.key`)? Otherwise the lenient-parser data-loss window isn't actually closed. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,686 @@ +<!-- + 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-source incremental reads** in Hudi Streamer via a **composition** +design: `InputBatch`, `Source`, and `SourceFormatAdapter` remain unchanged, and a new +`MultiSourceFormatAdapter` composes N ordinary per-source `SourceFormatAdapter`s (one per configured +Hudi table). `StreamSync.fetchNextBatchFromSource` branches once on adapter type at the top of the +method; the rest of the pipeline — schema deduction, error-event processing, write, commit — is +byte-for-byte identical for the single-source and multi-source paths. A new `MultiDatasetTransformer` +contract (implemented by `SqlFileBasedTransformer`) merges the per-source datasets with a single SQL +file referencing `<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible indexed, path-fingerprinted +checkpoint format** managed by `MultiTableCheckpointManager` records the per-source cursor set in a +single commit. The result is a single, atomic ingestion job — one commit, one checkpoint — for +related datasets, added without a public-API change to `InputBatch` or the `Source` hierarchy. + +## 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 (subject to the atomicity + contract in *Failure and atomicity* — this is transactional atomicity, not a distributed snapshot). +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Scope + +**v1 scope: Hudi-to-Hudi only.** `MultiSourceFormatAdapter`'s v1 implementation reads exclusively +from Hudi tables (per-source adapters wrap `HoodieIncrSource`). Extending multi-source semantics +to Kafka, S3/GCS events, JSON/DFS, JDBC, or other non-Hudi sources is out of scope for this RFC +and requires its own design work — specifically, per-source-type checkpoint serialization (Kafka +per-partition offsets, S3 event-queue positions, JDBC watermark columns) that does not fit the +`index=pathHash:instantTime` format used by `MultiTableCheckpointManager`. + +The composition design does not preclude future extension. `MultiSourceFormatAdapter` treats each +per-source `InputBatch<Dataset<Row>>` opaquely — every `SourceFormatAdapter` already normalizes +ROW / AVRO / JSON / PROTO to `Dataset<Row>` in `fetchNewDataInRowFormat` — so **homogeneous +multi-source of any type** (N Kafka, N JSON, N JDBC …) and **heterogeneous multi-source** (mixed +Hudi + Kafka + JSON in one pipeline) both fit the design architecturally. The outstanding work +per new source type is checkpoint encoding and its resume semantics, not new adapter plumbing. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). The design is **composition-first**: +`InputBatch`, `Source`, and `SourceFormatAdapter` are all unchanged; a new `MultiSourceFormatAdapter` +composes N per-source `SourceFormatAdapter`s (one per configured Hudi table); `StreamSync` branches +once on adapter type at the top of `fetchNextBatchFromSource`; the rest of the pipeline is identical +for single-source and multi-source. + +### 1. `MultiSourceFormatAdapter` (composition) + +A new class. **Composition, not inheritance** — `MultiSourceFormatAdapter` is *not* a subclass of +`SourceFormatAdapter`; it is a peer class that composes N single-source `SourceFormatAdapter` +instances. + +- Constructor takes `List<SourceFormatAdapter> perSourceAdapters` (one per configured index), a + `MultiTableCheckpointManager`, and the source-identifier registry (index → path, optional label). +- `fetchNewDataInRowFormat(Option<Checkpoint>, long)` → `List<InputBatch<Dataset<Row>>>`: + 1. Parses the composite `Checkpoint` via `MultiTableCheckpointManager` into `Map<Integer, Checkpoint>`. + 2. For each configured index `i`, dispatches the corresponding per-source `Checkpoint` + (or `Option.empty()` if new) to `perSourceAdapters.get(i).fetchNewDataInRowFormat(...)` — + sequentially on the driver. + 3. Collects the per-source `InputBatch<Dataset<Row>>` results in index order and returns the list. +- `fetchNewDataInAvroFormat(...)` throws `UnsupportedOperationException`. v1 is row-only. Non-ROW + source types produce `Dataset<Row>` inside their own `SourceFormatAdapter.fetchNewDataInRowFormat` + today; the multi adapter is source-type-agnostic on the row path. + +Per-source error events are handled inside each underlying `SourceFormatAdapter` — the existing +single-source code path, invoked N times. No branching inside `SourceFormatAdapter` is required. + +### 2. Multi-source configuration and identifiers + +Each source is identified by a **zero-based index** — the *durable* identifier used everywhere the +checkpoint, config keys, and SQL bindings reference a source. An optional human-readable `label` +exists purely for logs, metrics, and error messages and is **not** part of the checkpoint. + +``` +# How many Hudi sources are declared in this pipeline (triggers multi-source mode). +hoodie.streamer.source.multi.hudi.count=2 + +# Per-index configuration +hoodie.streamer.source.multi.hudi.0.path=/hudi/orders +hoodie.streamer.source.multi.hudi.0.num_instants=5 +hoodie.streamer.source.multi.hudi.0.missing_checkpoint_strategy=READ_LATEST +hoodie.streamer.source.multi.hudi.0.skip_on_schema_failure=false + +hoodie.streamer.source.multi.hudi.1.path=/hudi/users +hoodie.streamer.source.multi.hudi.1.num_instants=3 +hoodie.streamer.source.multi.hudi.1.missing_checkpoint_strategy=READ_LATEST + +# Optional non-durable label — used only in logs / metrics / error messages +hoodie.streamer.source.multi.hudi.0.label=orders +hoodie.streamer.source.multi.hudi.1.label=users +``` + +Design rationale for indexed identifiers (vs. dotted-name aliases with property-key mangling): + +- **No property-key collision by construction.** With a dotted-name scheme, `tables=db.orders,db_orders` + would mangle to the same property-key prefix (`...source.hudi.db_orders.*`) and one entry would + silently overwrite the other. Indices remove that failure mode entirely. +- **`label` is decoupled from durability.** Operators can rename freely for log/metric clarity + without touching the checkpoint or invalidating a running pipeline. +- **SQL bindings stay indexed** — `<SRC_0>`, `<SRC_1>`, … with `<SRC>` aliasing `<SRC_0>` for + backward compatibility with single-source `SqlFileBasedTransformer` SQL files. Silent misbinding + of `<SRC_N>` on a source reorder is caught at startup by the **checkpoint path-hash fingerprint** + (see *MultiTableCheckpointManager*), not by binding SQL by label. + +**Bootstrap trigger.** Presence of `hoodie.streamer.source.multi.hudi.count` with a value `> 0` +triggers multi-source mode: Streamer constructs a `MultiSourceFormatAdapter` wrapping N +`SourceFormatAdapter(HoodieIncrSource)` instances — one per index — instead of the ordinary +single-source `SourceFormatAdapter`. If `source.class` is also set, it must be a recognized +multi-source marker; otherwise the job fails at startup with a clear error, so operators are never +in a state where they think they set `source.class` but the count field was ignored (or vice versa). + +**Config compatibility.** All keys carry `hoodie.deltastreamer.*` alternatives for consistency with +the existing Streamer config surface. + +### 3. Per-source incremental reads + +Each per-source `SourceFormatAdapter` wraps a stock `HoodieIncrSource` configured from that index's +keys. That source runs the same per-table logic — `numInstantsPerFetch`, `missing_checkpoint_strategy`, +hollow-commit handling, timeline management, and "already caught up" detection — that a single-source +Streamer pipeline runs today. **No new `Source` subclass is introduced.** + +Per batch, `MultiSourceFormatAdapter`: + +1. Parses the composite checkpoint and dispatches per-source checkpoints in index order. +2. Reads sources **sequentially on the driver** (see *Scale and parallelism* below). +3. Returns the per-source `InputBatch<Dataset<Row>>` list in index order. Each entry is either: + - A dataset with new incremental rows for that source, or + - An **empty-but-schema'd** dataset (the source had no new instants in this window) — so + downstream `LEFT/FULL OUTER JOIN`s in the transformer SQL always resolve against a valid + relation and column references never explode on "unresolved column." + +### 4. Per-source schema resolution + +Each per-source `HoodieIncrSource` resolves its read schema independently against its own Hudi +table's metadata — heterogeneous or independently-evolving source schemas are supported directly. + +- A source with new commits: the read produces rows conformant with the current commit's schema. +- A source that is caught up: the read emits an empty `Dataset<Row>` built from that table's + current Hudi schema — **including the `_hoodie_*` meta columns and honoring + `hoodie.datasource.write.drop.all.meta.fields`** so the empty and non-empty shapes for the same + source are always identical. If they differed, `<SRC_K>` would change columns depending on + whether that source advanced in a given batch, and the merged output (plus the deduced write + schema) could flip batch-to-batch — a data-correctness hazard. +- Placeholder-schema resolution failure for any source: controlled by + `hoodie.streamer.source.multi.hudi.{index}.skip_on_schema_failure` (default `false` — fail loud + so a real failure does not silently advance the checkpoint). When set `true`, the per-source + read returns an empty dataset for that batch and emits a `<label-or-index>.schema_resolution_failure` + counter metric. + +**All-sources-failed under `skip_on_schema_failure=true`.** When every configured source has +`skip_on_schema_failure=true` and every source fails schema resolution in the same batch, the +merged transformer output is empty and the batch **does not commit** by default (respecting +Streamer's existing "commit on no-op batch" configuration) — no checkpoint advances, no timeline +noise, the schema-failure metric fires N times, and operator alerting on that metric is the +surface for detection. Users who have explicitly opted into committing empty batches (e.g. for +downstream freshness heartbeats) get an empty commit at the previous composite checkpoint. Either +way, no data is silently written and no cursor silently advances. The +`<label-or-index>.schema_resolution_failure` metric should be treated as a **hard-alert** signal, +not a soft one — a persistent non-zero value indicates a source is silently no-op'ing. + +**Checkpoint carry-forward invariant.** Whenever a source does not advance in a batch — whether +because it had no new data, because `skip_on_schema_failure=true` swallowed a real failure, or +because its previous entry was dropped by the lenient parser — its **previous checkpoint value +is carried forward unchanged** into the next committed multi-source checkpoint. The composite +checkpoint is always rebuilt from a full N-entry map covering every currently-configured source, +never from only the sources that advanced this batch. This closes the silent-data-loss window +that "no checkpoint → apply `missing_checkpoint_strategy` → `READ_LATEST` skips history" would +otherwise open on subsequent batches for a source that failed schema resolution once. + +**Sanitization at write time.** `RowSource#fetchNextBatch`'s per-row Avro-field-name sanitization +is bypassed at the per-source read (rows read from a Hudi table are already conformant with that +table's Avro schema, so sanitization would be redundant and would run twice — once per source, +once after the merge). Sanitization **at write time** on the merged transformer output continues +to flow through Streamer's normal write path when the operator has enabled it — this bypass +touches only the read side. This is covered by an integration test that writes the merged output +with sanitization enabled and asserts the writer's sanitization codepath was taken. + +**Schema providers.** The multi-source composition deliberately does not accept per-source +`SchemaProvider` configuration in v1. Every source is a Hudi table with a well-defined Avro +schema in its own metadata; the read path derives each source dataset's shape from that. Unlike +single-source Streamer inputs, there is no "source cannot describe itself" case to solve. +Per-source shape normalization — renames, casts, null-fills, dropping/pinning columns during +upstream evolution — is expressed in the transformer SQL against `<SRC_0>` / `<SRC_1>` / …. +`SchemaProvider` targets Avro conformance, not arbitrary row-shape coercion, so SQL is the right +layer. + +The merged output continues to honor the existing top-level `hoodie.streamer.schemaprovider.class`. +When present with a non-null target schema it wins for the write, exactly as in single-source +Streamer today (matching Sub-branch A of `getDeducedSchemaProvider`); when absent, the writer +schema is deduced from the merged transformer output reconciled against the latest target-table +schema. Registry-backed providers (`SchemaRegistryProvider` and variants) are not applicable to +the multi-source read path — Hudi tables self-describe, the same as with `HoodieIncrSource` today. + +**Startup rejection of per-source schema-provider keys.** Per-source schema-provider keys +(e.g. `hoodie.streamer.source.multi.hudi.{index}.schemaprovider.class`) are rejected at startup +with a clear error so operators pattern-matching from single-source Streamer configuration are +not surprised by a silent-ignore. + +**Future work.** If a real user need for per-source schema providers surfaces (e.g. registry-backed +evolution across source *and* target under one Streamer job), a follow-up RFC can add it; the +composition-based design does not preclude it. + +### 5. Scale and parallelism + +Per-source reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at +a time, not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-source +read times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads + keep the design simple and avoid contention on the shared timeline server and driver-side + timeline caches without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole + batch, so per-source `num_instants` should be sized to keep the slowest source within an + acceptable per-batch budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-source + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + feature is not a replacement for that at large fan-out. Parallel per-source reads can be added + later behind an opt-in config without changing the public contract if usage patterns demand it. + +### 6. `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 the per-index source declaration; an entry is always present for +each configured source (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 + `<SRC_0>` (backward compatible with single-source SQL files). +- 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. +- Validates inputs: rejects null/empty dataset lists and any dangling `<SRC_N>` placeholder that + has no corresponding dataset (typo guard), with clear error messages. + +**Temp-view namespace ownership and cleanup lifetime.** + +- **Cleanup runs in a `finally` block per batch** — the tracked view names accumulated during a + single `apply()` invocation are dropped at the end of that invocation whether it succeeds or + throws. A driver crash mid-batch will leak any views registered so far; this is accepted, + because on the next driver start every view registered in the leaked batch has a fresh UUID + and therefore cannot collide with anything a subsequent run registers. +- **Source-view names cannot collide.** `HOODIE_SRC_TMP_TABLE_<uuid>` names are UUID-mangled per + invocation. +- **User-created view names are not UUID-mangled.** `CREATE OR REPLACE TEMPORARY VIEW my_dim` in + the user's SQL creates a view called `my_dim` in the Spark session's global temp-view namespace. + The transformer takes **exclusive ownership of the Spark session's temp-view namespace during + `apply()`** — any pre-existing view with the same name will be replaced by the transformer's + `CREATE OR REPLACE` and cleaned up at the end of the batch. Operators who manage temp views + outside Streamer should namespace them (e.g. `myapp_dim`) to avoid this. This is documented as + a hard rule in the transformer's javadoc, not a soft convention. + +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; +``` + +### 7. `MultiTableCheckpointManager` + +Parses, formats, and manipulates per-source checkpoints stored in `deltastreamer.checkpoint.key`. +The checkpoint is **indexed with a path fingerprint**: + +``` +Multi-source : "0=abc123:20250606182826197,1=def456:20250606182830145" +Single/legacy: "20250606182826197" +``` + +Where `abc123` / `def456` are short truncated SHA-256 hashes of the source's base path — the +**path fingerprint** — and the value after `:` is the source's next instant to read (the +`HoodieIncrSource` checkpoint format is unchanged). + +- **Path fingerprint guards against silent misconfiguration.** On resume, the manager computes + the fingerprint of each currently-configured index's path and compares it to the stored + fingerprint. On mismatch, the job **fails at startup** with a clear diagnostic naming the + affected index and the two paths (previous vs. current), rather than silently resuming the + new source from an unrelated table's checkpoint. This catches source reorder and path-replace + at an existing index (see *Source-list evolution semantics*). +- **Legacy-format detection.** Any value containing `=` is parsed as multi-source. A bare value + with no `=` is treated as a legacy single-source checkpoint and migrated on the next commit + as `0=<computed-hash>:<timestamp>` — clean upgrade from an existing single-source job with no + manual surgery. +- **Deterministic serialization.** The formatted output uses a deterministic key order by + ascending index (backed by a `TreeMap`) so the same set of per-source checkpoints always + serializes to the same string, making commits and diffs reproducible. +- **Lenient parsing** preserves valid entries while skipping malformed ones. A skipped entry + is logged as `WARN`, and the source emits a `multi_source_checkpoint.malformed_entries` + counter metric so the drop is observable in monitoring, not silent. **Skipped entries are + always paired with the checkpoint carry-forward invariant** (see *Per-source schema + resolution*): the previous batch's committed checkpoint is the canonical source, and any + index whose entry the parser dropped is rebuilt from its prior value rather than falling + through to `missing_checkpoint_strategy`. This closes the silent-data-loss window that + "no checkpoint → `READ_LATEST` → skip history" would otherwise open. The metric should be + alerted on with threshold `> 0` in any pipeline where re-reading history is not equivalent + to a resume. + +**Operational checkpoint override / reset in multi-source mode.** + +Streamer's `--checkpoint <value>` override and `hoodie.deltastreamer.checkpoint.reset_key` accept +a **fully-formed composite** value in multi-source mode: + +``` +--checkpoint "0=abc123:20250606182826197,1=def456:20250606182830145" +``` + +A bare-timestamp override (e.g. `--checkpoint 20250606182826197`) is **rejected at startup** in +multi-source mode with a clear error message rather than silently mapped to a single default index +and dropping the other sources' cursors. To reset only one source, the operator provides the +composite string with only that source's timestamp updated (the other entries carried forward +from the current checkpoint), which the manager accepts and validates against configured paths. + +### 8. Source-list evolution semantics + +The following operational events on a running multi-source pipeline are handled as: + +- **Adding a source** at a new index (increasing `count` and adding the per-index keys): no prior + checkpoint entry exists for the new index, so the source applies its configured + `missing_checkpoint_strategy`. Because the default `READ_LATEST` implies a silent history skip, + operators seeding a new source should either (a) accept the skip, (b) set the new index's + `missing_checkpoint_strategy=READ_UPTO_LATEST_COMMIT` for its first batch, or (c) hand-seed a + starting instant via `--checkpoint`. +- **Removing a source** at an index (reducing `count` and removing the per-index keys): the stale Review Comment: 🤖 "Removing a source" says the stale entry is just pruned with a WARN and no operator intervention. But indices look contiguous zero-based (`.0.path`, `.1.path`, `count=N`), so removing a non-last source — e.g. index 1 of {0,1,2} — renumbers the survivors, and old index 2's path now sits at index 1. Its stored path fingerprint under index 1 won't match, so the fingerprint guard would fail the job at startup (same as reorder), contradicting the clean-prune description. Should removal be restricted to the highest index, or does a middle removal need a `--checkpoint` override like reorder? <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]
