From f222ba6024dd9728f2becb3b59d5d1f65bc7f29b Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Tue, 1 Sep 2026 23:01:55 +1000 Subject: [PATCH 1/3] [common] Keep the selected variant schema when evidence is gone finalizeAdaptiveSchema falls back to the previously selected schema when a node has no evidence in the current file, then runs it through the RowType branch as if it were evidence. Selected schemas carry no per-field counts, so getFieldCount throws IllegalStateException and the whole file fails to write. Reaching it needs a node that drifted across a type family, which degrades its combined evidence to VARIANT while its selected schema is a ROW, and that is then absent from the next file. Return the selected schema as it is instead: it is already a finalized selection and has nothing left to threshold. --- .../variant/InferVariantShreddingSchema.java | 6 ++- .../InferVariantShreddingSchemaTest.java | 41 +++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java index e983f28fdd5b..44d285b228af 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java @@ -574,7 +574,11 @@ private DataType finalizeAdaptiveSchema( if (current != null && !(current instanceof VariantType)) { combined = current; } else if (previousSelected != null) { - combined = previousSelected; + // This node has no evidence in the current file. A previously selected schema is + // not + // evidence - its fields carry no counts - so it cannot be run through admission and + // retention again; carry it forward unchanged. + return previousSelected; } else { return DataTypes.VARIANT(); } diff --git a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java index 1a38025e7bee..a837bba33011 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java @@ -31,6 +31,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -315,6 +316,46 @@ void testAdaptiveInferenceWidensScalarSelectedFromPriorEvidence() { new String[] {"historical"}))); } + /** + * A node can drift from scalar to object, which degrades its combined evidence to VARIANT while + * the selected schema for it is a ROW, and then be absent from the next file. With no evidence + * to fall back on, the selected schema is the only thing left - but it is not evidence: its + * fields carry no counts, so it cannot be run through admission and retention a second time. + */ + @Test + void testAdaptiveInferenceKeepsSelectedRowWhenEvidenceDegradedAndNodeIsAbsent() { + RowType schema = RowType.of(new DataType[] {DataTypes.VARIANT()}, new String[] {"v"}); + VariantShreddingInferenceSession session = + new VariantShreddingInferenceSession( + new InferVariantShreddingSchema(schema, 300, 50, 0.1), 256, 0.1, 0.05); + + session.inferSchema( + Collections.singletonList( + GenericRow.of(GenericVariant.fromJson("{\"k\":1,\"p\":5}")))); + session.commitPendingInference(); + session.inferSchema( + Collections.singletonList( + GenericRow.of(GenericVariant.fromJson("{\"k\":1,\"p\":{\"x\":1}}")))); + session.commitPendingInference(); + + RowType afterAbsence = + session.inferSchema( + Collections.singletonList( + GenericRow.of(GenericVariant.fromJson("{\"k\":1}")))); + + assertThat(afterAbsence.getField("v").type()) + .isEqualTo( + variantShreddingSchema( + RowType.of( + new DataType[] { + DataTypes.BIGINT(), + RowType.of( + new DataType[] {DataTypes.BIGINT()}, + new String[] {"x"}) + }, + new String[] {"k", "p"}))); + } + @Test void testInferSchemaWithDeepNesting() { // Schema: row From 6c52495cdcb66f127e108ce7a8069b8db16fd5e1 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Wed, 2 Sep 2026 10:23:37 +1000 Subject: [PATCH 2/3] Debit the shared width budget for a retained schema Returning the selected schema early only spent the single unit taken at method entry, so its descendants occupied width that maxSchemaWidth never counted and a later variant column could overspend the shared budget. Walk the retained schema instead, charging one unit per node and dropping what no longer fits, which is what the evidence-driven walk does. --- .../variant/InferVariantShreddingSchema.java | 41 ++++++++++++++++--- .../InferVariantShreddingSchemaTest.java | 41 +++++++++++++++++++ 2 files changed, 77 insertions(+), 5 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java index 44d285b228af..f49bce6d61a3 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java @@ -574,11 +574,11 @@ private DataType finalizeAdaptiveSchema( if (current != null && !(current instanceof VariantType)) { combined = current; } else if (previousSelected != null) { - // This node has no evidence in the current file. A previously selected schema is - // not - // evidence - its fields carry no counts - so it cannot be run through admission and - // retention again; carry it forward unchanged. - return previousSelected; + // This node has no evidence in the current file. A previously selected schema + // is not evidence - its fields carry no counts - so it cannot be run through + // admission and retention again; carry it forward as it is, debiting the shared + // width budget for what it holds. + return retainSelectedSchema(previousSelected, maxFields); } else { return DataTypes.VARIANT(); } @@ -658,6 +658,37 @@ private DataType finalizeAdaptiveSchema( return selectScalarType(combined, current, previousSelected); } + /** + * Carries a previously selected schema forward for a node the current file has no evidence for. + * The selection is already final, so nothing is re-thresholded, but its nodes still consume the + * shared width budget and are dropped once it runs out, on the same terms as the + * evidence-driven walk above: one unit per node, VARIANT once the budget is gone. + */ + private DataType retainSelectedSchema(DataType selected, MaxFields maxFields) { + if (selected instanceof RowType) { + List fields = new ArrayList<>(); + for (DataField field : ((RowType) selected).getFields()) { + maxFields.remaining--; + if (maxFields.remaining <= 0) { + break; + } + DataType retained = retainSelectedSchema(field.type(), maxFields); + fields.add(new DataField(fields.size(), field.name(), retained)); + } + return fields.isEmpty() ? DataTypes.VARIANT() : new RowType(fields); + } + if (selected instanceof ArrayType) { + maxFields.remaining--; + if (maxFields.remaining <= 0) { + return DataTypes.VARIANT(); + } + return new ArrayType( + retainSelectedSchema(((ArrayType) selected).getElementType(), maxFields)); + } + maxFields.remaining--; + return selected; + } + private DataType selectScalarType( DataType combined, DataType current, DataType previousSelected) { if (current == null) { diff --git a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java index a837bba33011..545ca97fc39b 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java @@ -356,6 +356,47 @@ void testAdaptiveInferenceKeepsSelectedRowWhenEvidenceDegradedAndNodeIsAbsent() new String[] {"k", "p"}))); } + /** + * maxSchemaWidth is one budget shared by every variant column. A schema carried forward for a + * node with no evidence still occupies it, so a later column must not get to spend what the + * carried-forward schema is holding. + */ + @Test + void testRetainedSchemaStillConsumesTheSharedWidthBudget() { + RowType schema = + RowType.of( + new DataType[] {DataTypes.VARIANT(), DataTypes.VARIANT()}, + new String[] {"a", "b"}); + VariantShreddingInferenceSession session = + new VariantShreddingInferenceSession( + new InferVariantShreddingSchema(schema, 8, 50, 0.1), 256, 0.1, 0.05); + + session.inferSchema( + Collections.singletonList( + GenericRow.of( + GenericVariant.fromJson("{\"p\":5}"), + GenericVariant.fromJson("{\"q\":1}")))); + session.commitPendingInference(); + session.inferSchema( + Collections.singletonList( + GenericRow.of( + GenericVariant.fromJson("{\"p\":{\"x\":1}}"), + GenericVariant.fromJson("{\"q\":1}")))); + session.commitPendingInference(); + + RowType afterAbsence = + session.inferSchema( + Collections.singletonList( + GenericRow.of( + GenericVariant.fromJson("{}"), + GenericVariant.fromJson("{\"q\":1,\"r\":1}")))); + + // "a" keeps ROW under "p", and the budget it holds leaves "r" untyped in "b". + assertThat(afterAbsence.getField("b").type().toString()) + .contains("`q` ROW<`value` BYTES, `typed_value` BIGINT>") + .doesNotContain("`r` ROW<`value` BYTES, `typed_value`"); + } + @Test void testInferSchemaWithDeepNesting() { // Schema: row From 20d2ec772928b9693bec708f184238c597e10fe5 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Wed, 2 Sep 2026 10:36:25 +1000 Subject: [PATCH 3/3] Keep the retained field at the last budget unit The helper decremented before its guard, so the field that consumed the final unit was dropped rather than kept with a VARIANT child, and a retained node holding only that field collapsed entirely. Enter a child only while budget remains, spend the unit on entry, and downgrade an exhausted child to VARIANT with its field or array container intact, which is what the evidence-driven walk does. --- .../variant/InferVariantShreddingSchema.java | 24 ++++++++----- .../InferVariantShreddingSchemaTest.java | 34 +++++++++++++++++++ 2 files changed, 49 insertions(+), 9 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java index f49bce6d61a3..47e26a5ebb42 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java @@ -661,29 +661,35 @@ private DataType finalizeAdaptiveSchema( /** * Carries a previously selected schema forward for a node the current file has no evidence for. * The selection is already final, so nothing is re-thresholded, but its nodes still consume the - * shared width budget and are dropped once it runs out, on the same terms as the - * evidence-driven walk above: one unit per node, VARIANT once the budget is gone. + * shared width budget. The entry unit for this node has already been spent by the caller, so + * this mirrors what finalizeAdaptiveSchema does from that point on: a child is entered only + * while budget remains, entering it spends one unit, and a child that exhausts the budget + * becomes VARIANT while its field or array container is still kept. */ private DataType retainSelectedSchema(DataType selected, MaxFields maxFields) { if (selected instanceof RowType) { List fields = new ArrayList<>(); for (DataField field : ((RowType) selected).getFields()) { - maxFields.remaining--; if (maxFields.remaining <= 0) { break; } - DataType retained = retainSelectedSchema(field.type(), maxFields); + maxFields.remaining--; + DataType retained = + maxFields.remaining <= 0 + ? DataTypes.VARIANT() + : retainSelectedSchema(field.type(), maxFields); fields.add(new DataField(fields.size(), field.name(), retained)); } return fields.isEmpty() ? DataTypes.VARIANT() : new RowType(fields); } if (selected instanceof ArrayType) { maxFields.remaining--; - if (maxFields.remaining <= 0) { - return DataTypes.VARIANT(); - } - return new ArrayType( - retainSelectedSchema(((ArrayType) selected).getElementType(), maxFields)); + DataType element = + maxFields.remaining <= 0 + ? DataTypes.VARIANT() + : retainSelectedSchema( + ((ArrayType) selected).getElementType(), maxFields); + return new ArrayType(element); } maxFields.remaining--; return selected; diff --git a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java index 545ca97fc39b..1778136d7cca 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java @@ -397,6 +397,40 @@ void testRetainedSchemaStillConsumesTheSharedWidthBudget() { .doesNotContain("`r` ROW<`value` BYTES, `typed_value`"); } + /** + * At the last budget unit the evidence-driven walk still keeps the field and downgrades its + * child to VARIANT. Retaining a schema has to do the same rather than drop the field, or the + * whole retained node collapses. + */ + @Test + void testRetainedSchemaKeepsItsFieldsAtTheLastBudgetUnit() { + RowType schema = + RowType.of( + new DataType[] {DataTypes.VARIANT(), DataTypes.VARIANT()}, + new String[] {"a", "b"}); + VariantShreddingInferenceSession session = + new VariantShreddingInferenceSession( + new InferVariantShreddingSchema(schema, 7, 50, 0.1), 256, 0.1, 0.05); + + session.inferSchema( + Collections.singletonList( + GenericRow.of(GenericVariant.fromJson("1"), GenericVariant.fromJson("5")))); + session.commitPendingInference(); + session.inferSchema( + Collections.singletonList( + GenericRow.of( + GenericVariant.fromJson("1"), + GenericVariant.fromJson("{\"q\":1}")))); + session.commitPendingInference(); + + RowType afterAbsence = + session.inferSchema( + Collections.singletonList( + GenericRow.of(GenericVariant.fromJson("{\"x\":1,\"y\":1}"), null))); + + assertThat(afterAbsence.getField("b").type().toString()).contains("`q`"); + } + @Test void testInferSchemaWithDeepNesting() { // Schema: row