[common] Fix inferred variant shredding failing on decimals with trailing zeros

paimon-common/src/main/java/org/apache/paimon/data/variant/InferVariantShreddingSchema.java

@@ -354,16 +354,15 @@ private DataType schemaOf(GenericVariant v, int maxDepth) {
 
             case DECIMAL:
                 BigDecimal dec = v.getDecimal();
-                int decPrecision = dec.precision();
-                int decScale = dec.scale();
-                // Ensure precision is at least scale + 1 to be valid
-                if (decPrecision < decScale) {
-                    decPrecision = decScale;
-                }
-                // Ensure precision is at least 1
-                if (decPrecision == 0) {
-                    decPrecision = 1;
+                if (dec.scale() < 0) {
+                    // getDecimal() strips trailing zeros, which turns 10.0 into 1E+1; a negative
+                    // scale is not a valid Paimon decimal, so fold the exponent back into digits
+                    dec = dec.setScale(0);
                 }
+                int decScale = dec.scale();
+                // precision() counts the digits of the unscaled value, so it is below the scale
+                // for a value under 0.1, which DecimalType rejects
+                int decPrecision = Math.max(dec.precision(), decScale);
                 return DataTypes.DECIMAL(decPrecision, decScale);
 
             case DATE:

paimon-common/src/test/java/org/apache/paimon/data/variant/InferVariantShreddingSchemaTest.java

@@ -179,6 +179,38 @@ void testInferSchemaWithMixedTypes() {
                 .isEqualTo(variantShreddingSchema(expectedType));
     }
 
+    @Test
+    void testInferSchemaWithDecimalTrailingZeros() {
+        RowType schema = RowType.of(new DataType[] {DataTypes.VARIANT()}, new String[] {"v"});
+
+        // getDecimal() strips trailing zeros, so 10.0 and 100.00 arrive with a negative scale
+        // and 0.05 with a precision below its scale; none of them may break inference
+        GenericVariant variant1 =
+                GenericVariant.fromJson(
+                        "{\"price\": 10.0, \"whole\": 100.00, \"tiny\": 0.05,"
+                                + " \"big\": 100000000000000000000}");
+        GenericVariant variant2 =
+                GenericVariant.fromJson(
+                        "{\"price\": 20.5, \"whole\": 7, \"tiny\": 0.001, \"big\": 1}");
+
+        List<InternalRow> rows = Arrays.asList(GenericRow.of(variant1), GenericRow.of(variant2));
+
+        InferVariantShreddingSchema inferrer = defaultInferVariantShreddingSchema(schema);
+        RowType inferredSchema = inferrer.inferSchema(rows);
+
+        RowType expectedType =
+                RowType.of(
+                        new DataType[] {
+                            DataTypes.DECIMAL(38, 0),
+                            DataTypes.DECIMAL(18, 1),
+                            DataTypes.DECIMAL(18, 3),
+                            DataTypes.BIGINT()
+                        },
+                        new String[] {"big", "price", "tiny", "whole"});
+        assertThat(inferredSchema.getField("v").type())
+                .isEqualTo(variantShreddingSchema(expectedType));
+    }
+
     @Test
     void testInferSchemaWithNullValues() {
         // Schema: row<v: variant>

paimon-format/src/test/java/org/apache/paimon/format/parquet/writer/InferVariantShreddingWriteTest.java

@@ -432,6 +432,39 @@ public void testInferSchemaWithMixedTypes() throws Exception {
         verifyShreddingSchema(expectShreddedType);
     }
 
+    @Test
+    public void testInferSchemaWithDecimalTrailingZeros() throws Exception {
+        ParquetFileFormat format = createFormat();
+        RowType writeType = DataTypes.ROW(DataTypes.FIELD(0, "v", DataTypes.VARIANT()));
+
+        // 10.0 and 100.00 strip to a negative scale, which used to fail the whole file write
+        FormatWriterFactory factory = format.createWriterFactory(writeType);
+        writeRows(
+                factory,
+                GenericRow.of(GenericVariant.fromJson("{\"price\":10.0,\"whole\":100.00}")),
+                GenericRow.of(GenericVariant.fromJson("{\"price\":20.5,\"whole\":7}")));
+
+        RowType expectShreddedType =
+                RowType.of(
+                        new DataType[] {DataTypes.DECIMAL(18, 1), DataTypes.BIGINT()},
+                        new String[] {"price", "whole"});
+        verifyShreddingSchema(expectShreddedType);
+
+        List<InternalRow> result = readRows(format, writeType);
+        assertThat(result.get(0).getVariant(0).toJson()).isEqualTo("{\"price\":10,\"whole\":100}");
+        assertThat(result.get(1).getVariant(0).toJson()).isEqualTo("{\"price\":20.5,\"whole\":7}");
+
+        RowType variantRowType =
+                VariantMetadataUtils.VariantRowTypeBuilder.builder()
+                        .field(DataTypes.DOUBLE(), "$.price")
+                        .field(DataTypes.BIGINT(), "$.whole")
+                        .build();
+        RowType readType = DataTypes.ROW(DataTypes.FIELD(0, "v", variantRowType));
+        List<InternalRow> result2 = readRows(format, readType);
+        assertThat(result2.get(0)).isEqualTo(GenericRow.of(GenericRow.of(10.0, 100L)));
+        assertThat(result2.get(1)).isEqualTo(GenericRow.of(GenericRow.of(20.5, 7L)));
+    }
+
     @Test
     public void testInferSchemaWithNullValues() throws Exception {
         ParquetFileFormat format = createFormat();

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala

@@ -431,6 +431,20 @@ abstract class VariantTestBase extends PaimonSparkTestBase {
     }
   }
 
+  test("Paimon Variant: decimals with trailing zeros under inferred shredding") {
+    sql("CREATE TABLE T (id INT, v VARIANT)")
+    // 10.0 and 100.00 strip to a negative scale, which used to fail inferred-shredding writes
+    sql("""INSERT INTO T VALUES
+          | (1, parse_json('{"price":10.0,"whole":100.00}')),
+          | (2, parse_json('{"price":20.5,"whole":7}'))
+          |""".stripMargin)
+
+    checkAnswer(
+      sql(
+        "SELECT id, variant_get(v, '$.price', 'double'), variant_get(v, '$.whole', 'bigint') FROM T ORDER BY id"),
+      Seq(Row(1, 10.0, 100L), Row(2, 20.5, 7L)))
+  }
+
   test("Paimon Variant: read and write variant with null value") {
     withTable("source_tbl", "target_tbl") {
       sql("CREATE TABLE source_tbl (id INT, js STRING) USING paimon")