This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit c4dff62194ab837747879ba2a93c39b9383f5c42 Author: Xiangyi Zhu <[email protected]> AuthorDate: Tue Sep 15 11:18:40 2026 +0800 [common] Read a shredded variant decimal as the same string as an unshredded one (#9818) --- .../paimon/data/variant/BaseVariantReader.java | 7 ++- .../org/apache/paimon/data/variant/VariantGet.java | 37 +++++++++------ .../data/variant/PaimonShreddingUtilsTest.java | 49 ++++++++++++++++++++ .../parquet/reader/VariantShreddingReadTest.java | 52 ++++++++++++++++++++++ .../apache/paimon/spark/sql/VariantTestBase.scala | 16 +++++++ 5 files changed, 147 insertions(+), 14 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java b/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java index ce3c4036aa..0d0177c607 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/BaseVariantReader.java @@ -21,6 +21,7 @@ package org.apache.paimon.data.variant; import org.apache.paimon.casting.CastExecutor; import org.apache.paimon.casting.CastExecutors; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; @@ -518,11 +519,15 @@ public class BaseVariantReader { } else if (scalaType.equals(DataTypes.BYTES())) { i = row.getBinary(typedValueIdx); } else if (scalaType instanceof DecimalType) { - i = + Decimal decimal = row.getDecimal( typedValueIdx, ((DecimalType) scalaType).getPrecision(), ((DecimalType) scalaType).getScale()); + // The typed_value carries the scale of the file schema, e.g. 10.0 as + // DECIMAL(18, 1), while the unshredded leg casts from the stripped value; cast + // from the same normalized decimal so both legs read "10" rather than "10.0". + i = noNeedCast ? decimal : VariantGet.normalizedDecimal(decimal.toBigDecimal()); } else if (scalaType instanceof DateType) { i = row.getInt(typedValueIdx); } else if (scalaType instanceof TimestampType) { diff --git a/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java index 37aa20b4af..7e355939eb 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantGet.java @@ -137,19 +137,9 @@ public class VariantGet { inputType = DataTypes.DOUBLE(); break; case DECIMAL: - BigDecimal decimal = v.getDecimal(); - if (decimal.scale() < 0) { - // stripTrailingZeros folds trailing zeros into a negative exponent, - // and a negative scale is not a Paimon decimal - decimal = decimal.setScale(0); - } - int scale = decimal.scale(); - // precision() counts the digits of the unscaled value, so it is smaller than - // the scale for a value below 0.1, which DecimalType rejects. The variant - // writer caps both at MAX_DECIMAL16_PRECISION, so this stays in range. - int precision = Math.max(decimal.precision(), scale); - input = Decimal.fromBigDecimal(decimal, precision, scale); - inputType = DataTypes.DECIMAL(precision, scale); + Decimal decimal = normalizedDecimal(v.getDecimal()); + input = decimal; + inputType = DataTypes.DECIMAL(decimal.precision(), decimal.scale()); break; case DATE: input = (int) v.getLong(); @@ -194,6 +184,27 @@ public class VariantGet { } } + /** + * The decimal a variant scalar is cast from: trailing zeros stripped, the way {@code toJson} + * renders it, with a precision and scale that {@code DecimalType} accepts. The shredded reader + * applies the same normalization to a {@code typed_value} decimal, whose scale comes from the + * file schema, so that a cast yields the same result for a plain and a shredded file. + */ + static Decimal normalizedDecimal(BigDecimal decimal) { + decimal = decimal.stripTrailingZeros(); + if (decimal.scale() < 0) { + // stripTrailingZeros folds trailing zeros into a negative exponent, and a negative + // scale is not a Paimon decimal + decimal = decimal.setScale(0); + } + int scale = decimal.scale(); + // precision() counts the digits of the unscaled value, so it is smaller than the scale + // for a value below 0.1, which DecimalType rejects. The variant writer caps both at + // MAX_DECIMAL16_PRECISION, so this stays in range. + int precision = Math.max(decimal.precision(), scale); + return Decimal.fromBigDecimal(decimal, precision, scale); + } + public static Object invalidCast(Variant v, DataType dataType, VariantCastArgs castArgs) { if (castArgs.failOnError()) { throw new RuntimeException( diff --git a/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java index b1952357d7..a65d88c6dc 100644 --- a/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/data/variant/PaimonShreddingUtilsTest.java @@ -291,6 +291,55 @@ public class PaimonShreddingUtilsTest { }))); } + @Test + public void testShreddedDecimalCastsLikeUnshredded() { + // A shredded typed_value carries the scale of the file schema, e.g. 10.0 as + // DECIMAL(18, 1), while the unshredded leg strips trailing zeros; both legs must + // produce the same string, and numeric targets must stay unaffected. + GenericVariant v = + GenericVariant.fromJson( + "{\"price\": 10.0, \"amount\": 1.50, \"tiny\": 0.05, \"zero\": 0.00}"); + VariantCastArgs castArgs = new VariantCastArgs(true, ZoneOffset.UTC); + + RowType shredded = + RowType.of( + new DataType[] { + DataTypes.DECIMAL(18, 2), + DataTypes.DECIMAL(18, 1), + DataTypes.DECIMAL(18, 2), + DataTypes.DECIMAL(18, 2) + }, + new String[] {"amount", "price", "tiny", "zero"}); + RowType unshredded = RowType.of(); + + for (RowType shape : new RowType[] {shredded, unshredded}) { + VariantSchema variantSchema = buildVariantSchema(variantShreddingSchema(shape)); + FieldToExtract[] fieldsToExtract = { + buildFieldsToExtract(DataTypes.STRING(), "$.price", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.STRING(), "$.amount", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.STRING(), "$.tiny", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.STRING(), "$.zero", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.DECIMAL(10, 2), "$.price", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.DOUBLE(), "$.amount", castArgs, variantSchema), + buildFieldsToExtract(DataTypes.BIGINT(), "$.price", castArgs, variantSchema) + }; + + assertThat( + assembleVariantStruct( + castShredded(v, variantSchema), variantSchema, fieldsToExtract)) + .as("shape %s", shape) + .isEqualTo( + GenericRow.of( + BinaryString.fromString("10"), + BinaryString.fromString("1.5"), + BinaryString.fromString("0.05"), + BinaryString.fromString("0"), + Decimal.fromBigDecimal(new BigDecimal("10.00"), 10, 2), + 1.5, + 10L)); + } + } + @Test public void testAssembleDecimalWithScaleAbovePrecision() { // the unshredded leg extracts through VariantGet, which used to build an invalid diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java index dffb879fdf..34c4bdd382 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VariantShreddingReadTest.java @@ -19,6 +19,7 @@ package org.apache.paimon.format.parquet.reader; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -47,6 +48,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; import java.util.UUID; @@ -492,6 +494,56 @@ public class VariantShreddingReadTest { assertThat(result.get(0).getRow(0, 1).getInt(0)).isEqualTo(2); } + @ParameterizedTest + @ValueSource( + strings = { + "null", + "{\"type\":\"ROW\",\"fields\":[{\"name\":\"v\",\"type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"price\",\"type\":\"DECIMAL(18, 1)\"},{\"name\":\"amount\",\"type\":\"DECIMAL(18, 2)\"}]}}]}" + }) + public void testReadDecimalAsStringConsistently(String shreddingSchema) throws Exception { + // A shredded typed_value keeps the scale of the file schema while the unshredded value + // is read with trailing zeros stripped; both layouts must extract the same string. + Options options = new Options(); + if (!shreddingSchema.equals("null")) { + options.set("parquet.variant.shreddingSchema", shreddingSchema); + } + ParquetFileFormat format = + new ParquetFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); + + RowType writeType = DataTypes.ROW(DataTypes.FIELD(0, "v", DataTypes.VARIANT())); + writeRows( + format.createWriterFactory(writeType), + GenericRow.of(GenericVariant.fromJson("{\"price\":10.0,\"amount\":1.50}")), + GenericRow.of(GenericVariant.fromJson("{\"price\":2.5,\"amount\":0.00}"))); + + RowType readType = + DataTypes.ROW( + DataTypes.FIELD( + 0, + "v", + VariantMetadataUtils.VariantRowTypeBuilder.builder() + .field(DataTypes.STRING(), "$.price") + .field(DataTypes.STRING(), "$.amount") + .field(DataTypes.DECIMAL(10, 2), "$.price") + .field(DataTypes.DOUBLE(), "$.amount") + .build())); + List<InternalRow> result = readRows(format, readType); + assertThat(result.get(0).getRow(0, 4)) + .isEqualTo( + GenericRow.of( + BinaryString.fromString("10"), + BinaryString.fromString("1.5"), + Decimal.fromBigDecimal(new BigDecimal("10.00"), 10, 2), + 1.5)); + assertThat(result.get(1).getRow(0, 4)) + .isEqualTo( + GenericRow.of( + BinaryString.fromString("2.5"), + BinaryString.fromString("0"), + Decimal.fromBigDecimal(new BigDecimal("2.50"), 10, 2), + 0.0)); + } + protected List<InternalRow> readRows(ParquetFileFormat format, RowType rowType) throws IOException { List<InternalRow> result = new ArrayList<>(); diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala index 0878e2c00a..70b58e0fe9 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala @@ -1069,6 +1069,22 @@ abstract class VariantTestBase extends PaimonSparkTestBase { checkAnswer(df, Seq(Row(0, null), Row(1, 1))) } + test("Paimon Variant pushdown: decimal extracted as string is the same for every layout") { + sql("CREATE TABLE T (id INT, v VARIANT)") + // Mixed scales make an inferred shredding schema DECIMAL(18, 2) / DECIMAL(18, 1), so the + // typed_value keeps trailing zeros (1.50, 0.0) that the unshredded value strips. + sql("""INSERT INTO T VALUES + | (1, parse_json('{"price":1.50,"amount":0.00}')), + | (2, parse_json('{"price":0.05,"amount":2.5}')) + |""".stripMargin) + + checkAnswer( + sql( + "SELECT id, variant_get(v, '$.price', 'string'), variant_get(v, '$.amount', 'string') FROM T ORDER BY id"), + Seq(Row(1, "1.5", "0"), Row(2, "0.05", "2.5")) + ) + } + test("Paimon Variant pushdown: nested variant column inside a struct") { assume(gteqSpark4_1) sql("CREATE TABLE T (id INT, nested STRUCT<v: VARIANT, x: INT>)")
