Fix decimal issue

Author: cashmand

parquet-avro/src/test/java/org/apache/parquet/avro/TestWriteVariant.java

@@ -305,10 +305,16 @@ public TestWriteVariant() throws Exception {
       fullVariant(b -> b.appendTimestampNtz(-1234567890L)),
       fullVariant(b -> b.appendDecimal(new BigDecimal("123456.789"))), // decimal4
       fullVariant(b -> b.appendDecimal(new BigDecimal("-123456.789"))), // decimal4
+      fullVariant(b -> b.appendDecimal(new BigDecimal("123456.7"))), // decimal4
+      fullVariant(b -> b.appendDecimal(new BigDecimal("-123456.7"))), // decimal4
       fullVariant(b -> b.appendDecimal(new BigDecimal("123456789.987654321"))), // decimal8
       fullVariant(b -> b.appendDecimal(new BigDecimal("-123456789.987654321"))), // decimal8
+      fullVariant(b -> b.appendDecimal(new BigDecimal("123456789.9876543"))), // decimal8
+      fullVariant(b -> b.appendDecimal(new BigDecimal("-123456789.9876543"))), // decimal8
       fullVariant(b -> b.appendDecimal(new BigDecimal("9876543210.123456789"))), // decimal16
       fullVariant(b -> b.appendDecimal(new BigDecimal("-9876543210.123456789"))), // decimal16
+      fullVariant(b -> b.appendDecimal(new BigDecimal("9876543210.12345678912345"))), // decimal16
+      fullVariant(b -> b.appendDecimal(new BigDecimal("-9876543210.12345678912345"))), // decimal16
       fullVariant(b -> b.appendBinary(ByteBuffer.wrap(new byte[] {0x0a, 0x0b, 0x0c, 0x0d}))),
       fullVariant(b -> b.appendString("iceberg")),
       fullVariant(b -> b.appendTime(1234567890)),

parquet-variant/src/main/java/org/apache/parquet/variant/VariantValueWriter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.parquet.variant;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.io.api.Binary;
@@ -79,7 +80,7 @@ void write(GroupType schema, Variant value) {
     Variant.Type variantType = value.getType();
 
     // Handle typed_value if present
-    if (isTypeCompatible(variantType, typedValueField)) {
+    if (isTypeCompatible(variantType, typedValueField, value)) {
       int typedValueIdx = schema.getFieldIndex("typed_value");
       recordConsumer.startField("typed_value", typedValueIdx);
       ByteBuffer residual = null;
@@ -107,7 +108,21 @@ void write(GroupType schema, Variant value) {
     }
   }
 
-  private boolean isTypeCompatible(Variant.Type variantType, Type typedValueField) {
+  // Return true if the logical type is a decimal with the same scale as the provided value, with enough
+  // precision to hold the value. The provided value must be a decimal.
+  private boolean compatibleDecimalType(Variant value, LogicalTypeAnnotation logicalType) {
+    if (!(logicalType instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation)) {
+      return false;
+    }
+    LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalType =
+        (LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) logicalType;
+
+    BigDecimal decimal = value.getDecimal();
+    return decimal.scale() == decimalType.getScale() &&
+        decimal.precision() <= decimalType.getPrecision();
+  }
+
+  private boolean isTypeCompatible(Variant.Type variantType, Type typedValueField, Variant value) {
     if (typedValueField == null) {
       return false;
     }
@@ -145,14 +160,14 @@ private boolean isTypeCompatible(Variant.Type variantType, Type typedValueField)
         case DOUBLE:
           return primitiveTypeName == PrimitiveType.PrimitiveTypeName.DOUBLE;
         case DECIMAL4:
-          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.INT32
-              && logicalType instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.INT32 &&
+              compatibleDecimalType(value, logicalType);
         case DECIMAL8:
-          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.INT64
-              && logicalType instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.INT64 &&
+              compatibleDecimalType(value, logicalType);
         case DECIMAL16:
-          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.BINARY
-              && logicalType instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+          return primitiveTypeName == PrimitiveType.PrimitiveTypeName.BINARY &&
+              compatibleDecimalType(value, logicalType);
         case DATE:
           return primitiveTypeName == PrimitiveType.PrimitiveTypeName.INT32
               && logicalType instanceof LogicalTypeAnnotation.DateLogicalTypeAnnotation;