This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b59f966438 [format] Fix reads and predicate pushdown when Parquet
files differ from the declared schema (#8995)
b59f966438 is described below
commit b59f9664387b2c636bcfca0eadcfa16631dbd8a0
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Mon Aug 3 18:03:05 2026 +0800
[format] Fix reads and predicate pushdown when Parquet files differ from
the declared schema (#8995)
---
.../paimon/table/format/FormatReadBuilder.java | 9 +-
.../format/parquet/ParquetSchemaConverter.java | 11 +
.../reader/ParquetVectorUpdaterFactory.java | 132 +++++-
.../parquet/filter2/predicate/ParquetFilters.java | 272 ++++++++++--
.../format/parquet/ParquetTypeWideningTest.java | 479 +++++++++++++++++++++
5 files changed, 859 insertions(+), 44 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 30a567b436..d05f0ebc12 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -238,7 +238,14 @@ public class FormatReadBuilder implements ReadBuilder {
formatReaderContext.filePath());
} catch (Exception e) {
FileUtils.checkExists(formatReaderContext.fileIO(),
formatReaderContext.filePath());
- throw e;
+ // A split spans many files that a Format Table's writers may have
written differently.
+ // Naming the one that failed is the only way to tell them apart
from the outside.
+ throw new IOException(
+ "Failed to read file "
+ + formatReaderContext.filePath()
+ + " of table "
+ + table.fullName(),
+ e);
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
index 3ce514cb75..12da716ef4 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
@@ -59,6 +59,17 @@ public class ParquetSchemaConverter {
public static final String MAP_VALUE_NAME = "value";
public static final String LIST_ELEMENT_NAME = "element";
+ /**
+ * Whether {@code type} is an unsigned integer. Such a column stores a
signed value whose bits
+ * have to be reinterpreted on read, and its statistics are ordered
unsigned, so both the reader
+ * and the predicate pushdown have to treat it apart from an ordinary int.
+ */
+ public static boolean isUnsignedInt(PrimitiveType type) {
+ LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
+ return logicalType instanceof
LogicalTypeAnnotation.IntLogicalTypeAnnotation
+ && !((LogicalTypeAnnotation.IntLogicalTypeAnnotation)
logicalType).isSigned();
+ }
+
/** Convert paimon {@link RowType} to parquet {@link MessageType}. */
public static MessageType convertToParquetMessageType(RowType rowType) {
return new MessageType(PAIMON_SCHEMA, convertToParquetTypes(rowType));
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
index 3dcd3fbd03..885d880514 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
@@ -73,6 +73,7 @@ import java.nio.ByteOrder;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import static
org.apache.paimon.format.parquet.ParquetSchemaConverter.isUnsignedInt;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Updater Factory to get {@link ParquetVectorUpdater}. */
@@ -180,7 +181,18 @@ public class ParquetVectorUpdaterFactory {
@Override
public UpdaterFactory visit(BigIntType bigIntType) {
- return c -> new LongUpdater();
+ return c -> {
+ if (c.getPrimitiveType().getPrimitiveTypeName()
+ == PrimitiveType.PrimitiveTypeName.INT32) {
+ // The file kept the narrower int, either because the
column was widened in
+ // the metastore after the data was written, or because it
is unsigned and
+ // BIGINT is the only Paimon type that can hold every
value.
+ return isUnsignedInt(c.getPrimitiveType())
+ ? new LongFromUnsignedIntegerUpdater()
+ : new LongFromIntegerUpdater();
+ }
+ return new LongUpdater();
+ };
}
@Override
@@ -196,7 +208,13 @@ public class ParquetVectorUpdaterFactory {
@Override
public UpdaterFactory visit(DoubleType doubleType) {
- return c -> new DoubleUpdater();
+ return c -> {
+ if (c.getPrimitiveType().getPrimitiveTypeName()
+ == PrimitiveType.PrimitiveTypeName.FLOAT) {
+ return new DoubleFromFloatUpdater();
+ }
+ return new DoubleUpdater();
+ };
}
@Override
@@ -563,6 +581,81 @@ public class ParquetVectorUpdaterFactory {
}
}
+ /** Reads a signed INT32 column into a BIGINT vector. */
+ private static class LongFromIntegerUpdater
+ implements ParquetVectorUpdater<WritableLongVector> {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableLongVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ values.setLong(offset + i, valuesReader.readInteger());
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader)
{
+ valuesReader.skipIntegers(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset, WritableLongVector values, VectorizedValuesReader
valuesReader) {
+ values.setLong(offset, valuesReader.readInteger());
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableLongVector values,
+ WritableIntVector dictionaryIds,
+ Dictionary dictionary) {
+ values.setLong(offset,
dictionary.decodeToInt(dictionaryIds.getInt(offset)));
+ }
+ }
+
+ /**
+ * Reads an unsigned INT32 column into a BIGINT vector. The stored bits
are a signed int, so
+ * every value above {@link Integer#MAX_VALUE} arrives negative and has to
be reinterpreted.
+ */
+ private static class LongFromUnsignedIntegerUpdater
+ implements ParquetVectorUpdater<WritableLongVector> {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableLongVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ values.setLong(offset + i,
Integer.toUnsignedLong(valuesReader.readInteger()));
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader)
{
+ valuesReader.skipIntegers(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset, WritableLongVector values, VectorizedValuesReader
valuesReader) {
+ values.setLong(offset,
Integer.toUnsignedLong(valuesReader.readInteger()));
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableLongVector values,
+ WritableIntVector dictionaryIds,
+ Dictionary dictionary) {
+ values.setLong(
+ offset,
+
Integer.toUnsignedLong(dictionary.decodeToInt(dictionaryIds.getInt(offset))));
+ }
+ }
+
private abstract static class AbstractTimestampUpdater
implements ParquetVectorUpdater<WritableColumnVector> {
@@ -876,6 +969,41 @@ public class ParquetVectorUpdaterFactory {
}
}
+ /** Reads a FLOAT column into a DOUBLE vector. */
+ private static class DoubleFromFloatUpdater
+ implements ParquetVectorUpdater<WritableDoubleVector> {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableDoubleVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ values.setDouble(offset + i, valuesReader.readFloat());
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader)
{
+ valuesReader.skipFloats(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset, WritableDoubleVector values,
VectorizedValuesReader valuesReader) {
+ values.setDouble(offset, valuesReader.readFloat());
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableDoubleVector values,
+ WritableIntVector dictionaryIds,
+ Dictionary dictionary) {
+ values.setDouble(offset,
dictionary.decodeToFloat(dictionaryIds.getInt(offset)));
+ }
+ }
+
private static class BinaryUpdater implements
ParquetVectorUpdater<WritableBytesVector> {
@Override
public void readValues(
diff --git
a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
index 27d414ad75..f7f5edbeeb 100644
---
a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
+++
b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java
@@ -21,6 +21,7 @@ package org.apache.parquet.filter2.predicate;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.Decimal;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.format.parquet.ParquetSchemaConverter;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FunctionVisitor;
import org.apache.paimon.predicate.LeafPredicate;
@@ -61,6 +62,8 @@ import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
+import javax.annotation.Nullable;
+
import java.io.Serializable;
import java.math.BigDecimal;
import java.math.RoundingMode;
@@ -306,20 +309,7 @@ public class ParquetFilters {
}
}
- if (value instanceof Number) {
- if (value instanceof Byte) {
- return ((Byte) value).intValue();
- } else if (value instanceof Short) {
- return ((Short) value).intValue();
- }
- return (Comparable<?>) value;
- } else if (value instanceof String) {
- return Binary.fromString((String) value);
- } else if (value instanceof BinaryString) {
- return Binary.fromString(value.toString());
- } else if (value instanceof byte[]) {
- return Binary.fromReusedByteArray((byte[]) value);
- } else if (value instanceof Timestamp) {
+ if (value instanceof Timestamp) {
Timestamp timestamp = (Timestamp) value;
timestampPrimitiveType(fieldRef, fileSchema, caseSensitive);
int precision = getTimestampPrecision(type);
@@ -334,6 +324,72 @@ public class ParquetFilters {
throw new UnsupportedOperationException();
}
+ // The literal has to speak whatever the file holds, not what the
table declares.
+ switch (pushdownTarget(fieldRef, fileSchema, caseSensitive).type) {
+ case INT32:
+ return toInt(value);
+ case INT64:
+ return toLong(value);
+ case FLOAT:
+ return toFloat(value);
+ case DOUBLE:
+ return toDouble(value);
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return toBinary(value);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private Comparable<?> toInt(Object value) {
+ long asLong = toLongValue(value);
+ if (asLong < Integer.MIN_VALUE || asLong > Integer.MAX_VALUE) {
+ // Truncating would change what the predicate means: `<
3000000000` would become
+ // `< -1294967296` and prune row groups that hold matching
rows.
+ throw new UnsupportedOperationException();
+ }
+ return (int) asLong;
+ }
+
+ private Comparable<?> toLong(Object value) {
+ return toLongValue(value);
+ }
+
+ private long toLongValue(Object value) {
+ if (value instanceof Byte
+ || value instanceof Short
+ || value instanceof Integer
+ || value instanceof Long) {
+ return ((Number) value).longValue();
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ private Comparable<?> toFloat(Object value) {
+ if (value instanceof Float) {
+ return (Float) value;
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ private Comparable<?> toDouble(Object value) {
+ // Float widens to double exactly; a double literal must never be
narrowed to float,
+ // which would round the bound and drop rows that sit next to it.
+ if (value instanceof Float || value instanceof Double) {
+ return ((Number) value).doubleValue();
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ private Comparable<?> toBinary(Object value) {
+ if (value instanceof String) {
+ return Binary.fromString((String) value);
+ } else if (value instanceof BinaryString) {
+ return Binary.fromString(value.toString());
+ } else if (value instanceof byte[]) {
+ return Binary.fromReusedByteArray((byte[]) value);
+ }
throw new UnsupportedOperationException();
}
@@ -430,23 +486,137 @@ public class ParquetFilters {
private static PrimitiveType primitiveType(
FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) {
- Type matched = null;
+ PrimitiveType matched = findPrimitiveType(fieldRef, fileSchema,
caseSensitive);
+ if (matched == null) {
+ throw new UnsupportedOperationException();
+ }
+ return matched;
+ }
+
+ /**
+ * The file's column for {@code fieldRef}, or null when the file has no
such column. A column
+ * that exists but is not primitive cannot carry a predicate at all, so it
is rejected outright.
+ */
+ @Nullable
+ private static PrimitiveType findPrimitiveType(
+ FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) {
// Paimon predicates currently reference top-level fields only. Nested
field
// predicates are rejected before reaching the format reader.
for (Type field : fileSchema.getFields()) {
if (caseSensitive
? field.getName().equals(fieldRef.name())
: field.getName().equalsIgnoreCase(fieldRef.name())) {
- matched = field;
- break;
+ if (!field.isPrimitive()) {
+ throw new UnsupportedOperationException();
+ }
+ return field.asPrimitiveType();
}
}
+ return null;
+ }
- if (matched == null || !matched.isPrimitive()) {
+ /**
+ * The physical type a pushed-down predicate on {@code fieldRef} has to be
expressed in. A
+ * Format Table takes its schema from the metastore while its files are
written by someone else,
+ * so the declared type is not necessarily what the file holds and the
predicate has to follow
+ * the file.
+ *
+ * <p>Falls back to what the table declares when the file has no such
column: parquet-mr skips
+ * validation for a column it cannot find and evaluates it as null, so the
predicate still
+ * prunes.
+ *
+ * <p>Throws {@link UnsupportedOperationException} when the two types
cannot be reconciled
+ * without changing what the predicate means. {@link #convert} swallows
that and drops the
+ * predicate, which costs pruning but never rows.
+ */
+ private static PushdownTarget pushdownTarget(
+ FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) {
+ PrimitiveType.PrimitiveTypeName[] acceptable =
acceptableTypes(fieldRef.type());
+ PrimitiveType fileType = findPrimitiveType(fieldRef, fileSchema,
caseSensitive);
+ if (fileType == null) {
+ return new PushdownTarget(fieldRef.name(), acceptable[0]);
+ }
+
+ if (ParquetSchemaConverter.isUnsignedInt(fileType)) {
+ // An unsigned column orders its statistics unsigned, so a signed
bound would prune the
+ // wrong row groups. The read still widens the column; only the
pruning is given up.
throw new UnsupportedOperationException();
}
- return matched.asPrimitiveType();
+ for (PrimitiveType.PrimitiveTypeName candidate : acceptable) {
+ if (fileType.getPrimitiveTypeName() == candidate) {
+ return new PushdownTarget(fileType.getName(), candidate);
+ }
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * The file column a predicate has to name and the type it has to speak.
The name matters in
+ * case-insensitive mode: parquet-mr resolves a predicate against the file
by exact column path,
+ * so a predicate carrying the metastore's spelling of a column that the
file spells differently
+ * reads as a column the file does not have - which the statistics filter
takes for all-null and
+ * prunes away, silently losing every row.
+ */
+ private static class PushdownTarget {
+
+ private final String name;
+ private final PrimitiveType.PrimitiveTypeName type;
+
+ private PushdownTarget(String name, PrimitiveType.PrimitiveTypeName
type) {
+ this.name = name;
+ this.type = type;
+ }
+ }
+
+ /**
+ * The physical types a predicate on this Paimon type can be expressed in,
most preferred first.
+ * The head is what Paimon itself writes; the tail is the type widening
the vectorized reader
+ * accepts, so pushdown stays available for exactly the files that can be
read.
+ */
+ private static PrimitiveType.PrimitiveTypeName[] acceptableTypes(
+ org.apache.paimon.types.DataType type) {
+ switch (type.getTypeRoot()) {
+ case BOOLEAN:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.BOOLEAN
+ };
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.INT32,
PrimitiveType.PrimitiveTypeName.INT64
+ };
+ case BIGINT:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.INT64,
PrimitiveType.PrimitiveTypeName.INT32
+ };
+ case FLOAT:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.FLOAT,
PrimitiveType.PrimitiveTypeName.DOUBLE
+ };
+ case DOUBLE:
+ // A double bound cannot be narrowed to float without rounding
it, so a FLOAT file
+ // column is read but never filtered on.
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.DOUBLE
+ };
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.INT32
+ };
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ return new PrimitiveType.PrimitiveTypeName[] {
+ PrimitiveType.PrimitiveTypeName.BINARY,
+ PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
+ };
+ default:
+ throw new UnsupportedOperationException();
+ }
}
private static int getTimestampPrecision(org.apache.paimon.types.DataType
type) {
@@ -462,81 +632,101 @@ public class ParquetFilters {
implements DataTypeVisitor<Operators.Column<?>> {
private final FieldRef fieldRef;
- private final String name;
private final MessageType fileSchema;
private final boolean caseSensitive;
public ConvertToColumnTypeVisitor(
FieldRef fieldRef, MessageType fileSchema, boolean
caseSensitive) {
this.fieldRef = fieldRef;
- this.name = fieldRef.name();
this.fileSchema = fileSchema;
this.caseSensitive = caseSensitive;
}
+ /** The column typed after the file, not after what the table
declares. */
+ private Operators.Column<?> column() {
+ PushdownTarget target = pushdownTarget(fieldRef, fileSchema,
caseSensitive);
+ switch (target.type) {
+ case BOOLEAN:
+ return FilterApi.booleanColumn(target.name);
+ case INT32:
+ return FilterApi.intColumn(target.name);
+ case INT64:
+ return FilterApi.longColumn(target.name);
+ case FLOAT:
+ return FilterApi.floatColumn(target.name);
+ case DOUBLE:
+ return FilterApi.doubleColumn(target.name);
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return FilterApi.binaryColumn(target.name);
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
@Override
public Operators.Column<?> visit(CharType charType) {
- return FilterApi.binaryColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(VarCharType varCharType) {
- return FilterApi.binaryColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(BooleanType booleanType) {
- return FilterApi.booleanColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(BinaryType binaryType) {
- return FilterApi.binaryColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(VarBinaryType varBinaryType) {
- return FilterApi.binaryColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(TinyIntType tinyIntType) {
- return FilterApi.intColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(SmallIntType smallIntType) {
- return FilterApi.intColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(IntType intType) {
- return FilterApi.intColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(BigIntType bigIntType) {
- return FilterApi.longColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(FloatType floatType) {
- return FilterApi.floatColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(DoubleType doubleType) {
- return FilterApi.doubleColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(DateType dateType) {
- return FilterApi.intColumn(name);
+ return column();
}
@Override
public Operators.Column<?> visit(TimeType timeType) {
- return FilterApi.intColumn(name);
+ return column();
}
@Override
@@ -544,12 +734,12 @@ public class ParquetFilters {
PrimitiveType primitiveType = decimalPrimitiveType(fieldRef,
fileSchema, caseSensitive);
switch (primitiveType.getPrimitiveTypeName()) {
case INT32:
- return FilterApi.intColumn(fieldRef.name());
+ return FilterApi.intColumn(primitiveType.getName());
case INT64:
- return FilterApi.longColumn(fieldRef.name());
+ return FilterApi.longColumn(primitiveType.getName());
case BINARY:
case FIXED_LEN_BYTE_ARRAY:
- return FilterApi.binaryColumn(fieldRef.name());
+ return FilterApi.binaryColumn(primitiveType.getName());
default:
throw new UnsupportedOperationException();
}
@@ -559,8 +749,8 @@ public class ParquetFilters {
public Operators.Column<?> visit(TimestampType timestampType) {
int precision = timestampType.getPrecision();
if (precision <= 6) {
- timestampPrimitiveType(fieldRef, fileSchema, caseSensitive);
- return FilterApi.longColumn(name);
+ return FilterApi.longColumn(
+ timestampPrimitiveType(fieldRef, fileSchema,
caseSensitive).getName());
}
// precision > 6 uses INT96, not supported for filter pushdown
throw new UnsupportedOperationException();
@@ -570,8 +760,8 @@ public class ParquetFilters {
public Operators.Column<?> visit(LocalZonedTimestampType
localZonedTimestampType) {
int precision = localZonedTimestampType.getPrecision();
if (precision <= 6) {
- timestampPrimitiveType(fieldRef, fileSchema, caseSensitive);
- return FilterApi.longColumn(name);
+ return FilterApi.longColumn(
+ timestampPrimitiveType(fieldRef, fileSchema,
caseSensitive).getName());
}
// precision > 6 uses INT96, not supported for filter pushdown
throw new UnsupportedOperationException();
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTypeWideningTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTypeWideningTest.java
new file mode 100644
index 0000000000..d4d71f9014
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetTypeWideningTest.java
@@ -0,0 +1,479 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.format.parquet;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.util.HadoopOutputFile;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads where the declared type is wider than the type stored in the Parquet
file.
+ *
+ * <p>A Format Table takes its schema from the metastore while its files are
written by someone
+ * else, so the two can disagree: an {@code ALTER TABLE ... CHANGE c int
BIGINT} leaves every
+ * existing file with an INT32 column, and an unsigned INT32 column is
imported as BIGINT because
+ * that is the only Paimon type that can hold it. Both the vectorized read and
the predicate
+ * pushdown have to cope.
+ *
+ * <p>Column names mirror midas_prod.tbl_imp, where this was found.
+ */
+class ParquetTypeWideningTest {
+
+ @TempDir File folder;
+
+ private static final RowType ECPM_BIGINT =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("ecpm", DataTypes.BIGINT())
+ .field("revenue", DataTypes.BIGINT())
+ .build();
+
+ private static final long[] ECPM_VALUES = {10L, 150L, 3000L};
+
+ private static final String[] PAGEVIEW_IDS = {"a", "b", "c"};
+
+ // ------------------------------------------------------------------
+ // Widening INT32 -> BIGINT, the shape reported from production.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testSignedInt32ReadAsBigInt() throws Exception {
+ Path path = write(ecpmSchema("int32 ecpm"));
+
+ assertThat(longs(read(ECPM_BIGINT, path, null),
1)).containsExactly(10L, 150L, 3000L);
+ }
+
+ @Test
+ void testSignedInt32ReadAsBigIntWithPushdown() throws Exception {
+ Path path = write(ecpmSchema("int32 ecpm"));
+ PredicateBuilder builder = new PredicateBuilder(ECPM_BIGINT);
+
+ List<Object[]> rows =
+ read(
+ ECPM_BIGINT,
+ path,
+ Arrays.asList(builder.lessThan(1, 5000L),
builder.greaterThan(2, 0L)));
+
+ assertThat(longs(rows, 1)).containsExactly(10L, 150L, 3000L);
+ }
+
+ /** A predicate that excludes the whole row group must still prune it
after the type fix. */
+ @Test
+ void testSignedInt32PushdownStillPrunes() throws Exception {
+ Path path = write(ecpmSchema("int32 ecpm"));
+ PredicateBuilder builder = new PredicateBuilder(ECPM_BIGINT);
+
+ List<Object[]> rows =
+ read(ECPM_BIGINT, path,
Collections.singletonList(builder.greaterThan(1, 99999L)));
+
+ assertThat(rows).isEmpty();
+ }
+
+ /**
+ * The literal does not fit in INT32. Narrowing it would turn a bound of
3000000000 into one of
+ * -1294967296, which prunes the row group and silently drops every row.
The predicate has to be
+ * abandoned instead.
+ */
+ @Test
+ void testLiteralOutsideInt32RangeIsNotTruncated() throws Exception {
+ Path path = write(ecpmSchema("int32 ecpm"));
+ PredicateBuilder builder = new PredicateBuilder(ECPM_BIGINT);
+
+ List<Object[]> rows =
+ read(
+ ECPM_BIGINT,
+ path,
+ Collections.singletonList(builder.lessThan(1,
3_000_000_000L)));
+
+ assertThat(longs(rows, 1)).containsExactly(10L, 150L, 3000L);
+ }
+
+ // ------------------------------------------------------------------
+ // Unsigned INT32 -> BIGINT. Spark and Trino both surface this as bigint,
+ // so an import writes BIGINT into the metastore while the file stays
INT32.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testUnsignedInt32ReadAsBigInt() throws Exception {
+ Path path =
+ write(
+ ecpmSchema("int32 ecpm (INTEGER(32,false))"),
+ (group, i) -> group.append("ecpm", UNSIGNED_RAW[i]));
+
+ // The raw ints are negative; read as BIGINT they must come back as
their unsigned value.
+ assertThat(longs(read(ECPM_BIGINT, path, null), 1))
+ .containsExactly(10L, 3_000_000_000L, 4_294_967_295L);
+ }
+
+ /**
+ * Statistics on an unsigned column are ordered unsigned, so a signed
predicate would prune the
+ * wrong row groups. The predicate must be abandoned rather than pushed.
+ */
+ @Test
+ void testUnsignedInt32PushdownDoesNotDropRows() throws Exception {
+ Path path =
+ write(
+ ecpmSchema("int32 ecpm (INTEGER(32,false))"),
+ (group, i) -> group.append("ecpm", UNSIGNED_RAW[i]));
+ PredicateBuilder builder = new PredicateBuilder(ECPM_BIGINT);
+
+ List<Object[]> rows =
+ read(ECPM_BIGINT, path,
Collections.singletonList(builder.greaterThan(1, 5L)));
+
+ assertThat(longs(rows, 1)).containsExactly(10L, 3_000_000_000L,
4_294_967_295L);
+ }
+
+ // ------------------------------------------------------------------
+ // FLOAT -> DOUBLE, the same hole in the other numeric family.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testFloatReadAsDouble() throws Exception {
+ RowType readType =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("rate", DataTypes.DOUBLE())
+ .build();
+ Path path =
+ writeSchema(
+ "message root {\n"
+ + " optional binary pageviewId (UTF8);\n"
+ + " optional float rate;\n"
+ + "}",
+ (group, i) ->
+ group.append("pageviewId", PAGEVIEW_IDS[i])
+ .append("rate", (float) (i + 1)));
+
+ List<Object[]> rows = read(readType, path, null);
+
+ assertThat(rows).hasSize(3);
+ assertThat((Double) rows.get(0)[1]).isEqualTo(1.0d);
+ assertThat((Double) rows.get(2)[1]).isEqualTo(3.0d);
+ }
+
+ @Test
+ void testFloatReadAsDoubleWithPushdown() throws Exception {
+ RowType readType =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("rate", DataTypes.DOUBLE())
+ .build();
+ Path path =
+ writeSchema(
+ "message root {\n"
+ + " optional binary pageviewId (UTF8);\n"
+ + " optional float rate;\n"
+ + "}",
+ (group, i) ->
+ group.append("pageviewId", PAGEVIEW_IDS[i])
+ .append("rate", (float) (i + 1)));
+ PredicateBuilder builder = new PredicateBuilder(readType);
+
+ List<Object[]> rows =
+ read(readType, path,
Collections.singletonList(builder.greaterThan(1, 0.5d)));
+
+ assertThat(rows).hasSize(3);
+ }
+
+ // ------------------------------------------------------------------
+ // Narrowing INT64 -> INT. The reader already handles it via
+ // IntegerFromLongUpdater; only the pushdown was left behind.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testInt64ReadAsIntWithPushdown() throws Exception {
+ RowType readType =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("ecpm", DataTypes.INT())
+ .build();
+ Path path = write(ecpmSchema("int64 ecpm"));
+ PredicateBuilder builder = new PredicateBuilder(readType);
+
+ List<Object[]> rows =
+ read(readType, path,
Collections.singletonList(builder.lessThan(1, 5000)));
+
+ assertThat(rows).hasSize(3);
+ assertThat((Integer) rows.get(2)[1]).isEqualTo(3000);
+ }
+
+ // ------------------------------------------------------------------
+ // Control and mixed-file cases.
+ // ------------------------------------------------------------------
+
+ /** A file that really holds int64 was never broken; it must stay that
way. */
+ @Test
+ void testInt64ReadAsBigIntIsUnchanged() throws Exception {
+ Path path = write(ecpmSchema("int64 ecpm"));
+ PredicateBuilder builder = new PredicateBuilder(ECPM_BIGINT);
+
+ List<Object[]> rows =
+ read(ECPM_BIGINT, path,
Collections.singletonList(builder.lessThan(1, 5000L)));
+
+ assertThat(longs(rows, 1)).containsExactly(10L, 150L, 3000L);
+ }
+
+ /**
+ * One reader factory serves every file of a split, so a partition that
mixes writers puts both
+ * physical types through the same factory. Neither file may break the
other.
+ */
+ @Test
+ void testMixedFilesInOneFactory() throws Exception {
+ Path int32File = write(ecpmSchema("int32 ecpm"));
+ Path int64File = write(ecpmSchema("int64 ecpm"));
+
+ ParquetReaderFactory factory =
+ new ParquetReaderFactory(
+ new Options(),
+ ECPM_BIGINT,
+ 1024,
+ Collections.singletonList(
+ new PredicateBuilder(ECPM_BIGINT).lessThan(1,
5000L)));
+
+ assertThat(longs(read(factory, ECPM_BIGINT, int64File), 1))
+ .containsExactly(10L, 150L, 3000L);
+ assertThat(longs(read(factory, ECPM_BIGINT, int32File), 1))
+ .containsExactly(10L, 150L, 3000L);
+ assertThat(longs(read(factory, ECPM_BIGINT, int64File), 1))
+ .containsExactly(10L, 150L, 3000L);
+ }
+
+ // ------------------------------------------------------------------
+ // Case-insensitive resolution. The metastore lowercases column names
while a
+ // Spark-written file keeps the original spelling, so a predicate carrying
the
+ // metastore's spelling names a column the file does not have - which
parquet-mr
+ // takes for all-null and prunes away, losing every row without a word.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testCaseInsensitivePushdownKeepsRows() throws Exception {
+ Path path =
+ writeSchema(
+ "message root {\n"
+ + " optional binary PageviewId (UTF8);\n"
+ + " optional int32 Ecpm;\n"
+ + "}",
+ (group, i) ->
+ group.append("PageviewId", "id" +
i).append("Ecpm", (i + 1) * 100));
+ RowType readType =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("ecpm", DataTypes.BIGINT())
+ .build();
+ PredicateBuilder builder = new PredicateBuilder(readType);
+
+ List<Object[]> rows =
+ read(
+ readType,
+ path,
+ Collections.singletonList(builder.lessThan(1,
100_000L)),
+ false);
+
+ assertThat(longs(rows, 1)).containsExactly(100L, 200L, 300L);
+ }
+
+ /** Same, with no type widening in play at all, so only the spelling is at
stake. */
+ @Test
+ void testCaseInsensitivePushdownKeepsRowsWithoutWidening() throws
Exception {
+ Path path =
+ writeSchema(
+ "message root {\n"
+ + " optional binary PageviewId (UTF8);\n"
+ + " optional int64 Ecpm;\n"
+ + "}",
+ (group, i) ->
+ group.append("PageviewId", "id" + i)
+ .append("Ecpm", (long) ((i + 1) *
100)));
+ RowType readType =
+ RowType.builder()
+ .field("pageviewId", DataTypes.STRING())
+ .field("ecpm", DataTypes.BIGINT())
+ .build();
+ PredicateBuilder builder = new PredicateBuilder(readType);
+
+ List<Object[]> rows =
+ read(
+ readType,
+ path,
+ Collections.singletonList(builder.lessThan(1,
100_000L)),
+ false);
+
+ assertThat(longs(rows, 1)).containsExactly(100L, 200L, 300L);
+ }
+
+ // ------------------------------------------------------------------
+ // Helpers
+ // ------------------------------------------------------------------
+
+ /** Raw int32 bit patterns for 10, 3000000000 and 4294967295 read as
unsigned. */
+ private static final int[] UNSIGNED_RAW = {10, (int) 3_000_000_000L, -1};
+
+ private static List<Long> longs(List<Object[]> rows, int field) {
+ List<Long> values = new ArrayList<>();
+ rows.forEach(row -> values.add((Long) row[field]));
+ return values;
+ }
+
+ private static String ecpmSchema(String ecpmType) {
+ return "message root {\n"
+ + " optional binary pageviewId (UTF8);\n"
+ + " optional "
+ + ecpmType
+ + ";\n"
+ + " optional int64 revenue;\n"
+ + "}";
+ }
+
+ private List<Object[]> read(RowType readType, Path path, List<Predicate>
filters)
+ throws Exception {
+ return read(readType, path, filters, true);
+ }
+
+ private List<Object[]> read(
+ RowType readType, Path path, List<Predicate> filters, boolean
caseSensitive)
+ throws Exception {
+ Options options = new Options();
+ options.set(CatalogOptions.CASE_SENSITIVE, caseSensitive);
+ return read(new ParquetReaderFactory(options, readType, 1024,
filters), readType, path);
+ }
+
+ private List<Object[]> read(ParquetReaderFactory factory, RowType
readType, Path path)
+ throws Exception {
+ LocalFileIO fileIO = new LocalFileIO();
+ List<Object[]> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ factory.createReader(
+ new FormatReaderContext(fileIO, path,
fileIO.getFileSize(path)))) {
+ // Row instances are reused across iterations, so materialize
every value.
+ reader.forEachRemaining(row -> rows.add(materialize(row,
readType)));
+ }
+ return rows;
+ }
+
+ private static Object[] materialize(InternalRow row, RowType readType) {
+ Object[] values = new Object[readType.getFieldCount()];
+ for (int i = 0; i < values.length; i++) {
+ if (row.isNullAt(i)) {
+ continue;
+ }
+ switch (readType.getTypeAt(i).getTypeRoot()) {
+ case VARCHAR:
+ values[i] = row.getString(i).toString();
+ break;
+ case INTEGER:
+ values[i] = row.getInt(i);
+ break;
+ case BIGINT:
+ values[i] = row.getLong(i);
+ break;
+ case DOUBLE:
+ values[i] = row.getDouble(i);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unhandled type in test: " +
readType.getTypeAt(i));
+ }
+ }
+ return values;
+ }
+
+ private Path write(String schemaText) throws Exception {
+ return write(schemaText, null);
+ }
+
+ private Path write(String schemaText, BiConsumer<Group, Integer>
ecpmAppender)
+ throws Exception {
+ MessageType schema = MessageTypeParser.parseMessageType(schemaText);
+ boolean ecpmIsLong =
+ schema.getType("ecpm")
+ .asPrimitiveType()
+ .getPrimitiveTypeName()
+ .name()
+ .equals("INT64");
+ return writeGroups(
+ schema,
+ (group, i) -> {
+ group.append("pageviewId", PAGEVIEW_IDS[i]);
+ if (ecpmAppender != null) {
+ ecpmAppender.accept(group, i);
+ } else if (ecpmIsLong) {
+ group.append("ecpm", ECPM_VALUES[i]);
+ } else {
+ group.append("ecpm", (int) ECPM_VALUES[i]);
+ }
+ group.append("revenue", (long) (i + 1));
+ });
+ }
+
+ private Path writeSchema(String schemaText, BiConsumer<Group, Integer>
appender)
+ throws Exception {
+ return writeGroups(MessageTypeParser.parseMessageType(schemaText),
appender);
+ }
+
+ private Path writeGroups(MessageType schema, BiConsumer<Group, Integer>
appender)
+ throws Exception {
+ Path path = new Path(folder.getPath(), UUID.randomUUID().toString());
+ Configuration conf = new Configuration();
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(
+ HadoopOutputFile.fromPath(
+ new
org.apache.hadoop.fs.Path(path.toString()), conf))
+ .withType(schema)
+ .withConf(conf)
+
.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_1_0)
+ .build()) {
+ for (int i = 0; i < 3; i++) {
+ Group group = new SimpleGroupFactory(schema).newGroup();
+ appender.accept(group, i);
+ writer.write(group);
+ }
+ }
+ return path;
+ }
+}
