ashokkumar-allu commented on PR #19299:
URL: https://github.com/apache/hudi/pull/19299#issuecomment-5080462060
@nsivabalan — thank you for the thorough review. All 11 asks are addressed
in the update just pushed (`82118d0d`). Structuring the response to match your
section headings; section/line references are into the updated RFC.
### 1. Tighten the "atomic across sources" wording
Adopted. The Abstract (L42–52) no longer implies a distributed snapshot; the
Motivating use cases now caveat with "subject to the atomicity contract in
*Failure and atomicity* — this is transactional atomicity, not a distributed
snapshot" (L92–93); and the new **Failure and atomicity contract** section
leads with "Atomicity is transactional, not distributed-snapshot" and spells
out that a source can advance between the moment the previous source's read
starts and this source's read starts (L537–543). Users needing row-level
cross-source point-in-time consistency are directed to layer their own
as-of-time reads on top.
### 2. `InputBatch#getBatch()` fail-loud — deferring
Went further than defer: the entire `InputBatch` change and the
`HoodieIncrMultiSource` `Source` subclass are removed. See the reply on your
InputBatch composition follow-up comment and item 3 of **Alternatives
Considered** (L595–610). Zero public-API changes to `InputBatch`, `Source`, or
`SourceFormatAdapter`.
### 3. Source identifier design — indices + label + fingerprint
Adopted verbatim. Section 2 (L156–201) introduces zero-based indexed
identifiers with an optional non-durable `label`, and calls out the
`tables=db.orders,db_orders` property-key collision failure mode by name
(L181–191). Section 7 (L356–392) defines the `<index>=<pathHash>:<instantTime>`
composite checkpoint. On resume, the manager computes the fingerprint of each
currently-configured index's path and fails the job at startup with a clear
diagnostic naming the affected index and both paths on mismatch.
### 4. Source-list evolution semantics
New **Section 8: Source-list evolution semantics** (L409–433) covers all
four events:
- **Add source at a new index:** no prior checkpoint →
`missing_checkpoint_strategy` applies, with operator guidance about
`READ_LATEST` silent skip.
- **Remove source at an index:** stale entry pruned on next commit with a
`WARN` for auditability.
- **Reorder source list:** fingerprint mismatch fails at startup with clear
diagnostic; requires an explicit `--checkpoint` override presenting the correct
new mapping.
- **Replace source's path at existing index:** same fingerprint-mismatch
failure mode; operators consciously proceed with `--checkpoint` override.
### 5. All-sources-failed under `skip_on_schema_failure=true`
Section 4 (L238–247) adds the semantics you suggested: merged transformer
output is empty, batch does not commit by default (respecting Streamer's
"commit on no-op batch" config), N `<label-or-index>.schema_resolution_failure`
counter metrics fire, operator alerting is the surface. The metric is
documented as a hard-alert signal.
### 6. Multi-statement SQL + temp-view namespace ownership
Section 6 (L331–346) specifies:
- **Cleanup lifetime:** per-batch `finally` — tracked view names accumulated
during a single `apply()` are dropped whether it succeeds or throws. Driver
crash mid-batch leaks views; safe because every leaked view has a fresh UUID
and can't collide with a subsequent run.
- **Source-view names:** UUID-mangled (`HOODIE_SRC_TMP_TABLE_<uuid>`), no
collision by construction.
- **User-created view names:** transformer takes **exclusive ownership of
the Spark session's temp-view namespace during `apply()`** — pre-existing
same-name views get replaced and cleaned up. Documented as a hard rule in the
transformer's javadoc, not a soft convention.
### 7. Confirm the schema-provider story for the merged output
Confirmed in Section 4 (L275–280):
- Top-level `hoodie.streamer.schemaprovider.class` continues to work
unchanged for the merged output (matches Sub-branch A of
`getDeducedSchemaProvider`).
- Registry-backed providers (`SchemaRegistryProvider` etc.) explicitly noted
as not applicable to the multi-source read path — Hudi tables self-describe,
same as with `HoodieIncrSource` today.
### 8. Scope statement: v1 is Hudi-to-Hudi only
New **Scope** section (L111–125) with the explicit statement you drafted,
plus a composition-extensibility note explaining that the outstanding work per
new source type is checkpoint encoding (Kafka per-partition offsets, S3
event-queue positions, JDBC watermark columns don't fit
`index=pathHash:instantTime`), not adapter plumbing.
### 9. Schema-provider scope for v1
Section 4 (L266–289) adds the paragraph you drafted:
- No per-source `SchemaProvider` in v1 — SQL is the right layer for shape
normalization.
- Merged output honors the existing top-level provider (as in #7).
- Per-source schema-provider keys
(`hoodie.streamer.source.multi.hudi.{index}.schemaprovider.class`) rejected at
startup with a clear error.
- Follow-up RFC option preserved for registry-backed evolution across source
*and* target under one Streamer job.
### 10. Empty-batch guard rewording
Section 9 (L473–480) reworded exactly as suggested — the source always emits
N datasets (some may individually be empty), so `getBatches()` is never empty.
The transformer is invoked whenever any non-empty dataset is returned; when
*every* dataset is empty (all sources caught up), the transformer is skipped
and the batch either commits with the unchanged composite checkpoint or no-ops
entirely, depending on Streamer config. Explicitly disambiguated from the
all-sources-failed-schema case (which routes through the schema-failure metric
path).
### 11. Confirm sanitization at write time
Confirmed in Section 4 (L258–264) — per-row sanitization is bypassed at the
per-source read (rows are already Avro-conformant and it would otherwise run
twice), but write-time sanitization on the merged output continues to flow
through Streamer's normal write path when the operator has enabled it. Added a
dedicated **Sanitization-at-write-time integration test** to the Test Plan:
writes the merged output with
`hoodie.datasource.write.sanitize.avro.field.names=true` and asserts the
writer's sanitization codepath was taken on the merged rows even though
per-source reads bypassed it.
---
Happy to iterate further on any of these. Ready for another round when you
are.
--
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]
