nsivabalan commented on PR #19299:
URL: https://github.com/apache/hudi/pull/19299#issuecomment-5028453422
Follow-up to my previous review, specifically on the `InputBatch` / `Source`
API changes proposed in the RFC (Section 1 "Multi-dataset `InputBatch`" and
Section 2 "`HoodieIncrMultiSource`"). Doing the walk-through end-to-end
surfaced a design shape that I think is cleaner than what the RFC currently
proposes. Sharing it here as a proposal, not a demand — you should weigh it on
its own merits.
## TL;DR
- The RFC changes `InputBatch` (adds `List<T> batches`, adds `getBatches()`,
makes `getBatch()` fail loud) and adds a new `Source` subclass
`HoodieIncrMultiSource`.
- **Alternative:** leave `InputBatch` and the `Source` hierarchy entirely
unchanged. Introduce a `MultiSourceFormatAdapter` that composes N per-source
`SourceFormatAdapter`s. `StreamSync` branches once on adapter type at the top
of `fetchNextBatchFromSource`; the rest of the pipeline is identical for
single-source and multi-source.
- This avoids three runtime breakage points, keeps the public `InputBatch`
API contract identical, and — as a side benefit — generalizes cleanly to
heterogeneous multi-source (Hudi + Kafka + JSON in one pipeline) if that ever
becomes a real requirement.
Below is the reasoning in full: three call-stack walk-throughs (baseline,
RFC, alternative), then a pros/cons comparison.
---
## Walk-through 1: current single-source flow (baseline, `HoodieIncrSource`)
```
StreamSync.sync()
└─ readFromSource(metaClient)
└─ fetchFromSourceAndPrepareRecords(resumeCheckpoint, metaClient)
└─ fetchNextBatchFromSource(resumeCheckpoint, metaClient)
[StreamSync.java:690]
│
└─ if (transformer.isPresent()) {
│
├─ SourceFormatAdapter.fetchNewDataInRowFormat(...)
[SourceFormatAdapter.java:242]
│ switch (source.getSourceType()) {
│ case ROW → ((Source<Dataset<Row>>)
source).fetchNext(...)
│ case AVRO / JSON / PROTO → cast + convert to
Dataset<Row>
│ }
│ │
│ └─ Source.fetchNext(lastCheckpoint, sourceLimit)
[Source.java:149]
│ ├─ translateCheckpoint(lastCheckpoint)
│ ├─ readFromCheckpoint(...) → HoodieIncrSource
impl
│ │ returns InputBatch<Dataset<Row>>(
│ │ Option.of(dataset), checkpoint,
schemaProvider)
│ ├─ batch.getBatch().ifPresent(this::persist)
← touches getBatch()
│ └─ wrap w/ overriddenSchemaProvider if set
← touches getBatch()
│ returns: InputBatch<Dataset<Row>> dataAndCheckpoint
│
├─ Option<Dataset<Row>> transformed =
│ dataAndCheckpoint.getBatch()
← touches getBatch()
│ .map(data -> transformer.get().apply(jsc,
spark, data, props));
│
├─ processErrorEvents(transformed,
CUSTOM_TRANSFORMER_FAILURE)
├─ checkpoint =
dataAndCheckpoint.getCheckpointForNextBatch()
│
└─ deduce writer schema (Sub-branch A: user-provided SP
wins;
Sub-branch B: deduce from
transformed output)
→ row-writer or Avro branch → build final InputBatch
for writer
```
`InputBatch` is used as: single `Option<T>` payload + `Checkpoint` +
`SchemaProvider`. Three `getBatch()` call sites in the critical path (persist,
wrap, transform).
---
## Walk-through 2: multi-source per the RFC (Option A)
Same stack, but `HoodieIncrMultiSource` is a new `Source<Dataset<Row>>`
returning an `InputBatch<Dataset<Row>>` where `batches.size() > 1` and
`getBatch()` throws.
```
StreamSync.fetchNextBatchFromSource(...)
│
├─ (startup guard elsewhere: HoodieIncrMultiSource requires
MultiDatasetTransformer,
│ otherwise HoodieStreamerException at init)
│
└─ if (transformer.isPresent()) {
│
├─ SourceFormatAdapter.fetchNewDataInRowFormat(...)
│ switch case ROW → ((Source<Dataset<Row>>) source).fetchNext(...)
│ │
│ └─ Source.fetchNext(...)
[Source.java:149]
│ ├─ readFromCheckpoint → HoodieIncrMultiSource impl
│ │ ├─ MultiTableCheckpointManager.parse(...)
│ │ ├─ FOR EACH configured source (sequential):
│ │ │ HoodieIncrSource-equivalent read → Dataset<Row>
│ │ └─ returns InputBatch<Dataset<Row>>(
│ │ batches = [src_0, src_1, ..., src_N-1], ← new
field
│ │ checkpoint = "0=t0,1=t1,...",
│ │ schemaProvider = NullSchemaProvider)
│ │
│ ├─ batch.getBatch().ifPresent(this::persist)
⚠ THROWS
│ │ Fail-loud getBatch() fires here — inside
Source.fetchNext(),
│ │ in the base class, before StreamSync ever sees the batch.
│ │
│ │ ▶ BREAKAGE POINT #1
│ │
│ │ Fix required: either Source.fetchNext() has to iterate
│ │ getBatches() and persist each, or HoodieIncrMultiSource
│ │ has to override fetchNext() to skip persist. RFC does
not say.
│ │
│ └─ (assuming patched) returns to SourceFormatAdapter
│
│ Back in SourceFormatAdapter.fetchNewDataInRowFormat, line 247:
│ return new InputBatch<>(
│ processErrorEvents(datasetInputBatch.getBatch(), ...),
⚠ THROWS
│ datasetInputBatch.getCheckpointForNextBatch(),
│ datasetInputBatch.getSchemaProvider());
│
│ ▶ BREAKAGE POINT #2
│
│ Fix required: SourceFormatAdapter has to branch on
│ "batches.size() > 1" and processErrorEvents per-dataset. RFC says
│ "SourceFormatAdapter processes error events per-dataset on the
│ multi-dataset row path" — but the exact patch is not shown.
│
├─ Option<Dataset<Row>> transformed =
│ dataAndCheckpoint.getBatch()
⚠ THROWS
│ .map(data -> transformer.get().apply(...));
│
│ ▶ BREAKAGE POINT #3
│
│ Fix required: branch on transformer being a MultiDatasetTransformer:
│ if (transformer.get() instanceof MultiDatasetTransformer
│ && dataAndCheckpoint.getBatches().size() > 1) {
│ Dataset<Row> merged = ((MultiDatasetTransformer)
transformer.get())
│ .apply(jsc, spark, dataAndCheckpoint.getBatches(),
props);
│ transformed = Option.of(merged);
│ } else {
│ transformed = dataAndCheckpoint.getBatch().map(...); //
single-source
│ }
│
└─ (rest of the pipeline unchanged)
}
```
Three sites where the existing base-class code calls `getBatch()`
unconditionally, three sites where the RFC's fail-loud contract will throw at
runtime unless each site is patched to branch on `getBatches().size() > 1`. The
compiler does not enforce this — miss one, and multi-source pipelines throw
`HoodieException` at runtime.
Every third-party integration that constructs or consumes an `InputBatch`
also inherits this hazard. Fine for callers that only ever emit / consume
single-dataset batches (99% of the ecosystem) — but any caller that touches
`getBatch()` on an `InputBatch` obtained from Streamer core is now a potential
regression site if multi-source is enabled downstream of it.
---
## Walk-through 3: multi-source via `MultiSourceFormatAdapter` (proposal)
`InputBatch` unchanged. `Source` hierarchy unchanged.
`HoodieIncrMultiSource` does not exist as a new `Source` subclass. Instead:
- **`MultiSourceFormatAdapter`** — new class. Composition, not inheritance.
Constructor takes `List<SourceFormatAdapter>` (one per underlying source, one
per configured index) plus whatever coordination state it needs. It is *not* a
subclass of `SourceFormatAdapter`.
- Public methods on `MultiSourceFormatAdapter`:
- `fetchNewDataInRowFormat(Option<Checkpoint>, long)` →
`List<InputBatch<Dataset<Row>>>`. Parses the composite checkpoint, dispatches
per-source `Checkpoint` to each underlying
`SourceFormatAdapter.fetchNewDataInRowFormat(...)`, collects the results.
- `fetchNewDataInAvroFormat(...)` → throws
`UnsupportedOperationException`. v1 is row-only.
- Any other format-specific method → throws.
```
StreamSync.fetchNextBatchFromSource(...)
│
└─ if (formatAdapter instanceof MultiSourceFormatAdapter) {
│
│ ── MULTI-SOURCE BRANCH ──
│
├─ List<InputBatch<Dataset<Row>>> perSourceBatches =
│ ((MultiSourceFormatAdapter)
formatAdapter).fetchNewDataInRowFormat(...);
│
│ Inside MultiSourceFormatAdapter.fetchNewDataInRowFormat:
│ ├─ MultiTableCheckpointManager.parse(compositeCheckpoint)
│ │ → Map<Integer, Checkpoint>
│ ├─ FOR EACH index i:
│ │ Checkpoint perSourceCkpt = parsed.get(i) // or
Option.empty() if new
│ │ InputBatch<Dataset<Row>> ib =
│ │ adapters.get(i).fetchNewDataInRowFormat(perSourceCkpt,
...)
│ │ // This is unchanged SourceFormatAdapter code — including
its own
│ │ // internal processErrorEvents call for that source. No
branching
│ │ // needed here; the existing single-source path just
works, N times.
│ │ perSourceBatches.add(ib)
│ └─ return perSourceBatches
│
├─ List<Dataset<Row>> datasets = perSourceBatches.stream()
│ .map(ib -> ib.getBatch().orElseGet(() ->
emptySchemaShapedDf(...)))
│ .collect(toList());
│
├─ Dataset<Row> merged = ((MultiDatasetTransformer) transformer.get())
│ .apply(jsc, spark, datasets, props);
│
├─ Option<Dataset<Row>> transformed = Option.of(merged);
├─ Checkpoint checkpoint = MultiTableCheckpointManager.format(
│ perSourceBatches.stream()
│ .map(InputBatch::getCheckpointForNextBatch)
│ .collect(toList()));
│
} else {
│
│ ── SINGLE-SOURCE BRANCH — completely unchanged ──
│
├─ InputBatch<Dataset<Row>> dataAndCheckpoint =
│ formatAdapter.fetchNewDataInRowFormat(resumeCheckpoint,
sourceLimit);
├─ Option<Dataset<Row>> transformed = dataAndCheckpoint.getBatch()
│ .map(data -> transformer.get().apply(jsc, spark, data, props));
├─ Checkpoint checkpoint =
dataAndCheckpoint.getCheckpointForNextBatch();
}
│
│ ── FROM HERE ON, IDENTICAL CODE FOR BOTH BRANCHES ──
│
├─ processErrorEvents(transformed, CUSTOM_TRANSFORMER_FAILURE)
├─ deduce writer schema (Sub-branch A or B)
└─ row-writer or Avro branch → build final InputBatch for writer
```
Zero changes to `InputBatch`. Zero changes to `Source`. Zero changes to
`SourceFormatAdapter`. One new class (`MultiSourceFormatAdapter`) that composes
existing ones. One `instanceof` branch in `StreamSync` that vanishes into the
rest of the pipeline after producing `transformed` and `checkpoint`.
---
## The three breakage points from Walk-through 2 — where they go under the
proposal
| Breakage point | RFC (Option A) | Proposal |
|---|---|---|
| `Source.fetchNext()` persist path | Requires patch to `Source.fetchNext()`
or an override in `HoodieIncrMultiSource` | Does not arise.
`Source.fetchNext()` is never called with a multi-batch — each per-source
adapter calls it with singular payload. |
| `SourceFormatAdapter.fetchNewDataInRowFormat()` error-event path |
Requires branch on `getBatches().size() > 1` | Does not arise. Error events are
handled per-source *inside* each underlying `SourceFormatAdapter` — the same
code that has always handled them. |
| `StreamSync.fetchNextBatchFromSource()` transformer path | Requires
`instanceof MultiDatasetTransformer` branch on the `InputBatch` | Requires
`instanceof MultiSourceFormatAdapter` branch on the adapter. Same idea, but at
a natural type boundary rather than mid-flow. |
All three of the RFC's implicit patches are either eliminated or moved to a
single, obvious branch point.
---
## Pros / cons
### RFC's approach (`InputBatch#getBatches()` + `HoodieIncrMultiSource` as a
Source)
**Pros:**
- One place where multi-source-ness lives — inside `HoodieIncrMultiSource`.
Small class inventory.
- The `Source` abstraction is where "read incrementally from something"
lives, so multi-source-as-a-source has intuitive appeal.
- `InputBatch#getBatches()` gives a fixed shape that any future multi-source
can use.
**Cons:**
- `InputBatch#getBatch()` is a documented public API. Making it throw when a
multi-batch is present is a behavior change on that API, and the fail-loud
contract is a runtime signal, not a compile-time one.
- Three call sites in the critical path (Source.fetchNext,
SourceFormatAdapter.fetchNewDataInRowFormat,
StreamSync.fetchNextBatchFromSource) all touch `getBatch()` unconditionally
today. Each has to be patched to branch. Miss one → runtime throw for
multi-source pipelines only.
- Third-party sources / transformers that touch `getBatch()` on an
`InputBatch` obtained from Streamer core inherit the same runtime hazard.
- Multi-source-ness lives in a `Source` subclass, but the `Source` contract
is naturally singular ("give me the next batch given a checkpoint"). Making a
`Source` implementation coordinate across N logical sources fights the
abstraction — sequential-on-driver, composite checkpoint parsing,
multi-transformer contract dispatch all end up inside one class.
### Proposal (`MultiSourceFormatAdapter` composition)
**Pros:**
- **`InputBatch` public API is byte-for-byte unchanged.** No new methods, no
behavior changes on existing methods. Every third-party source and transformer
continues to work exactly as before.
- **`Source` hierarchy is unchanged.** No new subclass needed. Each
`HoodieIncrSource` instance is a normal single-source `Source` with a normal
single-source `SourceFormatAdapter` around it.
- **Compile-time separation.** `SourceFormatAdapter` and
`MultiSourceFormatAdapter` are distinct types. The `instanceof` check in
`StreamSync` is the only place in the code that has to know which world we're
in. Anywhere else, holding one type or the other is unambiguous.
- **Composition over inheritance.** The multi adapter *has* N single-source
adapters; it does not *extend* one. This is a good invariant because per-source
config, error handling, and checkpoint parsing all stay local to their own
adapter.
- **The "rest of the pipeline is unchanged" property** — schema deduction,
row-writer vs Avro, commit — all shared code between single and multi. The
branching lives at exactly one point.
- **Naturally extensible to heterogeneous multi-source.** See next section.
**Cons:**
- The extension point for future multi-source (Kafka multi, S3 multi, etc.)
is at the *adapter* layer rather than the *source* layer. Someone adding a
`MultiKafkaSource` in the future extends `MultiSourceFormatAdapter` (or writes
a new adapter with similar shape), not `Source`. That's a legitimate
architectural inversion vs the RFC — worth naming.
- Introduces a new class that has to be documented and understood alongside
`SourceFormatAdapter`. `SourceFormatAdapter` was already a subtle part of
Streamer; another one adds surface area.
- The composite-checkpoint parsing logic has to live *somewhere*. Under the
RFC it lives in `HoodieIncrMultiSource`; under the proposal it lives in
`MultiSourceFormatAdapter`. Roughly equivalent complexity — just a different
home.
### Extensibility argument for the proposal
Under the composition model, the multi adapter's public method is
`fetchNewDataInRowFormat(...)` returning `List<InputBatch<Dataset<Row>>>` — the
uniform output shape is `Dataset<Row>`, regardless of what the underlying
sources were. Since every `SourceFormatAdapter` already normalizes ROW / AVRO /
JSON / PROTO to `Dataset<Row>` in `fetchNewDataInRowFormat`, this means:
- **Homogeneous multi-source of any type** — N Kafka sources, N JSON
sources, N Hudi sources — works because each per-source adapter handles its own
type independently.
- **Heterogeneous multi-source** — mixed Hudi + Kafka + JSON in one pipeline
— works because each per-source adapter normalizes to `Dataset<Row>` on its
own; the multi adapter is source-type-agnostic once each adapter has returned
its `InputBatch<Dataset<Row>>`.
- **Future new source types** — a new `MyCustomSource` gets multi-source
support automatically as long as its `SourceFormatAdapter` produces a
`Dataset<Row>` in `fetchNewDataInRowFormat`. No new plumbing needed in the
multi adapter.
The one thing that stays per-source-type-specific is the **checkpoint
encoding**. `HoodieIncrSource` uses instant times; Kafka uses per-partition
offset maps; DFS sources use max modification time. The multi-checkpoint format
generalizes to `<index>=<opaque-serialized-checkpoint>,<index>=...`: the multi
adapter splits by comma and by equals, but treats the inner value as opaque and
hands it to the corresponding per-source adapter, which knows how to interpret
it (via `Source#translateCheckpoint` or its own logic).
This is a stronger property than the RFC's design gives us. Under the RFC,
`HoodieIncrMultiSource` bakes in "N incremental reads of Hudi tables" — it is
not a general shape. A future `KafkaIncrMultiSource` would be a new `Source`
subclass with its own composite-checkpoint parser, its own multi-dataset
transformer contract, and its own per-source-config surface. Under the
proposal, all of that plumbing is already in `MultiSourceFormatAdapter` and
only needs per-source-type checkpoint serde.
**Reasonable v1 scoping:** the initial implementation can still ship "N Hudi
sources only" — matching the RFC's stated Hudi-to-Hudi scope. But the *design*
permits heterogeneous multi-source without further architectural work, so the
extension is a matter of testing and documentation rather than a re-design.
---
## Recommendation
Consider replacing the RFC's `InputBatch` changes and
`HoodieIncrMultiSource` `Source` subclass with a `MultiSourceFormatAdapter`
composition model. Trade-offs are as above, and I think the design is
meaningfully cleaner for this feature. If the concerns about the RFC's approach
(the three breakage points, the fail-loud `getBatch()` contract, the
third-party surface) don't resonate — happy to talk through any of them.
--
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]
