Respect Parquet Logical type when reading statistics

Praveen2112 · Praveen2112 · commit af38a3c0f14f · 2025-07-18T16:48:35.000+02:00
diff --git a/lib/trino-parquet/src/main/java/io/trino/parquet/predicate/TupleDomainParquetPredicate.java b/lib/trino-parquet/src/main/java/io/trino/parquet/predicate/TupleDomainParquetPredicate.java
@@ -28,7 +28,9 @@
 import io.trino.spi.predicate.SortedRangeSet;
 import io.trino.spi.predicate.TupleDomain;
 import io.trino.spi.predicate.ValueSet;
+import io.trino.spi.type.DecimalConversions;
 import io.trino.spi.type.DecimalType;
+import io.trino.spi.type.Decimals;
 import io.trino.spi.type.Int128;
 import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.Type;
@@ -48,6 +50,7 @@
 import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
 import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
 import org.apache.parquet.schema.PrimitiveType;
 import org.joda.time.DateTimeZone;
@@ -68,10 +71,12 @@
 import static io.trino.parquet.ParquetTimestampUtils.decodeInt96Timestamp;
 import static io.trino.parquet.ParquetTypeUtils.getShortDecimalValue;
 import static io.trino.parquet.predicate.PredicateUtils.isStatisticsOverflow;
+import static io.trino.parquet.reader.ColumnReaderFactory.isDecimalRescaled;
 import static io.trino.plugin.base.type.TrinoTimestampEncoderFactory.createTimestampEncoder;
 import static io.trino.spi.type.BigintType.BIGINT;
 import static io.trino.spi.type.BooleanType.BOOLEAN;
 import static io.trino.spi.type.DateType.DATE;
+import static io.trino.spi.type.Decimals.longTenToNth;
 import static io.trino.spi.type.DoubleType.DOUBLE;
 import static io.trino.spi.type.IntegerType.INTEGER;
 import static io.trino.spi.type.RealType.REAL;
