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..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 @@ -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 as it is, debiting the shared + // width budget for what it holds. + return retainSelectedSchema(previousSelected, maxFields); } else { return DataTypes.VARIANT(); } @@ -654,6 +658,43 @@ 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. 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()) { + if (maxFields.remaining <= 0) { + break; + } + 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--; + DataType element = + maxFields.remaining <= 0 + ? DataTypes.VARIANT() + : retainSelectedSchema( + ((ArrayType) selected).getElementType(), maxFields); + return new ArrayType(element); + } + 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 1a38025e7bee..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 @@ -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,121 @@ 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"}))); + } + + /** + * 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`"); + } + + /** + * 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