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 8096a82a5a [parquet] Fix that parquet filter pushdown wrong for
decimal (#8623)
8096a82a5a is described below
commit 8096a82a5a2e41319c4dc701e59f721255f3822b
Author: yuzelin <[email protected]>
AuthorDate: Sat Jul 18 13:34:24 2026 +0800
[parquet] Fix that parquet filter pushdown wrong for decimal (#8623)
---
.../paimon/format/parquet/ParquetFileFormat.java | 4 +-
.../format/parquet/ParquetReaderFactory.java | 48 +-
.../parquet/filter2/predicate/ParquetFilters.java | 304 ++++++---
.../apache/parquet/hadoop/ParquetFileReader.java | 2 +-
.../parquet/ParquetCaseInsensitiveReadTest.java | 4 +-
.../format/parquet/ParquetColumnVectorTest.java | 3 +-
.../paimon/format/parquet/ParquetFiltersTest.java | 718 ++++++++++++++++-----
.../format/parquet/ParquetReadWriteTest.java | 35 +-
.../format/parquet/ParquetSchemaCacheTest.java | 23 +-
.../reader/ParquetInt96RowSelectionTest.java | 3 +-
10 files changed, 849 insertions(+), 295 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
index 8093e714b0..d78441a840 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
@@ -38,7 +38,6 @@ import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.types.RowType;
import org.apache.parquet.ParquetReadOptions;
-import org.apache.parquet.filter2.predicate.ParquetFilters;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetOutputFormat;
@@ -85,8 +84,7 @@ public class ParquetFileFormat extends FileFormat implements
SupportsFieldMetada
RowType dataSchemaRowType,
RowType projectedRowType,
@Nullable List<Predicate> filters) {
- return new ParquetReaderFactory(
- options, projectedRowType, readBatchSize,
ParquetFilters.convert(filters));
+ return new ParquetReaderFactory(options, projectedRowType,
readBatchSize, filters);
}
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
index a94e068efd..dd425a23a7 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java
@@ -33,6 +33,7 @@ import
org.apache.paimon.format.shredding.ShreddingReadPlanFactories;
import org.apache.paimon.format.shredding.ShreddingReadPlanFactory;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataField;
@@ -45,7 +46,9 @@ import org.apache.paimon.utils.Preconditions;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.filter2.compat.FilterCompat;
+import org.apache.parquet.filter2.predicate.ParquetFilters;
import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.io.ColumnIOFactory;
import org.apache.parquet.io.MessageColumnIO;
import org.apache.parquet.schema.ConversionPatterns;
@@ -85,7 +88,7 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
private final RowType readType;
private final int batchSize;
private final boolean caseSensitive;
- @Nullable private final FilterCompat.Filter filter;
+ @Nullable private final List<Predicate> predicates;
/**
* Cache: fileSchema -> requestedSchema.
@@ -101,12 +104,12 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
new ConcurrentHashMap<>();
public ParquetReaderFactory(
- Options conf, RowType readType, int batchSize, @Nullable
FilterCompat.Filter filter) {
+ Options conf, RowType readType, int batchSize, @Nullable
List<Predicate> predicates) {
this.conf = conf;
this.readType = readType;
this.batchSize = batchSize;
this.caseSensitive =
conf.getOptional(CatalogOptions.CASE_SENSITIVE).orElse(true);
- this.filter = filter;
+ this.predicates = predicates;
}
@VisibleForTesting
@@ -117,18 +120,33 @@ public class ParquetReaderFactory implements
FormatReaderFactory {
@Override
public FileRecordReader<InternalRow>
createReader(FormatReaderFactory.Context context)
throws IOException {
- ParquetReadOptions.Builder builder =
- ParquetUtil.getParquetReadOptionsBuilder(conf)
- .withRecordFilter(filter)
- .withRange(0, context.fileSize());
-
- ParquetFileReader reader =
- new ParquetFileReader(
- ParquetInputFile.fromPath(
- context.fileIO(), context.filePath(),
context.fileSize()),
- builder.build(),
- context.selection());
- MessageType fileSchema = reader.getFileMetaData().getSchema();
+ ParquetInputFile inputFile =
+ ParquetInputFile.fromPath(context.fileIO(),
context.filePath(), context.fileSize());
+ ParquetReadOptions.Builder readOptionsBuilder =
+ ParquetUtil.getParquetReadOptionsBuilder(conf).withRange(0,
context.fileSize());
+ ParquetInputStream inputStream = inputFile.newStream();
+ ParquetMetadata footer =
+ ParquetFileReader.readFooter(
+ inputFile, readOptionsBuilder.build(), inputStream,
true);
+
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ ParquetFileReader reader;
+ try {
+ FilterCompat.Filter filter =
+ ParquetFilters.convert(predicates, fileSchema,
caseSensitive);
+ ParquetReadOptions readOptions =
readOptionsBuilder.withRecordFilter(filter).build();
+ reader =
+ new ParquetFileReader(
+ inputFile, footer, readOptions, inputStream,
context.selection());
+ } catch (Throwable t) {
+ try {
+ inputStream.close();
+ } catch (Throwable closeFailure) {
+ t.addSuppressed(closeFailure);
+ }
+ throw t;
+ }
+
ShreddingReadPlan readPlan =
ShreddingReadPlanFactories.createReadPlan(
readType,
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 29feeb1b5e..f6ffbed1e2 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,7 +21,6 @@ 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;
@@ -55,25 +54,34 @@ import org.apache.parquet.filter2.compat.FilterCompat;
import org.apache.parquet.filter2.predicate.Operators.DoubleColumn;
import org.apache.parquet.filter2.predicate.Operators.FloatColumn;
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.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
import java.io.Serializable;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
+import java.util.Objects;
import java.util.Set;
/** Convert {@link Predicate} to {@link FilterCompat.Filter}. */
public class ParquetFilters {
- private static final ConvertFilterToParquet CONVERTER = new
ConvertFilterToParquet();
-
private ParquetFilters() {}
- public static FilterCompat.Filter convert(List<Predicate> predicates) {
+ public static FilterCompat.Filter convert(
+ List<Predicate> predicates, MessageType fileSchema, boolean
caseSensitive) {
+ ConvertFilterToParquet converter = new
ConvertFilterToParquet(fileSchema, caseSensitive);
FilterPredicate result = null;
if (predicates != null) {
for (Predicate predicate : predicates) {
try {
- FilterPredicate parquetFilter = predicate.visit(CONVERTER);
+ FilterPredicate parquetFilter = predicate.visit(converter);
if (result == null) {
result = parquetFilter;
} else {
@@ -90,6 +98,14 @@ public class ParquetFilters {
@SuppressWarnings({"unchecked", "rawtypes"})
private static class ConvertFilterToParquet implements
FunctionVisitor<FilterPredicate> {
+ private final MessageType fileSchema;
+ private final boolean caseSensitive;
+
+ private ConvertFilterToParquet(MessageType fileSchema, boolean
caseSensitive) {
+ this.fileSchema = Objects.requireNonNull(fileSchema, "fileSchema");
+ this.caseSensitive = caseSensitive;
+ }
+
@Override
public FilterPredicate visitIsNotNull(FieldRef fieldRef) {
return new Operators.NotEq<>(toParquetColumn(fieldRef), null);
@@ -114,38 +130,35 @@ public class ParquetFilters {
@Override
public FilterPredicate visitLessThan(FieldRef fieldRef, Object
literal) {
- return new Operators.Lt(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ return new Operators.Lt(toParquetColumn(fieldRef),
toParquetObject(literal, fieldRef));
}
@Override
public FilterPredicate visitGreaterOrEqual(FieldRef fieldRef, Object
literal) {
return new Operators.GtEq(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef));
}
@Override
public FilterPredicate visitNotEqual(FieldRef fieldRef, Object
literal) {
return new Operators.NotEq(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef));
}
@Override
public FilterPredicate visitLessOrEqual(FieldRef fieldRef, Object
literal) {
return new Operators.LtEq(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef));
}
@Override
public FilterPredicate visitEqual(FieldRef fieldRef, Object literal) {
- return new Operators.Eq(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ return new Operators.Eq(toParquetColumn(fieldRef),
toParquetObject(literal, fieldRef));
}
@Override
public FilterPredicate visitGreaterThan(FieldRef fieldRef, Object
literal) {
- return new Operators.Gt(
- toParquetColumn(fieldRef), toParquetObject(literal,
fieldRef.type()));
+ return new Operators.Gt(toParquetColumn(fieldRef),
toParquetObject(literal, fieldRef));
}
@Override
@@ -189,22 +202,25 @@ public class ParquetFilters {
@Override
public FilterPredicate visitIn(FieldRef fieldRef, List<Object>
literals) {
Operators.Column<?> column = toParquetColumn(fieldRef);
- org.apache.paimon.types.DataType type = fieldRef.type();
if (column instanceof Operators.LongColumn) {
return FilterApi.in(
- (Operators.LongColumn) column, convertSets(literals,
Long.class, type));
+ (Operators.LongColumn) column, convertSets(literals,
Long.class, fieldRef));
} else if (column instanceof Operators.IntColumn) {
return FilterApi.in(
- (Operators.IntColumn) column, convertSets(literals,
Integer.class, type));
+ (Operators.IntColumn) column,
+ convertSets(literals, Integer.class, fieldRef));
} else if (column instanceof Operators.DoubleColumn) {
return FilterApi.in(
- (Operators.DoubleColumn) column, convertSets(literals,
Double.class, type));
+ (Operators.DoubleColumn) column,
+ convertSets(literals, Double.class, fieldRef));
} else if (column instanceof Operators.FloatColumn) {
return FilterApi.in(
- (Operators.FloatColumn) column, convertSets(literals,
Float.class, type));
+ (Operators.FloatColumn) column,
+ convertSets(literals, Float.class, fieldRef));
} else if (column instanceof Operators.BinaryColumn) {
return FilterApi.in(
- (Operators.BinaryColumn) column, convertSets(literals,
Binary.class, type));
+ (Operators.BinaryColumn) column,
+ convertSets(literals, Binary.class, fieldRef));
}
throw new UnsupportedOperationException();
@@ -213,22 +229,25 @@ public class ParquetFilters {
@Override
public FilterPredicate visitNotIn(FieldRef fieldRef, List<Object>
literals) {
Operators.Column<?> column = toParquetColumn(fieldRef);
- org.apache.paimon.types.DataType type = fieldRef.type();
if (column instanceof Operators.LongColumn) {
return FilterApi.notIn(
- (Operators.LongColumn) column, convertSets(literals,
Long.class, type));
+ (Operators.LongColumn) column, convertSets(literals,
Long.class, fieldRef));
} else if (column instanceof Operators.IntColumn) {
return FilterApi.notIn(
- (Operators.IntColumn) column, convertSets(literals,
Integer.class, type));
+ (Operators.IntColumn) column,
+ convertSets(literals, Integer.class, fieldRef));
} else if (column instanceof Operators.DoubleColumn) {
return FilterApi.notIn(
- (Operators.DoubleColumn) column, convertSets(literals,
Double.class, type));
+ (Operators.DoubleColumn) column,
+ convertSets(literals, Double.class, fieldRef));
} else if (column instanceof Operators.FloatColumn) {
return FilterApi.notIn(
- (Operators.FloatColumn) column, convertSets(literals,
Float.class, type));
+ (Operators.FloatColumn) column,
+ convertSets(literals, Float.class, fieldRef));
} else if (column instanceof Operators.BinaryColumn) {
return FilterApi.notIn(
- (Operators.BinaryColumn) column, convertSets(literals,
Binary.class, type));
+ (Operators.BinaryColumn) column,
+ convertSets(literals, Binary.class, fieldRef));
}
throw new UnsupportedOperationException();
@@ -238,88 +257,185 @@ public class ParquetFilters {
public FilterPredicate visitNonFieldLeaf(LeafPredicate predicate) {
throw new UnsupportedOperationException();
}
- }
- private static <T> Set<T> convertSets(
- List<Object> values, Class<T> kclass,
org.apache.paimon.types.DataType type) {
- Set<T> converted = new HashSet<>();
- for (Object value : values) {
- Comparable<?> cmp = toParquetObject(value, type);
- if (kclass.isInstance(cmp)) {
- converted.add((T) cmp);
- } else {
- throw new UnsupportedOperationException();
+ private <T> Set<T> convertSets(List<Object> values, Class<T> kclass,
FieldRef fieldRef) {
+ Set<T> converted = new HashSet<>();
+ for (Object value : values) {
+ Comparable<?> cmp = toParquetObject(value, fieldRef);
+ if (kclass.isInstance(cmp)) {
+ converted.add((T) cmp);
+ } else {
+ throw new UnsupportedOperationException();
+ }
}
+ return converted;
}
- return converted;
- }
- private static int getTimestampPrecision(org.apache.paimon.types.DataType
type) {
- if (type instanceof TimestampType) {
- return ((TimestampType) type).getPrecision();
- } else if (type instanceof LocalZonedTimestampType) {
- return ((LocalZonedTimestampType) type).getPrecision();
+ private Operators.Column<?> toParquetColumn(FieldRef fieldRef) {
+ return fieldRef.type()
+ .accept(new ConvertToColumnTypeVisitor(fieldRef,
fileSchema, caseSensitive));
}
- throw new IllegalArgumentException("Not a timestamp type: " + type);
- }
- private static Operators.Column<?> toParquetColumn(FieldRef fieldRef) {
- return fieldRef.type().accept(new
ConvertToColumnTypeVisitor(fieldRef.name()));
- }
+ private Comparable<?> toParquetObject(Object value, FieldRef fieldRef)
{
+ if (value == null) {
+ return null;
+ }
+
+ org.apache.paimon.types.DataType type = fieldRef.type();
+ if (type instanceof DecimalType) {
+ DecimalType decimalType = (DecimalType) fieldRef.type();
+ Decimal decimal = normalizeDecimal((Decimal) value,
decimalType);
+ PrimitiveType primitiveType =
+ decimalPrimitiveType(fieldRef, fileSchema,
caseSensitive);
+ switch (primitiveType.getPrimitiveTypeName()) {
+ case INT32:
+ long intValue = toUnscaledLong(decimal);
+ if (intValue < Integer.MIN_VALUE || intValue >
Integer.MAX_VALUE) {
+ throw new UnsupportedOperationException();
+ }
+ return (int) intValue;
+ case INT64:
+ return toUnscaledLong(decimal);
+ case BINARY:
+ return
Binary.fromConstantByteArray(decimal.toUnscaledBytes());
+ case FIXED_LEN_BYTE_ARRAY:
+ return decimalToBinary(decimal,
primitiveType.getTypeLength());
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
- private static Comparable<?> toParquetObject(
- Object value, org.apache.paimon.types.DataType type) {
- if (value == null) {
- return null;
+ 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) {
+ Timestamp timestamp = (Timestamp) value;
+ int precision = getTimestampPrecision(type);
+ if (precision <= 3) {
+ // milliseconds
+ return timestamp.getMillisecond();
+ } else if (precision <= 6) {
+ // microseconds
+ return timestamp.toMicros();
+ }
+ // precision > 6 uses INT96, not supported
+ throw new UnsupportedOperationException();
+ }
+
+ throw new UnsupportedOperationException();
}
- if (value instanceof Number) {
- if (value instanceof Byte) {
- return ((Byte) value).intValue();
- } else if (value instanceof Short) {
- return ((Short) value).intValue();
+ private Decimal normalizeDecimal(Decimal decimal, DecimalType
fieldType) {
+ try {
+ BigDecimal normalized =
+ decimal.toBigDecimal()
+ .setScale(fieldType.getScale(),
RoundingMode.UNNECESSARY);
+ Decimal result =
+ Decimal.fromBigDecimal(
+ normalized, fieldType.getPrecision(),
fieldType.getScale());
+ if (result == null) {
+ throw new UnsupportedOperationException();
+ }
+ return result;
+ } catch (ArithmeticException e) {
+ throw new UnsupportedOperationException(e);
}
- 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 Decimal) {
- Decimal decimal = (Decimal) value;
- int precision = decimal.precision();
- if (ParquetSchemaConverter.is32BitDecimal(precision)) {
- return (int) decimal.toUnscaledLong();
- } else if (ParquetSchemaConverter.is64BitDecimal(precision)) {
+ }
+
+ private long toUnscaledLong(Decimal decimal) {
+ try {
return decimal.toUnscaledLong();
- } else {
- return Binary.fromConstantByteArray(decimal.toUnscaledBytes());
+ } catch (ArithmeticException e) {
+ throw new UnsupportedOperationException(e);
+ }
+ }
+
+ private Binary decimalToBinary(Decimal decimal, int numBytes) {
+ byte[] unscaledBytes = decimal.toUnscaledBytes();
+ if (unscaledBytes.length > numBytes) {
+ throw new UnsupportedOperationException();
}
- } else if (value instanceof Timestamp) {
- Timestamp timestamp = (Timestamp) value;
- int precision = getTimestampPrecision(type);
- if (precision <= 3) {
- // milliseconds
- return timestamp.getMillisecond();
- } else if (precision <= 6) {
- // microseconds
- return timestamp.toMicros();
+ if (unscaledBytes.length == numBytes) {
+ return Binary.fromConstantByteArray(unscaledBytes);
}
- // precision > 6 uses INT96, not supported
+
+ byte[] paddedBytes = new byte[numBytes];
+ Arrays.fill(paddedBytes, unscaledBytes[0] < 0 ? (byte) -1 : (byte)
0);
+ System.arraycopy(
+ unscaledBytes,
+ 0,
+ paddedBytes,
+ numBytes - unscaledBytes.length,
+ unscaledBytes.length);
+ return Binary.fromConstantByteArray(paddedBytes);
+ }
+ }
+
+ private static PrimitiveType decimalPrimitiveType(
+ FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) {
+ Type matched = null;
+ // 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 (matched == null || !matched.isPrimitive()) {
throw new UnsupportedOperationException();
}
- throw new UnsupportedOperationException();
+ PrimitiveType primitiveType = matched.asPrimitiveType();
+ LogicalTypeAnnotation logicalType =
primitiveType.getLogicalTypeAnnotation();
+ if (!(logicalType instanceof DecimalLogicalTypeAnnotation)) {
+ throw new UnsupportedOperationException();
+ }
+
+ DecimalLogicalTypeAnnotation decimalLogicalType =
+ (DecimalLogicalTypeAnnotation) logicalType;
+ if (decimalLogicalType.getScale() != ((DecimalType)
fieldRef.type()).getScale()) {
+ throw new UnsupportedOperationException();
+ }
+ return primitiveType;
+ }
+
+ private static int getTimestampPrecision(org.apache.paimon.types.DataType
type) {
+ if (type instanceof TimestampType) {
+ return ((TimestampType) type).getPrecision();
+ } else if (type instanceof LocalZonedTimestampType) {
+ return ((LocalZonedTimestampType) type).getPrecision();
+ }
+ throw new IllegalArgumentException("Not a timestamp type: " + type);
}
private static class ConvertToColumnTypeVisitor
implements DataTypeVisitor<Operators.Column<?>> {
+ private final FieldRef fieldRef;
private final String name;
+ private final MessageType fileSchema;
+ private final boolean caseSensitive;
- public ConvertToColumnTypeVisitor(String name) {
- this.name = name;
+ public ConvertToColumnTypeVisitor(
+ FieldRef fieldRef, MessageType fileSchema, boolean
caseSensitive) {
+ this.fieldRef = fieldRef;
+ this.name = fieldRef.name();
+ this.fileSchema = fileSchema;
+ this.caseSensitive = caseSensitive;
}
@Override
@@ -389,13 +505,17 @@ public class ParquetFilters {
@Override
public Operators.Column<?> visit(DecimalType decimalType) {
- int precision = decimalType.getPrecision();
- if (ParquetSchemaConverter.is32BitDecimal(precision)) {
- return FilterApi.intColumn(name);
- } else if (ParquetSchemaConverter.is64BitDecimal(precision)) {
- return FilterApi.longColumn(name);
- } else {
- return FilterApi.binaryColumn(name);
+ PrimitiveType primitiveType = decimalPrimitiveType(fieldRef,
fileSchema, caseSensitive);
+ switch (primitiveType.getPrimitiveTypeName()) {
+ case INT32:
+ return FilterApi.intColumn(fieldRef.name());
+ case INT64:
+ return FilterApi.longColumn(fieldRef.name());
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return FilterApi.binaryColumn(fieldRef.name());
+ default:
+ throw new UnsupportedOperationException();
}
}
diff --git
a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
index 6f3a20be59..9f56c21769 100644
---
a/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
+++
b/paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
@@ -136,7 +136,7 @@ public class ParquetFileReader implements Closeable {
return readFooter(file, options, f, /*closeStreamOnFailure*/ false);
}
- private static final ParquetMetadata readFooter(
+ public static final ParquetMetadata readFooter(
InputFile file,
ParquetReadOptions options,
SeekableInputStream f,
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
index f8c22de58b..41933a3d29 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetCaseInsensitiveReadTest.java
@@ -33,7 +33,6 @@ 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.filter2.compat.FilterCompat;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
import org.apache.parquet.hadoop.util.HadoopOutputFile;
@@ -134,8 +133,7 @@ class ParquetCaseInsensitiveReadTest {
throws Exception {
Options options = new Options();
options.set(CatalogOptions.CASE_SENSITIVE, caseSensitive);
- ParquetReaderFactory factory =
- new ParquetReaderFactory(options, readType, 1024,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(options,
readType, 1024, null);
LocalFileIO fileIO = new LocalFileIO();
List<InternalRow> rows = new ArrayList<>();
try (RecordReader<InternalRow> reader =
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetColumnVectorTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetColumnVectorTest.java
index ee20945d64..25e5edf392 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetColumnVectorTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetColumnVectorTest.java
@@ -38,7 +38,6 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.StringUtils;
-import org.apache.parquet.filter2.compat.FilterCompat;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -593,7 +592,7 @@ public class ParquetColumnVectorTest {
writer.close();
ParquetReaderFactory readerFactory =
- new ParquetReaderFactory(new Options(), rowType, 1024,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), rowType, 1024, null);
RecordReader<InternalRow> reader =
readerFactory.createReader(
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
index a10efa2cf2..acb6e6e5bd 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
@@ -32,18 +32,37 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VarCharType;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.EncodingStats;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.DictionaryPageReadStore;
import org.apache.parquet.filter2.compat.FilterCompat;
import org.apache.parquet.filter2.compat.FilterCompat.FilterPredicateCompat;
+import org.apache.parquet.filter2.dictionarylevel.DictionaryFilter;
import org.apache.parquet.filter2.predicate.FilterApi;
import org.apache.parquet.filter2.predicate.FilterPredicate;
import org.apache.parquet.filter2.predicate.ParquetFilters;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Types;
import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import java.util.stream.LongStream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -52,28 +71,29 @@ class ParquetFiltersTest {
@Test
public void testLong() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "long1", new
BigIntType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0,
"long1", new BigIntType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
- test(builder.isNull(0), "eq(long1, null)", true);
+ test(schema, builder.isNull(0), "eq(long1, null)", true);
- test(builder.isNotNull(0), "noteq(long1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(long1, null)", true);
- test(builder.lessThan(0, 5L), "lt(long1, 5)", true);
+ test(schema, builder.lessThan(0, 5L), "lt(long1, 5)", true);
- test(builder.greaterThan(0, 5L), "gt(long1, 5)", true);
+ test(schema, builder.greaterThan(0, 5L), "gt(long1, 5)", true);
test(
+ schema,
builder.in(0, Arrays.asList(1L, 2L, 3L)),
"or(eq(long1, 1), or(eq(long1, 2), eq(long1, 3)))",
true);
- test(builder.between(0, 1L, 3L), "and(gteq(long1, 1), lteq(long1,
3))", true);
+ test(schema, builder.between(0, 1L, 3L), "and(gteq(long1, 1),
lteq(long1, 3))", true);
test(
+ schema,
builder.notIn(0, Arrays.asList(1L, 2L, 3L)),
"and(and(noteq(long1, 1), noteq(long1, 2)), noteq(long1, 3))",
true);
@@ -81,20 +101,22 @@ class ParquetFiltersTest {
@Test
public void testString() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "string1", new
VarCharType()))));
- test(builder.isNull(0), "eq(string1, null)", true);
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(new DataField(0, "string1",
new VarCharType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ test(schema, builder.isNull(0), "eq(string1, null)", true);
- test(builder.isNotNull(0), "noteq(string1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(string1, null)", true);
test(
+ schema,
builder.in(0, Arrays.asList("1", "2", "3")),
"or(eq(string1, Binary{\"1\"}), or(eq(string1, Binary{\"2\"}),
eq(string1, Binary{\"3\"})))",
true);
test(
+ schema,
builder.notIn(0, Arrays.asList("1", "2", "3")),
"and(and(noteq(string1, Binary{\"1\"}), noteq(string1,
Binary{\"2\"})), noteq(string1, Binary{\"3\"}))",
true);
@@ -102,12 +124,12 @@ class ParquetFiltersTest {
@Test
public void testInFilterLong() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "col1", new
BigIntType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "col1",
new BigIntType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
test(
+ schema,
builder.in(0, LongStream.range(1L,
22L).boxed().collect(Collectors.toList())),
FilterApi.in(
FilterApi.longColumn("col1"),
@@ -115,6 +137,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(0, LongStream.range(1L,
22L).boxed().collect(Collectors.toList())),
FilterApi.notIn(
FilterApi.longColumn("col1"),
@@ -124,12 +147,12 @@ class ParquetFiltersTest {
@Test
public void testInFilterDouble() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "col1", new
DoubleType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "col1",
new DoubleType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
test(
+ schema,
builder.in(
0,
LongStream.range(1L, 22L)
@@ -145,6 +168,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(
0,
LongStream.range(1L, 22L)
@@ -162,12 +186,12 @@ class ParquetFiltersTest {
@Test
public void testInFilterString() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "col1", new
VarCharType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "col1",
new VarCharType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
test(
+ schema,
builder.in(
0,
LongStream.range(1L, 22L)
@@ -183,6 +207,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(
0,
LongStream.range(1L, 22L)
@@ -200,14 +225,14 @@ class ParquetFiltersTest {
@Test
public void testIsNaNDouble() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "d1", new
DoubleType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "d1",
new DoubleType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Predicate predicate = builder.isNaN(0);
FilterCompat.Filter filter =
-
ParquetFilters.convert(Collections.singletonList(builder.isNaN(0)));
+ ParquetFilters.convert(Collections.singletonList(predicate),
schema, true);
FilterPredicateCompat compat = (FilterPredicateCompat) filter;
assertThat(compat.getFilterPredicate().toString())
.contains(
@@ -216,14 +241,14 @@ class ParquetFiltersTest {
@Test
public void testIsNaNFloat() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "f1", new
FloatType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "f1",
new FloatType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ Predicate predicate = builder.isNaN(0);
FilterCompat.Filter filter =
-
ParquetFilters.convert(Collections.singletonList(builder.isNaN(0)));
+ ParquetFilters.convert(Collections.singletonList(predicate),
schema, true);
FilterPredicateCompat compat = (FilterPredicateCompat) filter;
assertThat(compat.getFilterPredicate().toString())
.contains(
@@ -232,13 +257,13 @@ class ParquetFiltersTest {
@Test
public void testInFilterFloat() {
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "col1", new
FloatType()))));
+ RowType rowType =
+ new RowType(Collections.singletonList(new DataField(0, "col1",
new FloatType())));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
test(
+ schema,
builder.in(
0,
LongStream.range(1L, 22L)
@@ -254,6 +279,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(
0,
LongStream.range(1L, 22L)
@@ -274,6 +300,8 @@ class ParquetFiltersTest {
// precision <= 9 uses INT32
int precision = 9;
int scale = 2;
+ MessageType schema =
+ decimalSchema("decimal1", PrimitiveTypeName.INT32, 0,
precision, scale);
PredicateBuilder builder =
new PredicateBuilder(
new RowType(
@@ -286,14 +314,18 @@ class ParquetFiltersTest {
Decimal value = Decimal.fromBigDecimal(new BigDecimal("123.45"),
precision, scale);
int expectedIntVal = (int) value.toUnscaledLong(); // 12345
- test(builder.isNull(0), "eq(decimal1, null)", true);
- test(builder.isNotNull(0), "noteq(decimal1, null)", true);
- test(builder.equal(0, value), "eq(decimal1, " + expectedIntVal + ")",
true);
- test(builder.notEqual(0, value), "noteq(decimal1, " + expectedIntVal +
")", true);
- test(builder.lessThan(0, value), "lt(decimal1, " + expectedIntVal +
")", true);
- test(builder.lessOrEqual(0, value), "lteq(decimal1, " + expectedIntVal
+ ")", true);
- test(builder.greaterThan(0, value), "gt(decimal1, " + expectedIntVal +
")", true);
- test(builder.greaterOrEqual(0, value), "gteq(decimal1, " +
expectedIntVal + ")", true);
+ test(schema, builder.isNull(0), "eq(decimal1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(decimal1, null)", true);
+ test(schema, builder.equal(0, value), "eq(decimal1, " + expectedIntVal
+ ")", true);
+ test(schema, builder.notEqual(0, value), "noteq(decimal1, " +
expectedIntVal + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(decimal1, " +
expectedIntVal + ")", true);
+ test(schema, builder.lessOrEqual(0, value), "lteq(decimal1, " +
expectedIntVal + ")", true);
+ test(schema, builder.greaterThan(0, value), "gt(decimal1, " +
expectedIntVal + ")", true);
+ test(
+ schema,
+ builder.greaterOrEqual(0, value),
+ "gteq(decimal1, " + expectedIntVal + ")",
+ true);
}
@Test
@@ -301,6 +333,8 @@ class ParquetFiltersTest {
// 9 < precision <= 18 uses INT64
int precision = 18;
int scale = 4;
+ MessageType schema =
+ decimalSchema("decimal1", PrimitiveTypeName.INT64, 0,
precision, scale);
PredicateBuilder builder =
new PredicateBuilder(
new RowType(
@@ -314,21 +348,115 @@ class ParquetFiltersTest {
Decimal.fromBigDecimal(new BigDecimal("12345678901234.5678"),
precision, scale);
long expectedLongVal = value.toUnscaledLong();
- test(builder.isNull(0), "eq(decimal1, null)", true);
- test(builder.isNotNull(0), "noteq(decimal1, null)", true);
- test(builder.equal(0, value), "eq(decimal1, " + expectedLongVal + ")",
true);
- test(builder.notEqual(0, value), "noteq(decimal1, " + expectedLongVal
+ ")", true);
- test(builder.lessThan(0, value), "lt(decimal1, " + expectedLongVal +
")", true);
- test(builder.lessOrEqual(0, value), "lteq(decimal1, " +
expectedLongVal + ")", true);
- test(builder.greaterThan(0, value), "gt(decimal1, " + expectedLongVal
+ ")", true);
- test(builder.greaterOrEqual(0, value), "gteq(decimal1, " +
expectedLongVal + ")", true);
+ test(schema, builder.isNull(0), "eq(decimal1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(decimal1, null)", true);
+ test(schema, builder.equal(0, value), "eq(decimal1, " +
expectedLongVal + ")", true);
+ test(schema, builder.notEqual(0, value), "noteq(decimal1, " +
expectedLongVal + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(decimal1, " +
expectedLongVal + ")", true);
+ test(
+ schema,
+ builder.lessOrEqual(0, value),
+ "lteq(decimal1, " + expectedLongVal + ")",
+ true);
+ test(schema, builder.greaterThan(0, value), "gt(decimal1, " +
expectedLongVal + ")", true);
+ test(
+ schema,
+ builder.greaterOrEqual(0, value),
+ "gteq(decimal1, " + expectedLongVal + ")",
+ true);
}
@Test
public void testDecimalBinary() {
// precision > 18 uses Binary
+ int fieldPrecision = 20;
+ int literalPrecision = 8;
+ int scale = 0;
+ MessageType schema =
+ decimalSchema(
+ "decimal1",
+ PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
+ 9,
+ fieldPrecision,
+ scale);
+ PredicateBuilder builder =
+ new PredicateBuilder(
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ 0,
+ "decimal1",
+ new
DecimalType(fieldPrecision, scale)))));
+
+ Decimal positive =
+ Decimal.fromBigDecimal(new BigDecimal("10000939"),
literalPrecision, scale);
+ Binary expectedPositive =
+ Binary.fromConstantByteArray(
+ new byte[] {0, 0, 0, 0, 0, 0, (byte) 0x98, (byte)
0x9A, 0x2B});
+ Decimal negative =
+ Decimal.fromBigDecimal(new BigDecimal("-10000939"),
literalPrecision, scale);
+ Binary expectedNegative =
+ Binary.fromConstantByteArray(
+ new byte[] {
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x67,
+ 0x65,
+ (byte) 0xD5
+ });
+
+ test(schema, builder.isNull(0), "eq(decimal1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(decimal1, null)", true);
+ test(
+ schema,
+ builder.equal(0, positive),
+ FilterApi.eq(FilterApi.binaryColumn("decimal1"),
expectedPositive),
+ true);
+ test(
+ schema,
+ builder.notEqual(0, positive),
+ FilterApi.notEq(FilterApi.binaryColumn("decimal1"),
expectedPositive),
+ true);
+ test(
+ schema,
+ builder.lessThan(0, positive),
+ FilterApi.lt(FilterApi.binaryColumn("decimal1"),
expectedPositive),
+ true);
+ test(
+ schema,
+ builder.greaterThan(0, positive),
+ FilterApi.gt(FilterApi.binaryColumn("decimal1"),
expectedPositive),
+ true);
+ test(
+ schema,
+ builder.equal(0, negative),
+ FilterApi.eq(FilterApi.binaryColumn("decimal1"),
expectedNegative),
+ true);
+
+ Decimal fullWidth =
+ Decimal.fromBigDecimal(
+ new BigDecimal("99999999999999999999"),
fieldPrecision, scale);
+ assertThat(fullWidth.toUnscaledBytes()).hasSize(9);
+ test(
+ schema,
+ builder.equal(0, fullWidth),
+ FilterApi.eq(
+ FilterApi.binaryColumn("decimal1"),
+
Binary.fromConstantByteArray(fullWidth.toUnscaledBytes())),
+ true);
+ }
+
+ @Test
+ public void testDecimalBinaryMaxPrecision() {
int precision = 38;
int scale = 10;
+ MessageType schema =
+ decimalSchema(
+ "decimal1", PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
16, precision, scale);
PredicateBuilder builder =
new PredicateBuilder(
new RowType(
@@ -337,29 +465,284 @@ class ParquetFiltersTest {
0,
"decimal1",
new DecimalType(precision,
scale)))));
-
Decimal value =
Decimal.fromBigDecimal(
new BigDecimal("12345678901234567890.1234567890"),
precision, scale);
- Binary expectedBinary =
Binary.fromConstantByteArray(value.toUnscaledBytes());
+ Binary expected =
+ Binary.fromConstantByteArray(
+ new byte[] {
+ 0,
+ 0,
+ 0,
+ 1,
+ (byte) 0x8E,
+ (byte) 0xE9,
+ 0x0F,
+ (byte) 0xF6,
+ (byte) 0xC3,
+ 0x73,
+ (byte) 0xE0,
+ (byte) 0xEE,
+ 0x4E,
+ 0x3F,
+ 0x0A,
+ (byte) 0xD2
+ });
+
+ test(
+ schema,
+ builder.equal(0, value),
+ FilterApi.eq(FilterApi.binaryColumn("decimal1"), expected),
+ true);
+ }
+
+ @Test
+ public void testInFilterDecimalBinary() {
+ int fieldPrecision = 20;
+ int scale = 0;
+ MessageType schema =
+ decimalSchema(
+ "decimal1",
+ PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
+ 9,
+ fieldPrecision,
+ scale);
+ PredicateBuilder builder =
+ new PredicateBuilder(
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ 0,
+ "decimal1",
+ new
DecimalType(fieldPrecision, scale)))));
+
+ List<Object> literals =
+ IntStream.rangeClosed(1, 21)
+ .mapToObj(
+ value ->
+ (Object)
+ Decimal.fromBigDecimal(
+
BigDecimal.valueOf(value),
+ fieldPrecision,
+ scale))
+ .collect(Collectors.toList());
+ Set<Binary> expected =
+ IntStream.rangeClosed(1, 21)
+ .mapToObj(
+ value -> {
+ byte[] bytes = new byte[9];
+ bytes[8] = (byte) value;
+ return Binary.fromConstantByteArray(bytes);
+ })
+ .collect(Collectors.toSet());
+
+ test(
+ schema,
+ builder.in(0, literals),
+ FilterApi.in(FilterApi.binaryColumn("decimal1"), expected),
+ true);
+ test(
+ schema,
+ builder.notIn(0, literals),
+ FilterApi.notIn(FilterApi.binaryColumn("decimal1"), expected),
+ true);
+ }
+
+ @Test
+ public void testDecimalDictionaryFilter() {
+ int fieldPrecision = 20;
+ int scale = 0;
+ PredicateBuilder builder =
+ new PredicateBuilder(
+ new RowType(
+ Collections.singletonList(
+ new DataField(
+ 0,
+ "decimal1",
+ new
DecimalType(fieldPrecision, scale)))));
+ Decimal positive =
+ Decimal.fromBigDecimal(new BigDecimal("10000939"),
fieldPrecision, scale);
+ Decimal negative =
+ Decimal.fromBigDecimal(new BigDecimal("-10000939"),
fieldPrecision, scale);
+ Decimal missing = Decimal.fromBigDecimal(new BigDecimal("10000940"),
fieldPrecision, scale);
+
+ byte[] positiveBytes = new byte[] {0, 0, 0, 0, 0, 0, (byte) 0x98,
(byte) 0x9A, 0x2B};
+ byte[] negativeBytes =
+ new byte[] {
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ (byte) 0xFF,
+ 0x67,
+ 0x65,
+ (byte) 0xD5
+ };
+ DictionaryPage dictionaryPage =
+ new DictionaryPage(
+ BytesInput.concat(
+ BytesInput.from(positiveBytes),
BytesInput.from(negativeBytes)),
+ 2,
+ Encoding.PLAIN);
+ DictionaryPageReadStore dictionaries =
+ new DictionaryPageReadStore() {
+ @Override
+ public DictionaryPage readDictionaryPage(ColumnDescriptor
descriptor) {
+ return dictionaryPage;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ PrimitiveType primitiveType =
+ Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
+ .length(9)
+ .as(LogicalTypeAnnotation.decimalType(scale,
fieldPrecision))
+ .named("decimal1");
+ MessageType schema =
+ new MessageType("paimon_schema",
Collections.singletonList(primitiveType));
+ EncodingStats encodingStats =
+ new EncodingStats.Builder()
+ .addDictEncoding(Encoding.PLAIN)
+ .addDataEncoding(Encoding.RLE_DICTIONARY)
+ .build();
+ Set<Encoding> encodings =
+ new HashSet<>(Arrays.asList(Encoding.PLAIN,
Encoding.RLE_DICTIONARY, Encoding.RLE));
+ ColumnChunkMetaData metadata =
+ ColumnChunkMetaData.get(
+ ColumnPath.get("decimal1"),
+ primitiveType,
+ CompressionCodecName.UNCOMPRESSED,
+ encodingStats,
+ encodings,
+ null,
+ 0,
+ 0,
+ 2,
+ 0,
+ 0);
+
+ assertThat(
+ DictionaryFilter.canDrop(
+ convert(schema, builder.equal(0, positive)),
+ Collections.singletonList(metadata),
+ dictionaries))
+ .isFalse();
+ assertThat(
+ DictionaryFilter.canDrop(
+ convert(schema, builder.equal(0, negative)),
+ Collections.singletonList(metadata),
+ dictionaries))
+ .isFalse();
+ assertThat(
+ DictionaryFilter.canDrop(
+ convert(schema, builder.equal(0, missing)),
+ Collections.singletonList(metadata),
+ dictionaries))
+ .isTrue();
+ }
+
+ @Test
+ public void testDecimalLiteralOutsideFieldDomain() {
+ int fieldPrecision = 9;
+ int scale = 0;
+ PredicateBuilder builder = decimalPredicateBuilder(fieldPrecision,
scale);
+ Decimal value = Decimal.fromBigDecimal(new BigDecimal("4294967297"),
10, scale);
+ MessageType schema =
+ decimalSchema("decimal1", PrimitiveTypeName.INT32, 0,
fieldPrecision, scale);
+
+ test(schema, builder.equal(0, value), "", false);
+ }
+
+ @Test
+ public void testDecimalLiteralWiderThanPhysicalWidth() {
+ int fieldPrecision = 20;
+ int scale = 0;
+ PredicateBuilder builder = decimalPredicateBuilder(fieldPrecision,
scale);
+ Decimal value = Decimal.fromBigDecimal(new BigDecimal("3000000000"),
10, scale);
+ MessageType schema =
+ decimalSchema("decimal1",
PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, 4, 9, scale);
+
+ test(schema, builder.equal(0, value), "", false);
+ }
+
+ @Test
+ public void testDecimalLiteralWiderThanFieldDomain() {
+ int fieldPrecision = 20;
+ int scale = 0;
+ PredicateBuilder builder = decimalPredicateBuilder(fieldPrecision,
scale);
+ Decimal value =
+ Decimal.fromBigDecimal(
+ new
BigDecimal("99999999999999999999999999999999999999"), 38, scale);
+ MessageType schema =
+ decimalSchema(
+ "decimal1",
+ PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
+ 9,
+ fieldPrecision,
+ scale);
+
+ test(schema, builder.equal(0, value), "", false);
+ }
+
+ @Test
+ public void testDecimalScaleNormalization() {
+ int fieldPrecision = 20;
+ int fieldScale = 2;
+ PredicateBuilder builder = decimalPredicateBuilder(fieldPrecision,
fieldScale);
+ MessageType schema =
+ decimalSchema(
+ "decimal1",
+ PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
+ 9,
+ fieldPrecision,
+ fieldScale);
+ Decimal exact = Decimal.fromBigDecimal(new BigDecimal("100"), 3, 0);
+ Decimal inexact = Decimal.fromBigDecimal(new BigDecimal("100.001"), 6,
3);
+ Binary expected =
+ Binary.fromConstantByteArray(new byte[] {0, 0, 0, 0, 0, 0, 0,
0x27, 0x10});
- test(builder.isNull(0), "eq(decimal1, null)", true);
- test(builder.isNotNull(0), "noteq(decimal1, null)", true);
test(
+ schema,
+ builder.equal(0, exact),
+ FilterApi.eq(FilterApi.binaryColumn("decimal1"), expected),
+ true);
+ test(schema, builder.equal(0, inexact), "", false);
+ }
+
+ @Test
+ public void testDecimalPhysicalTypes() {
+ int precision = 9;
+ int scale = 2;
+ PredicateBuilder builder = decimalPredicateBuilder(precision, scale);
+ Decimal value = Decimal.fromBigDecimal(new BigDecimal("12.34"),
precision, scale);
+
+ test(
+ decimalSchema("decimal1", PrimitiveTypeName.INT32, 0,
precision, scale),
builder.equal(0, value),
- FilterApi.eq(FilterApi.binaryColumn("decimal1"),
expectedBinary),
+ FilterApi.eq(FilterApi.intColumn("decimal1"), 1234),
true);
test(
- builder.notEqual(0, value),
- FilterApi.notEq(FilterApi.binaryColumn("decimal1"),
expectedBinary),
+ decimalSchema("decimal1", PrimitiveTypeName.INT64, 0,
precision, scale),
+ builder.equal(0, value),
+ FilterApi.eq(FilterApi.longColumn("decimal1"), 1234L),
true);
test(
- builder.lessThan(0, value),
- FilterApi.lt(FilterApi.binaryColumn("decimal1"),
expectedBinary),
+ decimalSchema("decimal1", PrimitiveTypeName.BINARY, 0,
precision, scale),
+ builder.equal(0, value),
+ FilterApi.eq(
+ FilterApi.binaryColumn("decimal1"),
+ Binary.fromConstantByteArray(new byte[] {0x04, (byte)
0xD2})),
true);
test(
- builder.greaterThan(0, value),
- FilterApi.gt(FilterApi.binaryColumn("decimal1"),
expectedBinary),
+ decimalSchema(
+ "decimal1", PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, 4,
precision, scale),
+ builder.equal(0, value),
+ FilterApi.eq(
+ FilterApi.binaryColumn("decimal1"),
+ Binary.fromConstantByteArray(new byte[] {0, 0, 0x04,
(byte) 0xD2})),
true);
}
@@ -367,6 +750,8 @@ class ParquetFiltersTest {
public void testInFilterDecimal32Bit() {
int precision = 9;
int scale = 2;
+ MessageType schema =
+ decimalSchema("decimal1", PrimitiveTypeName.INT32, 0,
precision, scale);
PredicateBuilder builder =
new PredicateBuilder(
new RowType(
@@ -381,6 +766,7 @@ class ParquetFiltersTest {
Decimal v3 = Decimal.fromBigDecimal(new BigDecimal("300.00"),
precision, scale);
test(
+ schema,
builder.in(0, Arrays.asList(v1, v2, v3)),
"or(eq(decimal1, "
+ (int) v1.toUnscaledLong()
@@ -392,6 +778,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(0, Arrays.asList(v1, v2, v3)),
"and(and(noteq(decimal1, "
+ (int) v1.toUnscaledLong()
@@ -407,6 +794,8 @@ class ParquetFiltersTest {
public void testInFilterDecimal64Bit() {
int precision = 18;
int scale = 4;
+ MessageType schema =
+ decimalSchema("decimal1", PrimitiveTypeName.INT64, 0,
precision, scale);
PredicateBuilder builder =
new PredicateBuilder(
new RowType(
@@ -421,6 +810,7 @@ class ParquetFiltersTest {
Decimal v3 = Decimal.fromBigDecimal(new
BigDecimal("30000000000.0000"), precision, scale);
test(
+ schema,
builder.in(0, Arrays.asList(v1, v2, v3)),
"or(eq(decimal1, "
+ v1.toUnscaledLong()
@@ -432,6 +822,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(0, Arrays.asList(v1, v2, v3)),
"and(and(noteq(decimal1, "
+ v1.toUnscaledLong()
@@ -447,110 +838,110 @@ class ParquetFiltersTest {
public void testTimestampMillis() {
// precision <= 3 uses milliseconds (INT64)
int precision = 3;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "ts1", new
TimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
TimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp value = Timestamp.fromEpochMillis(1704067200000L); //
2024-01-01 00:00:00
long expectedMillis = value.getMillisecond();
- test(builder.isNull(0), "eq(ts1, null)", true);
- test(builder.isNotNull(0), "noteq(ts1, null)", true);
- test(builder.equal(0, value), "eq(ts1, " + expectedMillis + ")", true);
- test(builder.notEqual(0, value), "noteq(ts1, " + expectedMillis + ")",
true);
- test(builder.lessThan(0, value), "lt(ts1, " + expectedMillis + ")",
true);
- test(builder.lessOrEqual(0, value), "lteq(ts1, " + expectedMillis +
")", true);
- test(builder.greaterThan(0, value), "gt(ts1, " + expectedMillis + ")",
true);
- test(builder.greaterOrEqual(0, value), "gteq(ts1, " + expectedMillis +
")", true);
+ test(schema, builder.isNull(0), "eq(ts1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(ts1, null)", true);
+ test(schema, builder.equal(0, value), "eq(ts1, " + expectedMillis +
")", true);
+ test(schema, builder.notEqual(0, value), "noteq(ts1, " +
expectedMillis + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(ts1, " + expectedMillis +
")", true);
+ test(schema, builder.lessOrEqual(0, value), "lteq(ts1, " +
expectedMillis + ")", true);
+ test(schema, builder.greaterThan(0, value), "gt(ts1, " +
expectedMillis + ")", true);
+ test(schema, builder.greaterOrEqual(0, value), "gteq(ts1, " +
expectedMillis + ")", true);
}
@Test
public void testTimestampMicros() {
// 3 < precision <= 6 uses microseconds (INT64)
int precision = 6;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "ts1", new
TimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
TimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp value = Timestamp.fromEpochMillis(1704067200123L, 456000);
// with nanos
long expectedMicros = value.getMillisecond() * 1000 +
value.getNanoOfMillisecond() / 1000;
- test(builder.isNull(0), "eq(ts1, null)", true);
- test(builder.isNotNull(0), "noteq(ts1, null)", true);
- test(builder.equal(0, value), "eq(ts1, " + expectedMicros + ")", true);
- test(builder.notEqual(0, value), "noteq(ts1, " + expectedMicros + ")",
true);
- test(builder.lessThan(0, value), "lt(ts1, " + expectedMicros + ")",
true);
- test(builder.lessOrEqual(0, value), "lteq(ts1, " + expectedMicros +
")", true);
- test(builder.greaterThan(0, value), "gt(ts1, " + expectedMicros + ")",
true);
- test(builder.greaterOrEqual(0, value), "gteq(ts1, " + expectedMicros +
")", true);
+ test(schema, builder.isNull(0), "eq(ts1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(ts1, null)", true);
+ test(schema, builder.equal(0, value), "eq(ts1, " + expectedMicros +
")", true);
+ test(schema, builder.notEqual(0, value), "noteq(ts1, " +
expectedMicros + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(ts1, " + expectedMicros +
")", true);
+ test(schema, builder.lessOrEqual(0, value), "lteq(ts1, " +
expectedMicros + ")", true);
+ test(schema, builder.greaterThan(0, value), "gt(ts1, " +
expectedMicros + ")", true);
+ test(schema, builder.greaterOrEqual(0, value), "gteq(ts1, " +
expectedMicros + ")", true);
}
@Test
public void testLocalZonedTimestampMillis() {
// precision <= 3 uses milliseconds (INT64)
int precision = 3;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(
- 0,
- "ts1",
- new
LocalZonedTimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
LocalZonedTimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp value = Timestamp.fromEpochMillis(1704067200000L);
long expectedMillis = value.getMillisecond();
- test(builder.isNull(0), "eq(ts1, null)", true);
- test(builder.isNotNull(0), "noteq(ts1, null)", true);
- test(builder.equal(0, value), "eq(ts1, " + expectedMillis + ")", true);
- test(builder.notEqual(0, value), "noteq(ts1, " + expectedMillis + ")",
true);
- test(builder.lessThan(0, value), "lt(ts1, " + expectedMillis + ")",
true);
- test(builder.greaterThan(0, value), "gt(ts1, " + expectedMillis + ")",
true);
+ test(schema, builder.isNull(0), "eq(ts1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(ts1, null)", true);
+ test(schema, builder.equal(0, value), "eq(ts1, " + expectedMillis +
")", true);
+ test(schema, builder.notEqual(0, value), "noteq(ts1, " +
expectedMillis + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(ts1, " + expectedMillis +
")", true);
+ test(schema, builder.greaterThan(0, value), "gt(ts1, " +
expectedMillis + ")", true);
}
@Test
public void testLocalZonedTimestampMicros() {
// 3 < precision <= 6 uses microseconds (INT64)
int precision = 6;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(
- 0,
- "ts1",
- new
LocalZonedTimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
LocalZonedTimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp value = Timestamp.fromEpochMillis(1704067200123L, 456000);
long expectedMicros = value.getMillisecond() * 1000 +
value.getNanoOfMillisecond() / 1000;
- test(builder.isNull(0), "eq(ts1, null)", true);
- test(builder.isNotNull(0), "noteq(ts1, null)", true);
- test(builder.equal(0, value), "eq(ts1, " + expectedMicros + ")", true);
- test(builder.notEqual(0, value), "noteq(ts1, " + expectedMicros + ")",
true);
- test(builder.lessThan(0, value), "lt(ts1, " + expectedMicros + ")",
true);
- test(builder.greaterThan(0, value), "gt(ts1, " + expectedMicros + ")",
true);
+ test(schema, builder.isNull(0), "eq(ts1, null)", true);
+ test(schema, builder.isNotNull(0), "noteq(ts1, null)", true);
+ test(schema, builder.equal(0, value), "eq(ts1, " + expectedMicros +
")", true);
+ test(schema, builder.notEqual(0, value), "noteq(ts1, " +
expectedMicros + ")", true);
+ test(schema, builder.lessThan(0, value), "lt(ts1, " + expectedMicros +
")", true);
+ test(schema, builder.greaterThan(0, value), "gt(ts1, " +
expectedMicros + ")", true);
}
@Test
public void testInFilterTimestampMillis() {
int precision = 3;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "ts1", new
TimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
TimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp v1 = Timestamp.fromEpochMillis(1704067200000L);
Timestamp v2 = Timestamp.fromEpochMillis(1704153600000L);
Timestamp v3 = Timestamp.fromEpochMillis(1704240000000L);
test(
+ schema,
builder.in(0, Arrays.asList(v1, v2, v3)),
"or(eq(ts1, "
+ v1.getMillisecond()
@@ -562,6 +953,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(0, Arrays.asList(v1, v2, v3)),
"and(and(noteq(ts1, "
+ v1.getMillisecond()
@@ -576,11 +968,12 @@ class ParquetFiltersTest {
@Test
public void testInFilterTimestampMicros() {
int precision = 6;
- PredicateBuilder builder =
- new PredicateBuilder(
- new RowType(
- Collections.singletonList(
- new DataField(0, "ts1", new
TimestampType(precision)))));
+ RowType rowType =
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "ts1", new
TimestampType(precision))));
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
Timestamp v1 = Timestamp.fromEpochMillis(1704067200000L, 123000);
Timestamp v2 = Timestamp.fromEpochMillis(1704153600000L, 456000);
@@ -591,6 +984,7 @@ class ParquetFiltersTest {
long micros3 = v3.getMillisecond() * 1000 + v3.getNanoOfMillisecond()
/ 1000;
test(
+ schema,
builder.in(0, Arrays.asList(v1, v2, v3)),
"or(eq(ts1, "
+ micros1
@@ -602,6 +996,7 @@ class ParquetFiltersTest {
true);
test(
+ schema,
builder.notIn(0, Arrays.asList(v1, v2, v3)),
"and(and(noteq(ts1, "
+ micros1
@@ -613,8 +1008,13 @@ class ParquetFiltersTest {
true);
}
- private void test(Predicate predicate, FilterPredicate parquetPredicate,
boolean canPushDown) {
- FilterCompat.Filter filter =
ParquetFilters.convert(PredicateBuilder.splitAnd(predicate));
+ private void test(
+ MessageType schema,
+ Predicate predicate,
+ FilterPredicate parquetPredicate,
+ boolean canPushDown) {
+ FilterCompat.Filter filter =
+ ParquetFilters.convert(PredicateBuilder.splitAnd(predicate),
schema, true);
if (canPushDown) {
FilterPredicateCompat compat = (FilterPredicateCompat) filter;
assertThat(compat.getFilterPredicate()).isEqualTo(parquetPredicate);
@@ -623,8 +1023,16 @@ class ParquetFiltersTest {
}
}
- private void test(Predicate predicate, String expected, boolean
canPushDown) {
- FilterCompat.Filter filter =
ParquetFilters.convert(PredicateBuilder.splitAnd(predicate));
+ private FilterPredicate convert(MessageType schema, Predicate predicate) {
+ FilterCompat.Filter filter =
+ ParquetFilters.convert(PredicateBuilder.splitAnd(predicate),
schema, true);
+ return ((FilterPredicateCompat) filter).getFilterPredicate();
+ }
+
+ private void test(
+ MessageType schema, Predicate predicate, String expected, boolean
canPushDown) {
+ FilterCompat.Filter filter =
+ ParquetFilters.convert(PredicateBuilder.splitAnd(predicate),
schema, true);
if (canPushDown) {
FilterPredicateCompat compat = (FilterPredicateCompat) filter;
assertThat(compat.getFilterPredicate().toString()).isEqualTo(expected);
@@ -632,4 +1040,26 @@ class ParquetFiltersTest {
assertThat(filter).isEqualTo(FilterCompat.NOOP);
}
}
+
+ private static PredicateBuilder decimalPredicateBuilder(int precision, int
scale) {
+ return new PredicateBuilder(
+ new RowType(
+ Collections.singletonList(
+ new DataField(0, "decimal1", new
DecimalType(precision, scale)))));
+ }
+
+ private static MessageType decimalSchema(
+ String fieldName,
+ PrimitiveTypeName physicalType,
+ int fixedLength,
+ int precision,
+ int scale) {
+ Types.PrimitiveBuilder<PrimitiveType> builder =
Types.required(physicalType);
+ if (physicalType == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) {
+ builder.length(fixedLength);
+ }
+ PrimitiveType type =
+ builder.as(LogicalTypeAnnotation.decimalType(scale,
precision)).named(fieldName);
+ return new MessageType("paimon_schema",
Collections.singletonList(type));
+ }
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
index ef18dd0c29..f585a0cfdd 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
@@ -60,8 +60,6 @@ import org.apache.paimon.types.VarCharType;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
-import org.apache.parquet.filter2.compat.FilterCompat;
-import org.apache.parquet.filter2.predicate.ParquetFilters;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
@@ -309,7 +307,7 @@ public class ParquetReadWriteTest {
.fields(fieldTypes, new String[] {"f7", "f2",
"f4"})
.build(),
500,
- FilterCompat.NOOP);
+ null);
AtomicInteger cnt = new AtomicInteger(0);
RecordReader<InternalRow> reader =
@@ -353,7 +351,7 @@ public class ParquetReadWriteTest {
.fields(fieldTypes, new String[] {"f7", "f2",
"f4", "f99"})
.build(),
500,
- FilterCompat.NOOP);
+ null);
AtomicInteger cnt = new AtomicInteger(0);
RecordReader<InternalRow> reader =
@@ -392,7 +390,7 @@ public class ParquetReadWriteTest {
new Options(),
RowType.builder().fields(fieldTypes, new String[]
{"f7"}).build(),
batchSize,
- FilterCompat.NOOP);
+ null);
AtomicInteger cnt = new AtomicInteger(0);
try (RecordReader<InternalRow> reader =
@@ -434,15 +432,12 @@ public class ParquetReadWriteTest {
new PredicateBuilder(
new RowType(
Collections.singletonList(new DataField(0,
"f4", new IntType()))));
- FilterCompat.Filter filter =
- ParquetFilters.convert(
- PredicateBuilder.splitAnd(builder.greaterThan(0,
randomStart)));
ParquetReaderFactory format =
new ParquetReaderFactory(
new Options(),
RowType.builder().fields(fieldTypes, new String[]
{"f4"}).build(),
batchSize,
- filter);
+ PredicateBuilder.splitAnd(builder.greaterThan(0,
randomStart)));
AtomicBoolean isFirst = new AtomicBoolean(true);
try (RecordReader<InternalRow> reader =
@@ -480,8 +475,7 @@ public class ParquetReadWriteTest {
throw new RuntimeException("Unknown writer type.");
}
ParquetReaderFactory format =
- new ParquetReaderFactory(
- new Options(), NESTED_ARRAY_MAP_TYPE, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), NESTED_ARRAY_MAP_TYPE,
500, null);
RecordReader<InternalRow> reader =
format.createReader(
new FormatReaderContext(
@@ -498,8 +492,14 @@ public class ParquetReadWriteTest {
int number = new Random().nextInt(1000) + 100;
Path path = createDecimalFile(number, folder, 10);
+ PredicateBuilder builder = new PredicateBuilder(DECIMAL_TYPE);
+ Decimal filterValue = Decimal.fromBigDecimal(new
BigDecimal("1234567.67"), 9, 2);
ParquetReaderFactory format =
- new ParquetReaderFactory(new Options(), DECIMAL_TYPE, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(
+ new Options(),
+ DECIMAL_TYPE,
+ 500,
+ Collections.singletonList(builder.equal(2,
filterValue)));
RecordReader<InternalRow> reader =
format.createReader(
new FormatReaderContext(
@@ -507,6 +507,7 @@ public class ParquetReadWriteTest {
List<InternalRow> results = new ArrayList<>(number);
InternalRowSerializer internalRowSerializer = new
InternalRowSerializer(DECIMAL_TYPE);
reader.forEachRemaining(row ->
results.add(internalRowSerializer.copy(row)));
+ assertThat(results).hasSize(number);
BigDecimal decimalValue0 = new BigDecimal("123.67");
BigDecimal decimalValue1 = new BigDecimal("12345.67");
@@ -659,7 +660,7 @@ public class ParquetReadWriteTest {
.build();
ParquetReaderFactory format =
- new ParquetReaderFactory(new Options(), paimonRowType, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), paimonRowType, 500,
null);
RecordReader<InternalRow> reader =
format.createReader(
@@ -723,7 +724,7 @@ public class ParquetReadWriteTest {
.fields(new TimestampType(9), new ArrayType(new
TimestampType(9)))
.build();
ParquetReaderFactory format =
- new ParquetReaderFactory(new Options(), paimonRowType, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), paimonRowType, 500,
null);
AtomicInteger count = new AtomicInteger(0);
try (RecordReader<InternalRow> reader =
format.createReader(
@@ -755,8 +756,7 @@ public class ParquetReadWriteTest {
GenericRow.of(2, BinaryVector.fromPrimitiveArray(new
float[] {4, 5, 6})));
Path path = createTempParquetFileByPaimon(folder, rows, 1024, rowType);
- ParquetReaderFactory format =
- new ParquetReaderFactory(new Options(), rowType, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory format = new ParquetReaderFactory(new Options(),
rowType, 500, null);
RecordReader<InternalRow> reader =
format.createReader(
@@ -826,8 +826,7 @@ public class ParquetReadWriteTest {
}
private int testReadingFile(List<Integer> expected, Path path) throws
IOException {
- ParquetReaderFactory format =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory format = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
RecordReader<InternalRow> reader =
format.createReader(
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetSchemaCacheTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetSchemaCacheTest.java
index 146391021a..5a787cff90 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetSchemaCacheTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetSchemaCacheTest.java
@@ -30,7 +30,6 @@ import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.parquet.filter2.compat.FilterCompat;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -61,8 +60,7 @@ public class ParquetSchemaCacheTest {
@Test
void testCacheIsEmptyBeforeFirstRead() throws Exception {
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
assertThat(getSchemaCacheSize(factory)).isEqualTo(0);
}
@@ -71,8 +69,7 @@ public class ParquetSchemaCacheTest {
void testCacheIsPopulatedAfterFirstRead() throws Exception {
Path path = writeSingleFile();
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
readAll(factory, path);
@@ -89,8 +86,7 @@ public class ParquetSchemaCacheTest {
Path path2 = writeSingleFile();
Path path3 = writeSingleFile();
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
readAll(factory, path1);
readAll(factory, path2);
@@ -108,8 +104,7 @@ public class ParquetSchemaCacheTest {
void testDataReadCorrectlyOnFirstRead() throws Exception {
Path path = writeSingleFile();
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
assertThat(countRows(factory, path)).isEqualTo(3);
}
@@ -118,8 +113,7 @@ public class ParquetSchemaCacheTest {
void testDataReadCorrectlyOnSubsequentCachedReads() throws Exception {
Path path = writeSingleFile();
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
assertThat(countRows(factory, path)).isEqualTo(3);
assertThat(countRows(factory, path)).isEqualTo(3);
@@ -131,8 +125,7 @@ public class ParquetSchemaCacheTest {
Path path1 = writeSingleFile();
Path path2 = writeSingleFile();
- ParquetReaderFactory factory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ ParquetReaderFactory factory = new ParquetReaderFactory(new Options(),
ROW_TYPE, 500, null);
assertThat(countRows(factory, path1)).isEqualTo(3);
assertThat(countRows(factory, path2)).isEqualTo(3);
@@ -147,9 +140,9 @@ public class ParquetSchemaCacheTest {
Path path = writeSingleFile();
ParquetReaderFactory factory1 =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), ROW_TYPE, 500, null);
ParquetReaderFactory factory2 =
- new ParquetReaderFactory(new Options(), ROW_TYPE, 500,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), ROW_TYPE, 500, null);
readAll(factory1, path);
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetInt96RowSelectionTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetInt96RowSelectionTest.java
index d8fd08adad..2c998dbb25 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetInt96RowSelectionTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetInt96RowSelectionTest.java
@@ -37,7 +37,6 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.RoaringBitmap32;
-import org.apache.parquet.filter2.compat.FilterCompat;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -78,7 +77,7 @@ public class ParquetInt96RowSelectionTest {
fileIO.getFileSize(file),
RoaringBitmap32.bitmapOf(SELECTED_POSITIONS));
ParquetReaderFactory readerFactory =
- new ParquetReaderFactory(new Options(), ROW_TYPE, BATCH_SIZE,
FilterCompat.NOOP);
+ new ParquetReaderFactory(new Options(), ROW_TYPE, BATCH_SIZE,
null);
try (RecordReader<InternalRow> reader =
readerFactory.createReader(context)) {
int expectedIndex = 0;
