nsivabalan commented on PR #19299:
URL: https://github.com/apache/hudi/pull/19299#issuecomment-5027930035
First off — this is a genuinely useful addition for Hudi. The multi-source
drift problem is one of the most common footguns I see in production Streamer
deployments, and consolidating it into a single-commit, single-checkpoint
source is the right architectural direction. Thanks for taking this on and for
the level of detail in the write-up — the framing of the drift/atomicity
problem is accurate, the scoping is disciplined, and the Test Plan is thorough
enough that we can actually judge the design against how it'll be verified.
A few asks below before this can move to implementation. `InputBatch` is
left for a separate follow-up comment.
### 1. Tighten the "atomic across sources" wording
The Abstract and the Motivating cases imply a globally consistent
point-in-time snapshot across the source tables. The design does not deliver
that — each per-source read snapshots its own timeline as the driver iterates
sequentially, and a source can advance between the moment one source's read
starts and the next source's read starts.
The guarantee is: **the batch commits every source's checkpoint together, or
none of them**. That's real and worth having. But it's a weaker guarantee than
"distributed snapshot," and users doing tight cross-source foreign-key joins
need the distinction to be explicit. Please spell it out, roughly:
> Atomicity here means "the batch commits all source checkpoints together,
or none." It does **not** guarantee a globally consistent snapshot across
sources: each source is read against its own timeline as the driver iterates,
and a source can advance between the moment the previous source's read starts
and this source's read starts. Users who need cross-source point-in-time
consistency at the row level must layer their own snapshot mechanism (e.g.
as-of-time reads at a coordinated timestamp) on top.
### 2. `InputBatch#getBatch()` fail-loud is a behavior change on a public
interface — deferring detailed discussion
Marking this as a known concern; will follow up in a separate comment with a
design proposal. Short version: `InputBatch` is documented public API used by
every source, transformer, and third-party integration. Making `getBatch()`
throw when it would previously have returned a value is a behavior change worth
thinking about carefully, and there may be alternatives (separate type,
generic-parameter reuse) that avoid the fail-loud contract entirely by making
misuse a compile-time error rather than a runtime one. Detailed comment to
follow.
### 3. Source identifier design — switch to indices, add optional label,
fingerprint the checkpoint
The current scheme has three parallel identifiers in flight for the same
source: the dotted-name-in-`tables`-list, the
underscore-mangled-in-property-key form, and the positional-index-in-SQL form.
This is a maintenance and operator-confusion tax, and the mangling has a real
silent-collision failure mode: a user declaring `tables=db.orders,db_orders`
gets two sources whose property keys both resolve to
`...source.hudi.db_orders.*`, one silently overwrites the other.
Proposal: replace the alias-with-mangling scheme with **zero-based indices**
as the durable identifier everywhere.
```
hoodie.streamer.source.multi.hudi.count=2
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.1.path=/hudi/users
hoodie.streamer.source.multi.hudi.1.num_instants=3
# Optional non-durable human label — used only in logs / metrics / error
messages
hoodie.streamer.source.multi.hudi.0.label=orders
hoodie.streamer.source.multi.hudi.1.label=users
```
Checkpoint format: `<index>=<pathHash>:<instantTime>,…` — e.g.
`0=abc123:20250606182826197,1=def456:20250606182830145`. The path hash guards
against silent misconfiguration when a source's path is reordered or replaced
at an existing index — a fingerprint mismatch fails the job at startup with a
clear diagnostic pointing at the affected index, rather than silently resuming
from an unrelated table's checkpoint.
SQL bindings stay as `<SRC_0>`, `<SRC_1>`, … (with `<SRC>` still aliasing
`<SRC_0>` for single-source SQL file compatibility).
Benefits:
- No property-key collision possible by construction.
- Legacy single-timestamp checkpoints migrate cleanly: `<timestamp>` →
`0=<computed-hash>:<timestamp>` on next commit.
- Reorder/replace is caught at startup rather than silently corrupting the
resume point.
- Labels are decoupled from durability — operators can rename freely for log
clarity without touching the checkpoint.
### 4. Source-list evolution semantics need to be specified
The RFC covers migration from a legacy single-source checkpoint but does not
address the ongoing case where a running pipeline's source list changes. Please
specify the intended semantics for each of the following operational events, or
explicitly declare which are supported vs. out-of-scope for v1:
- **Adding a new source** to an existing multi-source pipeline. Does the new
source apply `missing_checkpoint_strategy`? Is that the right default, given
`READ_LATEST` implies a silent history skip?
- **Removing a source** from an existing multi-source pipeline. Is the
persisted checkpoint entry for the removed source ignored, warned about, or
cleaned up?
- **Reordering the source list**, given the `<SRC_0>`/`<SRC_1>` SQL bindings
are positional — reordering silently rebinds the SQL against different sources.
How is this prevented, or is it accepted as operator responsibility?
- **Replacing an existing source's path** with a different Hudi table at the
same position — would resume from the previous source's checkpoint, which is
meaningless for the new source (silent data loss or silent misuse).
These are distinct operational events, each with a different failure mode;
the RFC should own the answer to each.
### 5. All-sources-failed behavior under `skip_on_schema_failure=true`
`skip_on_schema_failure=true` is a well-scoped opt-in for tolerating a
single source's schema-resolution failure without stalling the pipeline. Please
clarify the behavior when *every* configured source has
`skip_on_schema_failure=true` and every source fails schema resolution in the
same batch.
Suggested semantics: 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 advance, no timeline noise, the
`<table>.schema_resolution_failure` metric fires N times, and the operator's
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.
Also worth documenting the `<table>.schema_resolution_failure` metric as a
**hard-alert** signal, not a soft one — a persistent non-zero value indicates a
source is silently no-op'ing.
### 6. Multi-statement SQL + temp-view namespace ownership
Two clarifications on the SQL transformer:
- **Cleanup lifetime.** User temp views tracked and cleaned up "in a
`finally` block" — please specify whether this is per-batch or
per-Streamer-process. A driver-restart mid-batch will leak them either way;
that's OK if it's stated.
- **Namespace collision.** `HOODIE_SRC_TMP_TABLE_<uuid>` is unique per
invocation. But user-created views (`CREATE OR REPLACE TEMPORARY VIEW my_dim`)
are not UUID-mangled and can collide with views created by an operator's own
SQL outside the transformer. Recommend documenting that the transformer takes
exclusive ownership of the Spark session's temp-view namespace during
`apply()`, or that view names inside the user's SQL should follow a documented
naming convention to avoid collision with anything the operator manages outside
Streamer.
### 7. Confirm the schema-provider story for the merged output
Reviewing the current single-source `HoodieIncrSource` flow,
`SchemaProvider` is used for two things: (a) as a fallback source of the
target-write schema when the transformed dataset is empty, and (b) as a way for
users to pin the target-write schema of the downstream table via
`hoodie.streamer.schemaprovider.class`. In practice, most `HoodieIncrSource`
deployments do not configure a `SchemaProvider` at all — Hudi tables
self-describe, so the schema flows through implicitly from the `Dataset<Row>`
returned by the source.
For `HoodieIncrMultiSource`, please confirm in the RFC that:
- The top-level `hoodie.streamer.schemaprovider.class` continues to work
unchanged for the merged output. When configured with a non-null target schema,
it wins for the write (matching Sub-branch A of `getDeducedSchemaProvider`).
- Registry-backed providers (`SchemaRegistryProvider` and variants) are not
applicable to the multi-source read path, since Hudi tables self-describe. This
should not surprise users given they aren't applicable to single-source
`HoodieIncrSource` either, but a one-line note prevents someone from wiring one
up expecting per-source behavior.
### 8. Scope statement: v1 is Hudi-to-Hudi only
The RFC introduces general-looking API surface (`InputBatch#getBatches()`,
`MultiDatasetTransformer`) but the only source implementation is Hudi-specific
— the checkpoint encoding (`table=instantTime` strings), the per-source config
keys (`hoodie.streamer.source.multi.hudi.*`), and the sequential-read-on-driver
design all assume Hudi tables. Please add an explicit scope statement:
> **v1 scope: Hudi-to-Hudi only.** `HoodieIncrMultiSource` reads exclusively
from Hudi tables. Extending multi-source semantics to Kafka, S3 events, GCS
events, JSON/DFS, JDBC, or other non-Hudi sources is out of scope for this RFC
and requires its own design work — specifically, a checkpoint encoding suitable
for each source type (Kafka per-partition offsets, S3 event-queue positions,
JDBC watermark columns) that does not fit the `table=instantTime` string format
used by `MultiTableCheckpointManager`. The `MultiDatasetTransformer` and
`InputBatch#getBatches()` API changes are structured so that a future
multi-source implementation for another source type could reuse them, but the
implementation and validation for those is future work.
### 9. Schema-provider scope for v1
Please add an explicit paragraph on per-source schema providers, roughly:
> **Schema providers — v1 scope.** `HoodieIncrMultiSource` 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; when
absent, the writer schema is deduced from the merged transformer output
reconciled against the latest target-table schema.
>
> **Startup behavior:** per-source schema-provider keys (e.g.
`hoodie.streamer.source.hudi.{table}.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 v1 API changes do not preclude
it.
### 10. Empty-batch guard rewording
The RFC's current phrasing "empty-batch guard skips the transformer when the
source produced no data" doesn't quite match how `HoodieIncrMultiSource`
behaves. The source always emits datasets (empty-but-schema-shaped for tables
with no new data), so `getBatches()` is never actually empty — the list
contains N entries, some of which may individually be empty. Please reword:
> **Empty-batch guard.** The transformer is invoked whenever
`HoodieIncrMultiSource` returns any non-empty dataset across its N configured
sources. When *every* dataset in `getBatches()` is empty (all sources caught up
in this batch), the transformer is skipped and the batch either commits with no
data or no-ops entirely, depending on Streamer's "commit on no-op batch"
configuration.
This also disambiguates from the "all sources failed schema resolution" case
in #5 — which routes through the schema-failure metric path, not the
empty-batch path.
### 11. Confirm sanitization at write time
The RFC states that `SanitizationUtils` is bypassed at the source read
(since Hudi table rows are already schema-conformant), but "sanitization still
applies at write time to the merged transformer output when the user has
enabled it." Please confirm this is actually the case in the implementation —
specifically, that field-name sanitization on the merged output flows through
whatever the top-level Streamer write path applies. If sanitization at write
time was only ever driven by the `RowSource` codepath being taken at read time,
this bypass silently disables it. A one-line integration test that runs the
merged output through a write with sanitization enabled would close this out.
---
## Overall
Solid, well-scoped design for a real pain point in multi-source ingestion.
Once the identifier design tightens (#3), the source-list evolution semantics
get spelled out (#4), and the scope statements land (#8, #9), the RFC is close
to `+1`. The `InputBatch` conversation is separate and I'll follow up on that.
--
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]
