[common] Fix invalid DecimalType when extracting a variant decimal

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

@@ -138,8 +138,16 @@ public static Object cast(GenericVariant v, DataType dataType, VariantCastArgs c
                     break;
                 case DECIMAL:
                     BigDecimal decimal = v.getDecimal();
-                    int precision = decimal.precision();
+                    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);
                     break;

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

@@ -276,6 +276,38 @@ public void testVariantGet() {
         assertThat(variant.variantGet("$.nullField", DataTypes.BOOLEAN(), castArgs)).isNull();
     }
 
+    @Test
+    public void testVariantGetDecimalWithScaleAbovePrecision() {
+        // precision() counts the digits of the unscaled value, so it is below the scale for any
+        // value under 0.1. The scale 38 case is the widest the reader admits, and it is the
+        // first one here that needs a non compact Decimal.
+        String tiny = "0.00000000000000000000000000000000000001";
+        Variant variant = GenericVariant.fromJson("{\"small\": 0.05, \"tiny\": " + tiny + "}");
+        VariantCastArgs castArgs = new VariantCastArgs(false, ZoneOffset.UTC);
+
+        assertThat(variant.variantGet("$.small", DataTypes.STRING(), castArgs))
+                .isEqualTo(BinaryString.fromString("0.05"));
+        assertThat(variant.variantGet("$.small", DataTypes.DECIMAL(5, 3), castArgs))
+                .isEqualTo(Decimal.fromBigDecimal(new BigDecimal("0.050"), 5, 3));
+        assertThat(variant.variantGet("$.tiny", DataTypes.STRING(), castArgs))
+                .isEqualTo(BinaryString.fromString(tiny));
+        assertThat(variant.variantGet("$.tiny", DataTypes.DECIMAL(38, 38), castArgs))
+                .isEqualTo(Decimal.fromBigDecimal(new BigDecimal(tiny), 38, 38));
+    }
+
+    @Test
+    public void testVariantGetDecimalWithNegativeScale() {
+        // getDecimal() strips trailing zeros, which turns 100.00 into 1E+2, a negative scale
+        Variant variant = GenericVariant.fromJson("{\"round\": 100.00}");
+        VariantCastArgs castArgs = new VariantCastArgs(false, ZoneOffset.UTC);
+
+        // rescaling rather than un-stripping keeps this in step with toJson
+        assertThat(variant.variantGet("$.round", DataTypes.STRING(), castArgs))
+                .isEqualTo(BinaryString.fromString("100"));
+        assertThat(variant.variantGet("$.round", DataTypes.DECIMAL(5, 1), castArgs))
+                .isEqualTo(Decimal.fromBigDecimal(new BigDecimal("100.0"), 5, 1));
+    }
+
     @Test
     public void testObjectFieldOrderingCompatibility() {
         String bmpKey = "\uE000";

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

@@ -291,6 +291,27 @@ void testAssembleAllTypes() {
                                         })));
     }
 
+    @Test
+    public void testAssembleDecimalWithScaleAbovePrecision() {
+        // the unshredded leg extracts through VariantGet, which used to build an invalid
+        // DecimalType for a value below 0.1 or one whose trailing zeros were stripped off
+        GenericVariant v = GenericVariant.fromJson("{\"round\": 100.00, \"small\": 0.05}");
+        VariantCastArgs castArgs = new VariantCastArgs(true, ZoneOffset.UTC);
+
+        VariantSchema variantSchema = buildVariantSchema(variantShreddingSchema(RowType.of()));
+        FieldToExtract[] fieldsToExtract = {
+            buildFieldsToExtract(DataTypes.STRING(), "$.round", castArgs, variantSchema),
+            buildFieldsToExtract(DataTypes.STRING(), "$.small", castArgs, variantSchema)
+        };
+
+        assertThat(
+                        assembleVariantStruct(
+                                castShredded(v, variantSchema), variantSchema, fieldsToExtract))
+                .isEqualTo(
+                        GenericRow.of(
+                                BinaryString.fromString("100"), BinaryString.fromString("0.05")));
+    }
+
     private static void assertVariantStructEquals(
             RowType shreddedType, RowType allTypes, GenericVariant v, GenericRow expected) {
         VariantCastArgs castArgs = new VariantCastArgs(true, ZoneOffset.UTC);