voonhous commented on code in PR #19144:
URL: https://github.com/apache/hudi/pull/19144#discussion_r3519904666
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java:
##########
@@ -223,19 +244,22 @@ private SchemaCompatibilityResult getCompatibility(final
HoodieSchema reader,
final HoodieSchema
writer,
final
Deque<LocationInfo> locations) {
log.debug("Checking compatibility of reader {} with writer {}", reader,
writer);
- final ReaderWriter pair = new ReaderWriter(reader, writer);
- SchemaCompatibilityResult result = mMemoizeMap.get(pair);
- if (result != null) {
- if (result.getCompatibilityType() ==
SchemaCompatibilityType.RECURSION_IN_PROGRESS) {
- // Break the recursion here.
- // schemas are compatible unless proven incompatible:
- result = SchemaCompatibilityResult.compatible();
- }
- } else {
- // Mark this reader/writer pair as "in progress":
- mMemoizeMap.put(pair, SchemaCompatibilityResult.recursionInProgress());
- result = calculateCompatibility(reader, writer, locations);
- mMemoizeMap.put(pair, result);
+ final ReaderWriter pair = new ReaderWriter(reader, writer,
namesValidatedAt(locations));
+ final SchemaCompatibilityResult memoized =
memoizedCompatibleResults.get(pair);
+ if (memoized != null) {
+ return memoized;
+ }
+ if (!inProgressPairs.add(pair)) {
+ // Break the recursion here.
+ // schemas are compatible unless proven incompatible:
+ return SchemaCompatibilityResult.compatible();
+ }
+ SchemaCompatibilityResult result = calculateCompatibility(reader,
writer, locations);
+ inProgressPairs.remove(pair);
+ if (result.getCompatibilityType() == SchemaCompatibilityType.COMPATIBLE)
{
+ // Incompatible results embed the field path they were found at, so
they are
+ // recomputed per occurrence; memoizing them would report only the
first path.
+ memoizedCompatibleResults.put(pair, result);
Review Comment:
Verified issue: a COMPATIBLE result computed while an ancestor pair is still
in `inProgressPairs` rests on the optimistic recursion-break `compatible()`,
but is memoized here unconditionally; a later occurrence of the pair outside
the cycle reuses the tainted verdict. Repro (executed): reader `Top{g:
union[G{p:X}, G2{}], s: Z}` with `X{bad:int, z:Z}`, `Z{x: array<X>}` vs writer
with `X.bad=string` -> full check returns COMPATIBLE while the `s` subtree
alone is INCOMPATIBLE at `/s/x/bad` (the in-progress `(X,X')` INCOMPATIBLE is
discarded by the reader-union branch probe, the one place failures are
dropped). Without the union it degrades to silently dropping `/s/x/bad` from
the report. Fix direction: skip memoizing results whose computation overlapped
a recursion break on a still-in-progress pair.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java:
##########
@@ -223,19 +244,22 @@ private SchemaCompatibilityResult getCompatibility(final
HoodieSchema reader,
final HoodieSchema
writer,
final
Deque<LocationInfo> locations) {
log.debug("Checking compatibility of reader {} with writer {}", reader,
writer);
- final ReaderWriter pair = new ReaderWriter(reader, writer);
- SchemaCompatibilityResult result = mMemoizeMap.get(pair);
- if (result != null) {
- if (result.getCompatibilityType() ==
SchemaCompatibilityType.RECURSION_IN_PROGRESS) {
- // Break the recursion here.
- // schemas are compatible unless proven incompatible:
- result = SchemaCompatibilityResult.compatible();
- }
- } else {
- // Mark this reader/writer pair as "in progress":
- mMemoizeMap.put(pair, SchemaCompatibilityResult.recursionInProgress());
- result = calculateCompatibility(reader, writer, locations);
- mMemoizeMap.put(pair, result);
+ final ReaderWriter pair = new ReaderWriter(reader, writer,
namesValidatedAt(locations));
+ final SchemaCompatibilityResult memoized =
memoizedCompatibleResults.get(pair);
+ if (memoized != null) {
+ return memoized;
+ }
+ if (!inProgressPairs.add(pair)) {
+ // Break the recursion here.
+ // schemas are compatible unless proven incompatible:
+ return SchemaCompatibilityResult.compatible();
+ }
+ SchemaCompatibilityResult result = calculateCompatibility(reader,
writer, locations);
+ inProgressPairs.remove(pair);
+ if (result.getCompatibilityType() == SchemaCompatibilityType.COMPATIBLE)
{
Review Comment:
Incompatible pairs recompute per occurrence, so a shared named type
referenced k times per level over d levels costs k^d subtree re-checks and
accumulates k^d `Incompatibility` objects (measured: k=2, d=22 -> 3.7s / 4.2M
objects, OOM near d=25; k=4 hits the wall around d=12). To be fair this
exposure predates the branch -- the old identity memo never hit across
occurrences either -- and the memo now makes the compatible case linear. But
Spark's `deduceWriterSchema` runs this unconditionally per write, so worth
fixing while here: memoize a location-free INCOMPATIBLE verdict for pruning and
re-derive paths on reuse, or cap accumulation.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java:
##########
@@ -182,7 +198,12 @@ public String toString() {
* </p>
*/
private static final class ReaderWriterCompatibilityChecker {
- private final Map<ReaderWriter, SchemaCompatibilityResult> mMemoizeMap =
new HashMap<>();
+ // Compatible results only: they carry no location state, so they are safe
to reuse
+ // when the same value pair recurs at a different field path in the same
naming context.
+ private final Map<ReaderWriter, SchemaCompatibilityResult>
memoizedCompatibleResults = new HashMap<>();
Review Comment:
Values in this map are always value-equal to `compatible()` (`mergedWith`
only yields COMPATIBLE from two COMPATIBLE inputs whose lists are empty), so
this can be a `Set<ReaderWriter>` with `return
SchemaCompatibilityResult.compatible()` on hit -- less garbage, and the
compatible-only invariant becomes structural instead of comment-enforced.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCache.java:
##########
@@ -28,21 +29,77 @@
* <p>This is a global cache which works for a JVM lifecycle.
* A collection of schema instances are maintained.
*
+ * <p>This value-keyed pool is the canonicalization mechanism behind
+ * {@link HoodieSchema#fromAvroSchema}, and can also be used directly to
intern schemas
+ * produced without an Avro source (builders, converters).
+ *
+ * <p>Interning is never lossy: entries are keyed on the schema's full
serialized content,
+ * NOT on {@link HoodieSchema#equals}. Avro equality (which HoodieSchema
equality delegates
+ * to) ignores doc strings and aliases, so keying on it would collapse schemas
that differ
+ * only in that metadata and silently drop it -- docs drive catalog sync
column comments and
+ * aliases drive schema-evolution field matching. Schemas that differ in docs
or aliases
+ * intern to distinct canonical instances even though they are {@code
equals()}.
+ *
* <p>NOTE: The schema which is used frequently should be cached through this
cache.
*/
public class HoodieSchemaCache {
// Ensure that there is only one variable instance of the same schema within
an entire JVM lifetime
- private static final LoadingCache<HoodieSchema, HoodieSchema> SCHEMA_CACHE =
- Caffeine.newBuilder().weakValues().maximumSize(1024).build(k -> k);
+ private static final LoadingCache<SchemaContentKey, HoodieSchema>
SCHEMA_CACHE =
+ Caffeine.newBuilder().weakValues().maximumSize(1024).build(key ->
key.schema);
/**
* Get schema variable from global cache. If not found, put it into the
cache and then return it.
*
+ * <p>Two schemas converge on one canonical instance only when their full
serialized form
+ * (including doc strings and aliases) is identical; see the class javadoc.
+ *
+ * <p>A schema that is valid in memory but cannot be serialized to JSON --
e.g. two distinct
+ * nested records that share a name, as some projection/reader paths produce
-- has no content
+ * key, so it is returned uncached instead of interned. Canonicalization is
only a
+ * de-duplication optimization, so skipping it stays correct.
+ *
* @param schema schema to get
* @return if found, return the exist schema variable, otherwise return the
param itself.
*/
public static HoodieSchema intern(HoodieSchema schema) {
- return SCHEMA_CACHE.get(schema);
+ SchemaContentKey key;
+ try {
+ key = new SchemaContentKey(schema);
+ } catch (AvroRuntimeException e) {
+ // Not serializable -> no content key derivable; skip interning rather
than fail the caller.
+ return schema;
+ }
+ return SCHEMA_CACHE.get(key);
+ }
+
+ /**
+ * Content-complete cache key: the serialized JSON form covers doc strings
and aliases that
+ * Avro equality ignores. The wrapper class is part of the key so a
logical-type subclass
+ * (e.g. {@link HoodieSchema.Decimal}) never collapses onto a plain wrapper
of equal content,
+ * which would break downcasts.
+ */
+ private static final class SchemaContentKey {
+ private final HoodieSchema schema;
+ private final String contentJson;
+
+ SchemaContentKey(HoodieSchema schema) {
+ this.schema = schema;
+ this.contentJson = schema.getAvroSchema().toString();
Review Comment:
Verified issue (Spark 4 profiles): on Avro 1.12.x `toString()` does not
throw for a second, structurally different nested record sharing a name -- it
emits a bare name ref -- so two different schemas produce byte-identical
`contentJson` and intern to one canonical with the wrong structure. Reproduced
end-to-end against this branch's classes with avro-1.12.0: two different
projections of a schema that reuses a named record (`projectSchema` builds
same-named pruned copies; they reach intern via
`HoodieSchemaUtils#projectSchema -> fromAvroSchema` and the per-record path) ->
the second query gets the first query's wrapper and reads the wrong nested
fields. The `AvroRuntimeException` catch only covers the 1.11 throwing
behavior. The key needs a structural walk that tracks named types by instance
-- or refuse to intern on redefinition -- rather than trusting `toString()`.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java:
##########
@@ -161,16 +177,16 @@ public boolean equals(Object obj) {
return false;
}
final ReaderWriter that = (ReaderWriter) obj;
- // Use pointer comparison here:
- return (this.mReader == that.mReader) && (this.mWriter == that.mWriter);
+ return this.mNamesValidated == that.mNamesValidated
Review Comment:
`HoodieSchema.equals` is Avro equality, which ignores reader field aliases
-- but `lookupWriterField` resolves writer fields through aliases, so the
verdict depends on metadata this key is blind to. Repro (executed): reader with
two equal-but-distinct `Inner` occurrences, one carrying `x`'s alias `oldX` and
one without, vs writer field `oldX` -> full check COMPATIBLE though the
alias-less occurrence alone is INCOMPATIBLE
(READER_FIELD_MISSING_DEFAULT_VALUE); swapping field order flips the verdict.
Hard to hit today (a single parse cannot produce the shape), but same fix locus
as the memo rework: fold aliases into the key or skip memoization when the
reader subtree carries aliases.
##########
hudi-common/src/main/java/org/apache/hudi/avro/AvroRecordContext.java:
##########
@@ -71,10 +70,10 @@ public AvroRecordContext() {
public static Object getFieldValueFromIndexedRecord(
IndexedRecord record,
String fieldName) {
- // Interning returns the canonical wrapper for this schema, whose lazily
built field list and
- // field map survive across calls, so the per-record cost is a cache hit
instead of an
- // O(schema width) wrapper rebuild.
- HoodieSchema currentSchema =
HoodieAvroSchemaCache.intern(record.getSchema());
+ // fromAvroSchema returns the canonical wrapper for this schema, whose
lazily built field
+ // list and field map survive across calls, so the per-record cost is a
cache hit instead
+ // of an O(schema width) wrapper rebuild.
+ HoodieSchema currentSchema =
HoodieSchema.fromAvroSchema(record.getSchema());
Review Comment:
Pre-existing (since #13646), but since this method is being touched: in the
loop below, a dotted path with a null intermediate struct NPEs --
`getValue(record, "a.b")` with `a == null` assigns `currentRecord = null` and
the next iteration derefs it. `HoodieAvroUtils#getNestedFieldVal` guards this
case and returns null. Worth adding the null check here plus a
`TestAvroRecordContext` case with `address = null` (nested ordering/precombine
fields reach this path).
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCache.java:
##########
@@ -28,21 +29,77 @@
* <p>This is a global cache which works for a JVM lifecycle.
* A collection of schema instances are maintained.
*
+ * <p>This value-keyed pool is the canonicalization mechanism behind
+ * {@link HoodieSchema#fromAvroSchema}, and can also be used directly to
intern schemas
+ * produced without an Avro source (builders, converters).
+ *
+ * <p>Interning is never lossy: entries are keyed on the schema's full
serialized content,
+ * NOT on {@link HoodieSchema#equals}. Avro equality (which HoodieSchema
equality delegates
+ * to) ignores doc strings and aliases, so keying on it would collapse schemas
that differ
+ * only in that metadata and silently drop it -- docs drive catalog sync
column comments and
+ * aliases drive schema-evolution field matching. Schemas that differ in docs
or aliases
+ * intern to distinct canonical instances even though they are {@code
equals()}.
+ *
* <p>NOTE: The schema which is used frequently should be cached through this
cache.
*/
public class HoodieSchemaCache {
// Ensure that there is only one variable instance of the same schema within
an entire JVM lifetime
- private static final LoadingCache<HoodieSchema, HoodieSchema> SCHEMA_CACHE =
- Caffeine.newBuilder().weakValues().maximumSize(1024).build(k -> k);
+ private static final LoadingCache<SchemaContentKey, HoodieSchema>
SCHEMA_CACHE =
+ Caffeine.newBuilder().weakValues().maximumSize(1024).build(key ->
key.schema);
/**
* Get schema variable from global cache. If not found, put it into the
cache and then return it.
*
+ * <p>Two schemas converge on one canonical instance only when their full
serialized form
+ * (including doc strings and aliases) is identical; see the class javadoc.
+ *
+ * <p>A schema that is valid in memory but cannot be serialized to JSON --
e.g. two distinct
+ * nested records that share a name, as some projection/reader paths produce
-- has no content
+ * key, so it is returned uncached instead of interned. Canonicalization is
only a
+ * de-duplication optimization, so skipping it stays correct.
+ *
* @param schema schema to get
* @return if found, return the exist schema variable, otherwise return the
param itself.
*/
public static HoodieSchema intern(HoodieSchema schema) {
- return SCHEMA_CACHE.get(schema);
+ SchemaContentKey key;
+ try {
+ key = new SchemaContentKey(schema);
Review Comment:
This now serializes the full schema JSON on every `intern()` call including
hits (~100us for a 500-col schema vs ~2ns for the old cached-hashCode lookup).
The direct callers run per log block (`HoodieDataBlock` read ctor,
`KeyBased`/`PositionBasedFileGroupRecordBuffer`), and `FileGroupRecordBuffer`
re-interns the schema handler's already-interned `requiredSchema` -- ~0.2-0.3ms
per block on wide tables. Cache `contentJson` lazily in a transient field on
`HoodieSchema`, or short-circuit already-canonical instances.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java:
##########
@@ -338,13 +346,42 @@ private HoodieSchema(Schema avroSchema,
List<HoodieSchemaField> fields) {
/**
* Factory method to create HoodieSchema from an Avro schema.
*
+ * <p>Returns the canonical instance for the given schema, converting and
interning it on
+ * first use: distinct Avro schema instances with identical serialized
content converge on
+ * one shared wrapper through {@link HoodieSchemaCache}, with a weak
identity fast path for
+ * the per-record hot path where all records of a file share the same live
{@link Schema}
+ * instance.
+ *
+ * <p>Interning is never lossy: the canonicalization key covers the schema's
full content,
+ * including doc strings and aliases, which Avro equality ignores. Schemas
that differ only
+ * in docs or aliases stay distinct wrappers (even though they are {@code
equals()}), so
+ * metadata consumed downstream (e.g. catalog sync column comments,
alias-based field
+ * matching) is always preserved.
+ *
+ * <p>Because the canonical wrapper may have been created from a
content-identical but
+ * different Avro schema instance, {@code fromAvroSchema(s).getAvroSchema()}
does not
+ * necessarily return {@code s} itself. Canonical instances are shared:
neither the wrapper
+ * nor its underlying Avro schema may be mutated.
Review Comment:
This contract is enforced only by javadoc: `addProp`/`setFields` (and
`HoodieSchemaField.addProp`) remain public and unguarded, and `parse()`'s
owned-instance protection is depth-0 -- navigated children are interned, so
`parse(json).getField(f).schema().addProp(...)` mutates a global canonical, and
a first-seen child's canonical aliases the owned Avro subtree. Verified
mechanically that a mutated canonical keeps being served under its stale
pre-mutation content key. Suggest a transient frozen flag set at intern time
(pool loader, the `AVRO_SCHEMA_CACHE.put` winner, and the uncached fallback),
with mutators throwing when frozen; `parse()` and builders stay unfrozen.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java:
##########
@@ -338,13 +346,42 @@ private HoodieSchema(Schema avroSchema,
List<HoodieSchemaField> fields) {
/**
* Factory method to create HoodieSchema from an Avro schema.
*
+ * <p>Returns the canonical instance for the given schema, converting and
interning it on
+ * first use: distinct Avro schema instances with identical serialized
content converge on
+ * one shared wrapper through {@link HoodieSchemaCache}, with a weak
identity fast path for
+ * the per-record hot path where all records of a file share the same live
{@link Schema}
+ * instance.
+ *
+ * <p>Interning is never lossy: the canonicalization key covers the schema's
full content,
Review Comment:
Nit: this rationale is spelled out in full here, on `equals()`, and in
`HoodieSchemaCache`'s class doc -- keep the full version once in
`HoodieSchemaCache` and a one-liner plus link in the two public javadocs. Also
'never lossy' needs the caveat that non-JSON-serializable schemas are returned
uncached on Avro 1.11, so not every call yields the canonical instance.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java:
##########
@@ -94,6 +96,12 @@
public class HoodieSchema implements Serializable {
private static final long serialVersionUID = 1L;
+ // Avro-identity fast path onto the value-keyed HoodieSchemaCache, backing
fromAvroSchema:
+ // records of one file share the same live Schema instance, so the
per-record hot path is a
+ // single weak-identity hit with no wrapper allocation or type dispatch.
+ private static final Cache<Schema, HoodieSchema> AVRO_SCHEMA_CACHE =
Review Comment:
This 1024-cap cache is now populated by every navigation call, not just the
old top-level intern sites: one parsed 500-col nullable schema is ~1500
identity-distinct Avro nodes (Avro primitives are not shared singletons), ~500
of which enter on a typical read, so two wide tables exceed the cap.
Single-table jobs are fine (TinyLFU keeps the per-record-hot keys), but
multi-table long-lived JVMs sustain insert pressure between hot-key accesses.
Consider `weakKeys` with no `maximumSize` (the usual weak-interner pattern) or
a much larger cap.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCache.java:
##########
@@ -28,21 +29,77 @@
* <p>This is a global cache which works for a JVM lifecycle.
* A collection of schema instances are maintained.
*
+ * <p>This value-keyed pool is the canonicalization mechanism behind
+ * {@link HoodieSchema#fromAvroSchema}, and can also be used directly to
intern schemas
+ * produced without an Avro source (builders, converters).
+ *
+ * <p>Interning is never lossy: entries are keyed on the schema's full
serialized content,
+ * NOT on {@link HoodieSchema#equals}. Avro equality (which HoodieSchema
equality delegates
+ * to) ignores doc strings and aliases, so keying on it would collapse schemas
that differ
+ * only in that metadata and silently drop it -- docs drive catalog sync
column comments and
+ * aliases drive schema-evolution field matching. Schemas that differ in docs
or aliases
+ * intern to distinct canonical instances even though they are {@code
equals()}.
+ *
* <p>NOTE: The schema which is used frequently should be cached through this
cache.
*/
public class HoodieSchemaCache {
// Ensure that there is only one variable instance of the same schema within
an entire JVM lifetime
- private static final LoadingCache<HoodieSchema, HoodieSchema> SCHEMA_CACHE =
- Caffeine.newBuilder().weakValues().maximumSize(1024).build(k -> k);
+ private static final LoadingCache<SchemaContentKey, HoodieSchema>
SCHEMA_CACHE =
+ Caffeine.newBuilder().weakValues().maximumSize(1024).build(key ->
key.schema);
/**
* Get schema variable from global cache. If not found, put it into the
cache and then return it.
*
+ * <p>Two schemas converge on one canonical instance only when their full
serialized form
+ * (including doc strings and aliases) is identical; see the class javadoc.
+ *
+ * <p>A schema that is valid in memory but cannot be serialized to JSON --
e.g. two distinct
+ * nested records that share a name, as some projection/reader paths produce
-- has no content
+ * key, so it is returned uncached instead of interned. Canonicalization is
only a
+ * de-duplication optimization, so skipping it stays correct.
+ *
* @param schema schema to get
* @return if found, return the exist schema variable, otherwise return the
param itself.
*/
public static HoodieSchema intern(HoodieSchema schema) {
- return SCHEMA_CACHE.get(schema);
+ SchemaContentKey key;
+ try {
+ key = new SchemaContentKey(schema);
+ } catch (AvroRuntimeException e) {
+ // Not serializable -> no content key derivable; skip interning rather
than fail the caller.
+ return schema;
+ }
+ return SCHEMA_CACHE.get(key);
+ }
+
+ /**
+ * Content-complete cache key: the serialized JSON form covers doc strings
and aliases that
+ * Avro equality ignores. The wrapper class is part of the key so a
logical-type subclass
+ * (e.g. {@link HoodieSchema.Decimal}) never collapses onto a plain wrapper
of equal content,
+ * which would break downcasts.
+ */
+ private static final class SchemaContentKey {
+ private final HoodieSchema schema;
Review Comment:
The key strongly references the value, so `weakValues()` never fires (same
shape as the old cache), and each entry now also retains the full JSON string
(~80KB for a 1000-col schema) until size eviction -- worst case tens of MB of
dead strings in long-lived multi-table JVMs. A fixed-size digest of
`toString()` plus the wrapper class keeps the doc/alias sensitivity without
retaining the JSON.
##########
hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaCompatibilityChecker.java:
##########
@@ -548,16 +572,23 @@ private SchemaCompatibilityResult
checkTimestampCompatibility(final HoodieSchema
return SchemaCompatibilityResult.compatible();
}
- private SchemaCompatibilityResult checkSchemaNames(final HoodieSchema
reader, final HoodieSchema writer,
- final
Deque<LocationInfo> locations) {
+ /**
+ * Whether fully-qualified schema names are validated at the current
location. Feeds both
+ * {@link #checkSchemaNames} and the memo key, which must agree on the
naming context.
+ */
+ private boolean namesValidatedAt(final Deque<LocationInfo> locations) {
checkState(!locations.isEmpty());
// NOTE: We're only going to validate schema names in following cases
// - This is a top-level schema (ie enclosing one)
// - This is a schema enclosed w/in a union (since in that case
schemas could be
// reverse-looked up by their fully-qualified names)
- boolean shouldCheckNames = checkNaming && (locations.size() == 1 ||
locations.peekLast().type == HoodieSchemaType.UNION);
+ return checkNaming && (locations.size() == 1 ||
locations.peekLast().type == HoodieSchemaType.UNION);
Review Comment:
Note (pre-existing, faithfully preserved by the context key):
`peekLast().type == UNION` applies to all descendants until the next field
push, so a record inside an array/map under any nullable `[null, T]` field is
name-checked while the same structure under a required field is not. Hoisting
name validation to the top-level/union call sites would fix the asymmetry and
let this key revert to a plain pair, but it changes verdicts for the
array-under-union case -- follow-up with sign-off, not this PR.
--
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]