@@ -377,11 +382,8 @@ private static Domain getDomain(
             SortedRangeSet.Builder rangesBuilder = SortedRangeSet.builder(type, minimums.size());
             if (decimalType.isShort()) {
                 for (int i = 0; i < minimums.size(); i++) {
-                    Object min = minimums.get(i);
-                    Object max = maximums.get(i);
-
-                    long minValue = min instanceof Slice minSlice ? getShortDecimalValue(minSlice.getBytes()) : asLong(min);
-                    long maxValue = max instanceof Slice maxSlice ? getShortDecimalValue(maxSlice.getBytes()) : asLong(max);
+                    long minValue = getShortDecimal(minimums.get(i), decimalType, column);
+                    long maxValue = getShortDecimal(maximums.get(i), decimalType, column);
 
                     if (isStatisticsOverflow(type, minValue, maxValue)) {
                         return Domain.create(ValueSet.all(type), hasNullValue);
@@ -392,11 +394,8 @@ private static Domain getDomain(
             }
             else {
                 for (int i = 0; i < minimums.size(); i++) {
-                    Object min = minimums.get(i);
-                    Object max = maximums.get(i);
-
-                    Int128 minValue = min instanceof Slice minSlice ? Int128.fromBigEndian(minSlice.getBytes()) : Int128.valueOf(asLong(min));
-                    Int128 maxValue = max instanceof Slice maxSlice ? Int128.fromBigEndian(maxSlice.getBytes()) : Int128.valueOf(asLong(max));
+                    Int128 minValue = getLongDecimal(minimums.get(i), decimalType, column);
+                    Int128 maxValue = getLongDecimal(maximums.get(i), decimalType, column);
 
                     rangesBuilder.addRangeInclusive(minValue, maxValue);
                 }
@@ -494,6 +493,61 @@ private static Domain getDomain(
         return Domain.create(ValueSet.all(type), hasNullValue);
     }
 
+    private static long getShortDecimal(Object value, DecimalType columnType, ColumnDescriptor column)
+    {
+        LogicalTypeAnnotation annotation = column.getPrimitiveType().getLogicalTypeAnnotation();
+
+        if (annotation instanceof DecimalLogicalTypeAnnotation decimalAnnotation) {
+            if (isDecimalRescaled(decimalAnnotation, columnType)) {
+                if (decimalAnnotation.getPrecision() <= Decimals.MAX_SHORT_PRECISION) {
+                    long rescale = longTenToNth(Math.abs(columnType.getScale() - decimalAnnotation.getScale()));
+                    return DecimalConversions.shortToShortCast(
+                            value instanceof Slice slice ? getShortDecimalValue(slice.getBytes()) : asLong(value),
+                            decimalAnnotation.getPrecision(),
+                            decimalAnnotation.getScale(),
+                            columnType.getPrecision(),
+                            columnType.getScale(),
+                            rescale,
+                            rescale / 2);
+                }
+                Int128 int128Representation = value instanceof Slice minSlice ? Int128.fromBigEndian(minSlice.getBytes()) : Int128.valueOf(asLong(value));
+                return DecimalConversions.longToShortCast(
+                        int128Representation,
+                        decimalAnnotation.getPrecision(),
+                        decimalAnnotation.getScale(),
+                        columnType.getPrecision(),
+                        columnType.getScale());
+            }
+        }
+        return value instanceof Slice slice ? getShortDecimalValue(slice.getBytes()) : asLong(value);
+    }
+
+    private static Int128 getLongDecimal(Object value, DecimalType columnType, ColumnDescriptor column)
+    {
+        LogicalTypeAnnotation annotation = column.getPrimitiveType().getLogicalTypeAnnotation();
+
+        if (annotation instanceof DecimalLogicalTypeAnnotation decimalAnnotation) {
+            if (isDecimalRescaled(decimalAnnotation, columnType)) {
+                if (decimalAnnotation.getPrecision() <= Decimals.MAX_SHORT_PRECISION) {
+                    return DecimalConversions.shortToLongCast(
+                            value instanceof Slice slice ? getShortDecimalValue(slice.getBytes()) : asLong(value),
+                            decimalAnnotation.getPrecision(),
+                            decimalAnnotation.getScale(),
+                            columnType.getPrecision(),
+                            columnType.getScale());
+                }
+                Int128 int128Representation = value instanceof Slice slice ? Int128.fromBigEndian(slice.getBytes()) : Int128.valueOf(asLong(value));
+                return DecimalConversions.longToLongCast(
+                        int128Representation,
+                        decimalAnnotation.getPrecision(),
+                        decimalAnnotation.getScale(),
+                        columnType.getPrecision(),
+                        columnType.getScale());
+            }
+        }
+        return value instanceof Slice slice ? Int128.fromBigEndian(slice.getBytes()) : Int128.valueOf(asLong(value));
+    }
+
     @VisibleForTesting
     public static Domain getDomain(
             Type type,
diff --git a/lib/trino-parquet/src/main/java/io/trino/parquet/reader/ColumnReaderFactory.java b/lib/trino-parquet/src/main/java/io/trino/parquet/reader/ColumnReaderFactory.java
@@ -330,7 +330,7 @@ public Optional<Boolean> visit(UUIDLogicalTypeAnnotation uuidLogicalType)
                 .orElse(FALSE);
     }
 
-    private static boolean isDecimalRescaled(DecimalLogicalTypeAnnotation decimalAnnotation, DecimalType trinoType)
+    public static boolean isDecimalRescaled(DecimalLogicalTypeAnnotation decimalAnnotation, DecimalType trinoType)
     {
         return decimalAnnotation.getPrecision() != trinoType.getPrecision()
                 || decimalAnnotation.getScale() != trinoType.getScale();
diff --git a/lib/trino-parquet/src/test/java/io/trino/parquet/TestTupleDomainParquetPredicate.java b/lib/trino-parquet/src/test/java/io/trino/parquet/TestTupleDomainParquetPredicate.java
@@ -101,6 +101,7 @@
 import static java.util.Collections.singletonList;
 import static java.util.Collections.singletonMap;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.decimalType;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
@@ -276,6 +277,43 @@ public void testShortDecimalWithNoScale()
                 .withMessage("Malformed Parquet file. Corrupted statistics for column \"[] required int32 ShortDecimalColumnWithNoScale\": [min: 100, max: 10, num_nulls: 0] [testFile]");
     }
 
+    @Test
+    public void testShortDecimalWithLongDecimalAnnotation()
+            throws Exception
+    {
+        ColumnDescriptor columnDescriptor = createColumnDescriptor(FIXED_LEN_BYTE_ARRAY, decimalType(2, 38), "ShortDecimalColumnWithDecimalAnnotation");
+        BigInteger maximum = new BigInteger("12345");
+
+        Type type = createDecimalType(5, 2);
+        assertThat(getDomain(columnDescriptor, type, 0, null, ID, UTC)).isEqualTo(all(type));
+        assertThat(getDomain(columnDescriptor, type, 10, binaryColumnStats(maximum, maximum), ID, UTC)).isEqualTo(singleValue(type, 12345L));
+
+        assertThat(getDomain(columnDescriptor, type, 10, binaryColumnStats(0L, 100L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, 0L, true, 100L, true)), false));
+        assertThat(getDomain(columnDescriptor, type, 10, intColumnStats(0, 100), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, 0L, true, 100L, true)), false));
+
+        type = createDecimalType(15, 2);
+        assertThat(getDomain(columnDescriptor, type, 0, null, ID, UTC)).isEqualTo(all(type));
+        assertThat(getDomain(columnDescriptor, type, 10, binaryColumnStats(maximum, maximum), ID, UTC)).isEqualTo(singleValue(type, 12345L));
+
+        assertThat(getDomain(columnDescriptor, type, 10, binaryColumnStats(0L, 100L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, 0L, true, 100L, true)), false));
+        assertThat(getDomain(columnDescriptor, type, 10, intColumnStats(0, 100), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, 0L, true, 100L, true)), false));
+
+        Type typeWithDifferentScale = createDecimalType(5, 1);
+        assertThat(getDomain(columnDescriptor, typeWithDifferentScale, 0, null, ID, UTC)).isEqualTo(all(typeWithDifferentScale));
+
+        assertThat(getDomain(columnDescriptor, typeWithDifferentScale, 10, longColumnStats(10012L, 10012L), ID, UTC)).isEqualTo(singleValue(typeWithDifferentScale, 1001L));
+
+        // Test that statistics overflowing the size of the type are not used
+        assertThat(getDomain(columnDescriptor, typeWithDifferentScale, 10, longColumnStats(100012L, 100012L), ID, UTC)).isEqualTo(singleValue(typeWithDifferentScale, 10001L));
+
+        assertThat(getDomain(columnDescriptor, typeWithDifferentScale, 10, longColumnStats(0L, 100L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(typeWithDifferentScale, 0L, true, 10L, true)), false));
+
+        // fail on higher precision values
+        assertThatExceptionOfType(ParquetCorruptionException.class)
+                .isThrownBy(() -> getDomain(columnDescriptor, createDecimalType(4, 2), 10, binaryColumnStats(maximum, maximum), ID, UTC))
+                .withMessage("Malformed Parquet file. Corrupted statistics for column \"[] required fixed_len_byte_array(0) ShortDecimalColumnWithDecimalAnnotation (DECIMAL(38,2))\": [min: 0x00000000000000000000000000003039, max: 0x00000000000000000000000000003039, num_nulls: 0] [testFile]");
+    }
+
     @Test
     public void testLongDecimal()
             throws Exception
