ashokkumar-allu commented on code in PR #19299: URL: https://github.com/apache/hudi/pull/19299#discussion_r3650976819
########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,385 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-108: Multi-dataset incremental reads in Hudi Streamer + +## Proposers + +- @ashokkumar-allu + +## Approvers + +- @nsivabalan + +## Status + +Issue: https://github.com/apache/hudi/issues/19281 + +> Please keep the status updated in `rfc/README.md`. + +## Abstract + +Apache Hudi Streamer (DeltaStreamer) can read incrementally from exactly one source per job. Pipelines +that need to combine incremental changes from **several Hudi tables** into one target table must +therefore run one Streamer job per source and merge the results downstream, or fall back to expensive +full-snapshot reads for all-but-one table. The multi-job approach has no cross-source transaction +boundary, so the tables it stitches together drift in time, produce incomplete/inconsistent joined +output, and require backfills to repair; it also multiplies watermark and monitoring overhead. + +This RFC proposes **native multi-table incremental reads** in Hudi Streamer: a new source +(`HoodieIncrMultiSource`) that reads N Hudi tables incrementally in a single batch with **independent +per-table checkpoints**, a **multi-dataset transformer** contract (`MultiDatasetTransformer`, +implemented by `SqlFileBasedTransformer`) that merges the datasets with a single SQL file referencing +`<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible multi-table checkpoint format** managed by +`MultiTableCheckpointManager`. The result is a single, atomic ingestion job — one commit, one +checkpoint — for related datasets. + +## Background + +### Current single-source flow + +``` +Streamer CLI + └─ StreamSync.sync() + └─ SourceFormatAdapter (adapts ROW / JSON / AVRO to a common shape) + └─ HoodieIncrSource (incremental read of ONE Hudi table via + hoodie.deltastreamer.source.hoodieincr.path) + └─ Transformer (single Dataset<Row> in → Dataset<Row> out) + └─ write + commit (checkpoint = single timestamp, e.g. "20250606182826197") +``` + +`InputBatch` carries a single `Option<T>` payload and one checkpoint string. Checkpoints are opaque +single timestamps stored in `deltastreamer.checkpoint.key` in the Streamer commit metadata. + +### Why this is insufficient for multi-source pipelines + +- **Data inconsistency / silent failures.** N independent jobs advance at different rates. A join of + Table A (at T+10) and Table B (at T+7) yields incomplete or misleading data with no error raised — + discovered only later, and repaired only with backfills. +- **No atomicity across sources.** There is no shared transaction boundary. If one job fails after + others commit, the unified downstream table is left partially updated. +- **Operational overhead.** Each job maintains its own watermark (watermark drift complicates replay + and debugging) and its own monitoring/alerting surface. +- **Expensive workaround.** Reading one table incrementally while snapshot-reading the last X days of + every other table consumes far more compute than a true incremental read. + +### Motivating use cases + +- **Cross-entity enrichment for analytics or serving.** A downstream table is populated by joining + two Hudi tables — e.g. a fact table (transactions, events, clicks, orders) with a dimension table + (accounts, users, products) on a foreign key. Both sources ingest continuously from different + upstreams and each is already a Hudi table. When run as two independent Streamer jobs plus a + downstream merge, the merge step periodically observes fact rows whose dimension row has not yet + been ingested and either drops them (inner join) or emits rows with null dimension fields (outer + join) that must be repaired on a later run. A native multi-source incremental read advances both + tables under one commit, so the join always sees a consistent slice. +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). Four building blocks plus wiring. + +### 1. Multi-dataset `InputBatch` + +`InputBatch<T>` gains a `List<T> batches` alongside the existing `Option<T> batch`: + +- New constructor `InputBatch(List<T> batches, String checkpointForNextBatch, SchemaProvider)`. +- `getBatches(): List<T>` returns all datasets (defensive copy). +- `getBatch(): Option<T>` is retained for single-dataset sources and now **fails loud** — it throws + via `ValidationUtils.checkState` if more than one batch is present, so a multi-dataset batch can + never be silently narrowed to one dataset. For single-dataset construction, `getBatches()` returns + a singleton and `getBatch()` returns that element (unchanged behavior). + +This keeps every existing single-source source and transformer working unchanged. + +### 2. `HoodieIncrMultiSource` + +A new source extending `Source<Dataset<Row>>` (SourceType `ROW`). It extends `Source` directly rather +than `RowSource` because `RowSource#fetchNextBatch` calls `SanitizationUtils` on the returned dataset +(masking Avro-invalid field names) before wrapping the result in an `InputBatch` — that per-row scan +would (a) run twice, once here and again after the transformer merges the datasets, and (b) only +supports a single `Option<Dataset<Row>>` payload, not the multi-dataset list this source needs. +Rows read from a Hudi table are already conformant with that table's Avro schema, so this bypass is +safe; sanitization still applies at write time to the merged transformer output when the user has +enabled it. Per batch the source: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +2. Iterates the parsed tables **sequentially on the driver** and performs a `HoodieIncrSource`-equivalent + incremental read for each — its own `numInstantsPerFetch`, missing-checkpoint strategy, hollow-commit + handling, timeline management, and "already caught up" detection. Tables that ingest at different + rates never block one another; see *Scale and parallelism* below for the wall-clock implications. +3. Returns a dataset for **every** configured table, in declaration order — an empty dataset carrying + the Hudi table's schema when a table has no new incremental data — so downstream `LEFT/FULL OUTER + JOIN`s always resolve against a valid relation. +4. Packages all datasets into one `InputBatch` via the multi-dataset constructor, with the combined + multi-table checkpoint (or a plain timestamp when multi-table checkpointing is disabled). The + `InputBatch` carries a `NullSchemaProvider` — Streamer infers the write schema from the merged + transformer output rather than from any single input source (see *Per-table schema resolution*). + +#### Per-table schema resolution + +Each source table's read schema is resolved independently against that table's own Hudi metadata, +so heterogeneous or independently-evolving source schemas are supported directly: + +- A table with new commits: the read produces rows conformant with the current commit's schema. +- A table that is caught up or empty: the source emits an empty `Dataset<Row>` built from that table's + current Hudi schema, so downstream joins and column references remain valid rather than exploding + on an "unresolved column" error. +- Placeholder-schema resolution failure for any table: controlled by + `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` (default `false` — fail loud so a real + failure does not silently advance the checkpoint; `true` returns an empty dataset for that table + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `hoodie.streamer.source.hudi.{table}.num_instants` | `HoodieIncrSourceConfig.NUM_INSTANTS_PER_FETCH` default | Per-table instants per fetch | +| `hoodie.streamer.source.hudi.{table}.missing_checkpoint_strategy` | `READ_LATEST` | `READ_LATEST` \| `READ_UPTO_LATEST_COMMIT` | +| `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` | `false` | On placeholder-schema resolution failure: `false` fails loud (checkpoint not advanced past a real failure); `true` returns an empty dataset and emits a `<table>.schema_resolution_failure` metric | +| `hoodie.streamer.source.hudi.{table}.partition.fields` / `.partition.extractor.class` | — | Optional per-table partitioning | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`SqlFileBasedTransformer` implements it: + +- Registers each input dataset as a unique Spark temp view (`HOODIE_SRC_TMP_TABLE_<uuid>`). +- SQL references datasets by index — `<SRC_0>`, `<SRC_1>`, … — with `<SRC>` kept as an alias for the Review Comment: Addressed structurally rather than by rebinding to labels — the **path fingerprint** in the composite checkpoint catches silent reorders/replaces at startup with a clear diagnostic naming the affected index and both paths (Section 2 L189–191, Section 8 L423–428). Reordering therefore requires an explicit `--checkpoint` override presenting the new correct mapping. Labels are decoupled from durability (Section 2 L186–187) so operators can rename freely without touching the checkpoint. ########## rfc/rfc-108/rfc-108.md: ########## @@ -0,0 +1,385 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +# RFC-108: Multi-dataset incremental reads in Hudi Streamer + +## Proposers + +- @ashokkumar-allu + +## Approvers + +- @nsivabalan + +## Status + +Issue: https://github.com/apache/hudi/issues/19281 + +> Please keep the status updated in `rfc/README.md`. + +## Abstract + +Apache Hudi Streamer (DeltaStreamer) can read incrementally from exactly one source per job. Pipelines +that need to combine incremental changes from **several Hudi tables** into one target table must +therefore run one Streamer job per source and merge the results downstream, or fall back to expensive +full-snapshot reads for all-but-one table. The multi-job approach has no cross-source transaction +boundary, so the tables it stitches together drift in time, produce incomplete/inconsistent joined +output, and require backfills to repair; it also multiplies watermark and monitoring overhead. + +This RFC proposes **native multi-table incremental reads** in Hudi Streamer: a new source +(`HoodieIncrMultiSource`) that reads N Hudi tables incrementally in a single batch with **independent +per-table checkpoints**, a **multi-dataset transformer** contract (`MultiDatasetTransformer`, +implemented by `SqlFileBasedTransformer`) that merges the datasets with a single SQL file referencing +`<SRC_0>`, `<SRC_1>`, …, and a **backward-compatible multi-table checkpoint format** managed by +`MultiTableCheckpointManager`. The result is a single, atomic ingestion job — one commit, one +checkpoint — for related datasets. + +## Background + +### Current single-source flow + +``` +Streamer CLI + └─ StreamSync.sync() + └─ SourceFormatAdapter (adapts ROW / JSON / AVRO to a common shape) + └─ HoodieIncrSource (incremental read of ONE Hudi table via + hoodie.deltastreamer.source.hoodieincr.path) + └─ Transformer (single Dataset<Row> in → Dataset<Row> out) + └─ write + commit (checkpoint = single timestamp, e.g. "20250606182826197") +``` + +`InputBatch` carries a single `Option<T>` payload and one checkpoint string. Checkpoints are opaque +single timestamps stored in `deltastreamer.checkpoint.key` in the Streamer commit metadata. + +### Why this is insufficient for multi-source pipelines + +- **Data inconsistency / silent failures.** N independent jobs advance at different rates. A join of + Table A (at T+10) and Table B (at T+7) yields incomplete or misleading data with no error raised — + discovered only later, and repaired only with backfills. +- **No atomicity across sources.** There is no shared transaction boundary. If one job fails after + others commit, the unified downstream table is left partially updated. +- **Operational overhead.** Each job maintains its own watermark (watermark drift complicates replay + and debugging) and its own monitoring/alerting surface. +- **Expensive workaround.** Reading one table incrementally while snapshot-reading the last X days of + every other table consumes far more compute than a true incremental read. + +### Motivating use cases + +- **Cross-entity enrichment for analytics or serving.** A downstream table is populated by joining + two Hudi tables — e.g. a fact table (transactions, events, clicks, orders) with a dimension table + (accounts, users, products) on a foreign key. Both sources ingest continuously from different + upstreams and each is already a Hudi table. When run as two independent Streamer jobs plus a + downstream merge, the merge step periodically observes fact rows whose dimension row has not yet + been ingested and either drops them (inner join) or emits rows with null dimension fields (outer + join) that must be repaired on a later run. A native multi-source incremental read advances both + tables under one commit, so the join always sees a consistent slice. +- **Consolidating multiple partitioned or sharded upstreams into one target table.** When a logical + entity is split across multiple Hudi tables — for example one table per shard, per region, or per + legacy vs. new system during a data-platform migration — a downstream consumer typically wants a + single unified view. Running one Streamer job per upstream leaves the unified table stale for + whichever upstream lagged in the most recent window; a single-batch multi-source read makes the + unified table advance atomically for all upstreams. +- **Slowly-changing-dimension (SCD) style pipelines.** A pipeline that maintains an SCD table by + combining a change stream from a source-of-truth table with reference/lookup tables benefits from + reading all inputs in the same batch — inconsistent temporal cuts between the change stream and + the lookup tables are a common source of incorrect history rows. +- **Compute cost of the current workaround.** Today, teams that need cross-source consistency fall + back to reading one table incrementally and snapshot-scanning the last N days of every other + table on every batch. That snapshot scan grows with the retention window rather than with the + new data, so a Streamer batch that would ingest a few thousand rows of true delta ends up + reading many gigabytes per additional source — every batch. The proposed source removes that + amplification because every source is read incrementally. + +## Implementation + +The change is confined to `hudi-utilities` (Hudi Streamer). Four building blocks plus wiring. + +### 1. Multi-dataset `InputBatch` + +`InputBatch<T>` gains a `List<T> batches` alongside the existing `Option<T> batch`: + +- New constructor `InputBatch(List<T> batches, String checkpointForNextBatch, SchemaProvider)`. +- `getBatches(): List<T>` returns all datasets (defensive copy). +- `getBatch(): Option<T>` is retained for single-dataset sources and now **fails loud** — it throws + via `ValidationUtils.checkState` if more than one batch is present, so a multi-dataset batch can + never be silently narrowed to one dataset. For single-dataset construction, `getBatches()` returns + a singleton and `getBatch()` returns that element (unchanged behavior). + +This keeps every existing single-source source and transformer working unchanged. + +### 2. `HoodieIncrMultiSource` + +A new source extending `Source<Dataset<Row>>` (SourceType `ROW`). It extends `Source` directly rather +than `RowSource` because `RowSource#fetchNextBatch` calls `SanitizationUtils` on the returned dataset +(masking Avro-invalid field names) before wrapping the result in an `InputBatch` — that per-row scan +would (a) run twice, once here and again after the transformer merges the datasets, and (b) only +supports a single `Option<Dataset<Row>>` payload, not the multi-dataset list this source needs. +Rows read from a Hudi table are already conformant with that table's Avro schema, so this bypass is +safe; sanitization still applies at write time to the merged transformer output when the user has +enabled it. Per batch the source: + +1. Parses `hoodie.streamer.source.multi.hudi.tables` (ordered, comma-separated) into per-table configs. +2. Iterates the parsed tables **sequentially on the driver** and performs a `HoodieIncrSource`-equivalent + incremental read for each — its own `numInstantsPerFetch`, missing-checkpoint strategy, hollow-commit + handling, timeline management, and "already caught up" detection. Tables that ingest at different + rates never block one another; see *Scale and parallelism* below for the wall-clock implications. +3. Returns a dataset for **every** configured table, in declaration order — an empty dataset carrying + the Hudi table's schema when a table has no new incremental data — so downstream `LEFT/FULL OUTER + JOIN`s always resolve against a valid relation. +4. Packages all datasets into one `InputBatch` via the multi-dataset constructor, with the combined + multi-table checkpoint (or a plain timestamp when multi-table checkpointing is disabled). The + `InputBatch` carries a `NullSchemaProvider` — Streamer infers the write schema from the merged + transformer output rather than from any single input source (see *Per-table schema resolution*). + +#### Per-table schema resolution + +Each source table's read schema is resolved independently against that table's own Hudi metadata, +so heterogeneous or independently-evolving source schemas are supported directly: + +- A table with new commits: the read produces rows conformant with the current commit's schema. +- A table that is caught up or empty: the source emits an empty `Dataset<Row>` built from that table's + current Hudi schema, so downstream joins and column references remain valid rather than exploding + on an "unresolved column" error. +- Placeholder-schema resolution failure for any table: controlled by + `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` (default `false` — fail loud so a real + failure does not silently advance the checkpoint; `true` returns an empty dataset for that table + and emits a `<table>.schema_resolution_failure` metric). +- The target write schema is not tied to any single source: `HoodieIncrMultiSource` attaches a + `NullSchemaProvider` to the `InputBatch`, and Streamer resolves the final write schema from the + merged `Dataset<Row>` returned by the `MultiDatasetTransformer`. This lets the user's SQL freely + add/drop/rename columns across source-schema versions without needing a schema-provider override. + +#### Scale and parallelism + +Per-table reads run **sequentially on the driver** — one Hudi `beginInstant..endInstant` read at a time, +not `N` reads in parallel. Wall-clock time for a batch is therefore the **sum** of per-table read +times, not the maximum. Practical implications: + +- Expected `N` for this feature is small — typically 2–5 related tables — so sequential reads keep the + design simple and avoid contention on the shared timeline server and driver-side timeline caches + without materially hurting throughput. +- One slow source (e.g. a table with a large backlog or many small files) stretches the whole batch, + so per-table `num_instants` should be sized to keep the slowest source within an acceptable per-batch + budget. +- If a pipeline needs cross-source consistency across a large `N` **or** truly parallel per-table + reads, the right tradeoff today is still separate Streamer jobs plus a downstream merge — this + source is not a replacement for that at large fan-out. Parallel per-table reads can be added later + behind an opt-in config without changing the public contract if usage patterns demand it. + +**Configuration** (`HoodieIncrMultiSourceConfig`, group `HUDI_STREAMER` / `DELTA_STREAMER_SOURCE`): + +| Key | Default | Purpose | +|-----|---------|---------| +| `hoodie.streamer.source.multi.hudi.tables` | (required) | Ordered, comma-separated source table names | +| `hoodie.streamer.source.multi.hudi.checkpoint.enable` | `true` | Multi-table checkpoint format; `false` → legacy single-checkpoint behavior | +| `hoodie.streamer.source.hudi.{table}.path` | (required) | Per-table base path | +| `hoodie.streamer.source.hudi.{table}.num_instants` | `HoodieIncrSourceConfig.NUM_INSTANTS_PER_FETCH` default | Per-table instants per fetch | +| `hoodie.streamer.source.hudi.{table}.missing_checkpoint_strategy` | `READ_LATEST` | `READ_LATEST` \| `READ_UPTO_LATEST_COMMIT` | +| `hoodie.streamer.source.hudi.{table}.skip_on_schema_failure` | `false` | On placeholder-schema resolution failure: `false` fails loud (checkpoint not advanced past a real failure); `true` returns an empty dataset and emits a `<table>.schema_resolution_failure` metric | +| `hoodie.streamer.source.hudi.{table}.partition.fields` / `.partition.extractor.class` | — | Optional per-table partitioning | + +All keys carry `hoodie.deltastreamer.*` alternatives for compatibility. **Table-name convention:** a +dotted name (`db.orders`) has its dots replaced by underscores when building property keys +(`...source.hudi.db_orders.path`) so property-file parsers don't split them; the original dotted name +is always used inside the checkpoint map. + +### 3. `MultiDatasetTransformer` and `SqlFileBasedTransformer` + +`MultiDatasetTransformer extends Transformer` adds: + +```java +Dataset<Row> apply(JavaSparkContext jsc, SparkSession spark, + List<Dataset<Row>> datasets, TypedProperties props); +``` + +Contract: list order matches `...source.multi.hudi.tables`; an entry is always present for each table +(empty-but-schema'd when no new data); the single-dataset `Transformer#apply` remains implemented so +the transformer still works in single-source pipelines. + +`SqlFileBasedTransformer` implements it: + +- Registers each input dataset as a unique Spark temp view (`HOODIE_SRC_TMP_TABLE_<uuid>`). +- SQL references datasets by index — `<SRC_0>`, `<SRC_1>`, … — with `<SRC>` kept as an alias for the + first dataset (backward compatible). +- Supports multi-statement SQL files and `CREATE [OR REPLACE] TEMPORARY VIEW` (e.g. CTE-style setup); + user-created views are tracked and dropped alongside the source temp views in a `finally` block. +- Validates inputs: rejects null/empty dataset lists and any dangling `<SRC_N>` placeholder that has + no corresponding dataset (typo guard), with clear error messages. + +Example SQL (`<SRC_0>` = orders, `<SRC_1>` = users): + +```sql +SELECT o.order_id, o.amount, o.user_id, u.user_name, u.email +FROM <SRC_0> o +FULL OUTER JOIN <SRC_1> u ON o.user_id = u.user_id; +``` + +### 4. `MultiTableCheckpointManager` + +Parses, formats, and manipulates per-table checkpoints stored in `deltastreamer.checkpoint.key`: + +``` +Multi-table : "table1=20250606182826197,table2=20250606182830145,table3=20250606182828456" +Single/legacy: "20250606182826197" +``` + +- Any value containing `=` is parsed as multi-table via a lenient parser that skips malformed entries + while preserving valid ones (so a single corrupt entry never causes the whole string to be + misclassified as single-table and collapsed under a default key). +- A bare value with no `=` is treated as a legacy single checkpoint (mapped to a default table), Review Comment: Addressed in new **Section 7: Operational checkpoint override / reset in multi-source mode** (L394–407). `--checkpoint` and `hoodie.deltastreamer.checkpoint.reset_key` accept a fully-formed composite value (`0=abc123:20250606182826197,1=def456:20250606182830145`); bare-timestamp overrides are **rejected at startup** with a clear error rather than silently mapped to a single default index and dropping other sources' cursors. Path-fingerprints are validated on the override too. -- 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]