@@ -319,6 +357,49 @@ public void testLongDecimalWithNoScale()
                 .withMessage("Malformed Parquet file. Corrupted statistics for column \"[] required fixed_len_byte_array(0) LongDecimalColumnWithNoScale\": [min: 0x00000000000000000000000000000064, max: 0x0000000000000000000000000000000A, num_nulls: 0] [testFile]");
     }
 
+    @Test
+    public void testLongDecimalWithShortDecimalAnnotation()
+            throws Exception
+    {
+        ColumnDescriptor columnDescriptor = createColumnDescriptor(INT32, decimalType(2, 5), "ShortDecimalColumn");
+        DecimalType type = createDecimalType(20, 2);
+
+        assertThat(getDomain(columnDescriptor, type, 0, null, ID, UTC)).isEqualTo(all(type));
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(10012L, 10012L), ID, UTC)).isEqualTo(singleValue(type, Int128.valueOf(10012L)));
+
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(0L, 10012L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, Int128.valueOf(0L), true, Int128.valueOf(10012L), true)), false));
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(0, 100L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, Int128.valueOf(0L), true, Int128.valueOf(100L), true)), false));
+
+        // fail on corrupted statistics
+        assertThatExceptionOfType(ParquetCorruptionException.class)
+                .isThrownBy(() -> getDomain(columnDescriptor, type, 10, longColumnStats(100L, 10L), ID, UTC))
+                .withMessage("Malformed Parquet file. Corrupted statistics for column \"[] required int32 ShortDecimalColumn (DECIMAL(5,2))\": [min: 100, max: 10, num_nulls: 0] [testFile]");
+    }
+
+    @Test
+    public void testLongDecimalWithInt64DecimalAnnotation()
+            throws Exception
+    {
+        ColumnDescriptor columnDescriptor = createColumnDescriptor(INT64, decimalType(2, 5), "ShortDecimalColumn");
+        DecimalType type = createDecimalType(20, 2);
+        BigInteger maximum = new BigInteger("12345");
+
+        Int128 zero = Int128.ZERO;
+        Int128 hundred = Int128.valueOf(100L);
+        Int128 max = Int128.valueOf(maximum);
+
+        assertThat(getDomain(columnDescriptor, type, 0, null, ID, UTC)).isEqualTo(all(type));
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(maximum.longValue(), maximum.longValue()), ID, UTC)).isEqualTo(singleValue(type, max));
+
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(0L, 100L), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, zero, true, hundred, true)), false));
+        assertThat(getDomain(columnDescriptor, type, 10, longColumnStats(0, 100), ID, UTC)).isEqualTo(create(ValueSet.ofRanges(range(type, zero, true, hundred, true)), false));
+
+        // fail on corrupted statistics
+        assertThatExceptionOfType(ParquetCorruptionException.class)
+                .isThrownBy(() -> getDomain(columnDescriptor, type, 10, longColumnStats(100L, 10L), ID, UTC))
+                .withMessage("Malformed Parquet file. Corrupted statistics for column \"[] required int64 ShortDecimalColumn (DECIMAL(5,2))\": [min: 100, max: 10, num_nulls: 0] [testFile]");
+    }
+
     @Test
     public void testDouble()
             throws Exception
@@ -744,6 +825,11 @@ private ColumnDescriptor createColumnDescriptor(PrimitiveTypeName typeName, Stri
         return new ColumnDescriptor(new String[] {}, new PrimitiveType(REQUIRED, typeName, columnName), 0, 0);
     }
 
+    private ColumnDescriptor createColumnDescriptor(PrimitiveTypeName typeName, LogicalTypeAnnotation typeAnnotation, String columnName)
+    {
+        return new ColumnDescriptor(new String[] {}, new PrimitiveType(REQUIRED, typeName, columnName).withLogicalTypeAnnotation(typeAnnotation), 0, 0);
+    }
+
     private TupleDomain<ColumnDescriptor> getEffectivePredicate(ColumnDescriptor column, VarcharType type, Slice value)
     {
         ColumnDescriptor predicateColumn = new ColumnDescriptor(column.getPath(), column.getPrimitiveType(), 0, 0);

Original file line number	Diff line number	Diff line change
`@@ -330,7 +330,7 @@ public Optional<Boolean> visit(UUIDLogicalTypeAnnotation uuidLogicalType)`
`330`	`330`	`.orElse(FALSE);`
`331`	`331`	`}`
`332`	`332`
`333`		`- private static boolean isDecimalRescaled(DecimalLogicalTypeAnnotation decimalAnnotation, DecimalType trinoType)`
	`333`	`+ public static boolean isDecimalRescaled(DecimalLogicalTypeAnnotation decimalAnnotation, DecimalType trinoType)`
`334`	`334`	`{`
`335`	`335`	`return decimalAnnotation.getPrecision() != trinoType.getPrecision()`
`336`	`336`	`\|\| decimalAnnotation.getScale() != trinoType.getScale();`