This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 29b7a52564f4 fix(trino): skip predicate pushdown on type-evolved
parquet columns (#19467)
29b7a52564f4 is described below
commit 29b7a52564f4a3fe3979994c97605098733d6997
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Aug 26 15:18:50 2026 +0700
fix(trino): skip predicate pushdown on type-evolved parquet columns (#19467)
* fix(trino): skip predicate pushdown on type-evolved parquet columns
* test(trino): remove the dead INT96 branch in the parquet statistics test
helper
* fix(trino): drop domains the parquet bloom filter and decimal rescale
would misread
* test(trino): name the test that actually pins the bloom-width mechanism
The type-table comment on the int -> long / long -> int rows pointed at
testABigintLookupNeverFindsAnInt32ColumnsBloomHashes, which does not exist.
The test that pins the bloom miss on a real read is
TestHudiEvolvedColumnPredicates.testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows,
which is what the class Javadoc and ParquetStatisticsDomains already name.
---------
Co-authored-by: Vova Kolmakov <[email protected]>
Co-authored-by: voon <[email protected]>
---
.../trino/plugin/hudi/HudiPageSourceProvider.java | 6 +-
.../plugin/hudi/util/ParquetStatisticsDomains.java | 207 +++++++++++
.../TestHudiConnectorParquetColumnNamesTest.java | 7 +-
.../hudi/TestHudiEvolvedColumnPredicates.java | 387 ++++++++++++++++++++
.../hudi/TestHudiPageSourceProviderTest.java | 144 +-------
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 98 ++++--
.../plugin/hudi/TestingBaseFilePageSource.java | 201 +++++++++++
.../SchemaEvolutionHudiTablesInitializer.java | 174 +++++++++
.../hudi/util/TestParquetStatisticsDomains.java | 390 +++++++++++++++++++++
9 files changed, 1453 insertions(+), 161 deletions(-)
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
index 1d58eb7adc1c..aafc7ea1c5ff 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
@@ -114,6 +114,7 @@ import static
io.trino.plugin.hudi.HudiUtil.getLatestTableSchema;
import static
io.trino.plugin.hudi.HudiUtil.prependHudiMetaAndMergeRequiredColumns;
import static io.trino.plugin.hudi.HudiUtil.resolveMergeModeAndStrategyId;
import static io.trino.plugin.hudi.HudiUtil.usesNonProjectionCompatibleMerger;
+import static
io.trino.plugin.hudi.util.ParquetStatisticsDomains.dropIncomparableDomains;
import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@@ -409,9 +410,12 @@ public class HudiPageSourceProvider
Map<List<String>, ColumnDescriptor> descriptorsByPath =
getDescriptors(fileSchema, requestedSchema);
+ // A domain typed by the metastore cannot be matched against the
statistics of a column the file stores
+ // under the type it had before a schema evolution, so those are
dropped before the parquet predicate
+ // ever sees them. See ParquetStatisticsDomains.
TupleDomain<ColumnDescriptor> parquetTupleDomain =
options.isIgnoreStatistics() || !enablePredicatePushDown
? TupleDomain.all()
- : getParquetTupleDomain(descriptorsByPath,
getPushdownPredicate(hudiSplit, dynamicFilter, physicalIndexMap), fileSchema,
useColumnNames);
+ :
dropIncomparableDomains(getParquetTupleDomain(descriptorsByPath,
getPushdownPredicate(hudiSplit, dynamicFilter, physicalIndexMap), fileSchema,
useColumnNames));
TupleDomainParquetPredicate parquetPredicate =
buildPredicate(requestedSchema, parquetTupleDomain, descriptorsByPath,
timeZone);
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java
new file mode 100644
index 000000000000..00b16562d504
--- /dev/null
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java
@@ -0,0 +1,207 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import io.airlift.log.Logger;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.type.CharType;
+import io.trino.spi.type.DecimalType;
+import io.trino.spi.type.TimestampType;
+import io.trino.spi.type.Type;
+import io.trino.spi.type.UuidType;
+import io.trino.spi.type.VarbinaryType;
+import io.trino.spi.type.VarcharType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import
org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+import
org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static io.trino.spi.type.DateType.DATE;
+import static io.trino.spi.type.DoubleType.DOUBLE;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.RealType.REAL;
+import static io.trino.spi.type.SmallintType.SMALLINT;
+import static io.trino.spi.type.TinyintType.TINYINT;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
+import static
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96;
+
+/**
+ * Keeps a pushed-down predicate from being matched against statistics it
cannot be compared with.
+ * <p>
+ * {@code TupleDomainParquetPredicate.getDomain} selects its branch on the
type of the pushed-down DOMAIN and then
+ * reads the parquet statistics as that type - {@code Double min = (Double)
minimums.get(i)} and so on. The domain's
+ * type comes from the metastore while the statistics come from the file, and
Hudi's type evolution is exactly what
+ * makes those two disagree: after a column evolves and the metastore is
synced, every base file written before the
+ * evolution still stores the old physical type. Handing such a domain to the
parquet predicate either fails the
+ * whole split with {@code Malformed Parquet file. Corrupted statistics for
column ...} wrapping a
+ * {@link ClassCastException}, or - where the two types happen to share a
representation, as a decimal and a varchar
+ * both do through {@code Slice} - silently prunes row groups on a comparison
that means nothing.
+ * <p>
+ * The read path has no such problem: {@code ColumnReaderFactory} decodes
parquet {@code FLOAT} into Trino
+ * {@code DOUBLE} and {@code INT32} into {@code BIGINT} natively, and {@code
ParquetTypeTranslator.createCoercer}
+ * covers the rest of the promotions the page source is asked for. Only the
statistics side is blind, so only the
+ * statistics side needs the guard.
+ * <p>
+ * This is a stopgap for the Trino the module is pinned to.
trinodb/trino#30545 moves the same check down into
+ * {@code lib/trino-parquet}, where it can reconcile the two types rather than
drop the domain - so it keeps
+ * {@code float -> double} pruning - and it covers the bloom filter path too.
Once {@code trino.sha} in the root pom
+ * passes that commit, delete this class and its test instead of maintaining
the two side by side.
+ */
+public final class ParquetStatisticsDomains
+{
+ private static final Logger log =
Logger.get(ParquetStatisticsDomains.class);
+
+ private ParquetStatisticsDomains() {}
+
+ /**
+ * Drops every domain whose type cannot be compared against its column's
statistics, keeping the rest untouched.
+ * <p>
+ * Dropping loses row group pruning for that column but never a row:
{@code HudiMetadata.applyFilter} hands the
+ * whole regular predicate back to the engine as the remaining filter and
does not precalculate statistics for
+ * the pushdown, so a connector-side domain is an optimization and nothing
else. The dynamic half of the
+ * predicate is redundant with the join above the scan by construction. It
is the same trade
+ * {@code HudiPageSourceProvider.remapPredicateColumnIndicesToPhysical}
already makes for a predicate column the
+ * file does not carry, and the same one {@code
HudiColumnStatsIndexSupport.getDomainFromColumnStats} makes when
+ * the metadata table's column statistics do not match the column's type.
+ * <p>
+ * The filter runs on the descriptor-keyed domain rather than on the
column handles it was built from. That is
+ * what the parquet predicate itself will be evaluated against, so the
check and the evaluation cannot disagree
+ * about which column or which physical type is meant; it covers both
values of
+ * {@code hudi.parquet.use-column-names} in one pass, since a handle is
resolved to a descriptor before either;
+ * and a dereference handle contributes the leaf field's type without any
extra work.
+ */
+ public static TupleDomain<ColumnDescriptor>
dropIncomparableDomains(TupleDomain<ColumnDescriptor> parquetTupleDomain)
+ {
+ if (parquetTupleDomain.isAll() || parquetTupleDomain.isNone()) {
+ return parquetTupleDomain;
+ }
+
+ Map<ColumnDescriptor, Domain> domains =
parquetTupleDomain.getDomains().orElseThrow();
+ Map<ColumnDescriptor, Domain> comparableDomains = new
LinkedHashMap<>();
+ for (Map.Entry<ColumnDescriptor, Domain> entry : domains.entrySet()) {
+ if (hasComparableStatistics(entry.getValue().getType(),
entry.getKey().getPrimitiveType())) {
+ comparableDomains.put(entry.getKey(), entry.getValue());
+ }
+ else {
+ log.debug("Not pushing down a %s predicate on %s: the file
stores it as %s, so the column statistics cannot answer it",
+ entry.getValue().getType(), entry.getKey(),
entry.getKey().getPrimitiveType());
+ }
+ }
+ if (comparableDomains.size() == domains.size()) {
+ return parquetTupleDomain;
+ }
+ return TupleDomain.withColumnDomains(comparableDomains);
+ }
+
+ /**
+ * Whether the parquet predicate can read a {@code fileType} column as
{@code domainType} without misreading it,
+ * which is the case only when the two describe the same physical values.
+ * <p>
+ * Two dispatch tables have to agree here, not one. {@code
TupleDomainParquetPredicate.getDomain} picks its branch
+ * on the domain type and reads the min/max statistics as that type;
{@code checkInBloomFilter} picks a branch of
+ * its own and hashes the looked-up value at the DOMAIN's width, while
parquet-mr hashed the column at the FILE's.
+ * A pair the first reads happily can therefore still make the second miss
every lookup, and a bloom miss drops
+ * the row group outright - that loses rows rather than merely failing to
prune them. A pair is kept only when
+ * each of the two either reads the file's own representation or declines
to use it.
+ * <p>
+ * Dropping is not free either, so nothing is dropped without cause.
{@code getDomain} answers a null-count
+ * predicate before it dispatches on type at all - all-null statistics
become {@code Domain.onlyNull} and a zero
+ * null count becomes a not-null domain - so a dropped domain also
forfeits {@code IS NULL} / {@code IS NOT NULL}
+ * pruning, and a dropped {@code VARBINARY} or {@code UUID} domain
forfeits bloom pruning that would have worked.
+ * That is why the types {@code getDomain} has no branch for are kept
rather than rejected: its fallthrough
+ * returns a domain covering every value, which is safe whatever the file
holds, and the null count and the bloom
+ * filter still pay for themselves.
+ * <p>
+ * {@code TestParquetStatisticsDomains} pins every pair below against the
real {@code getDomain}. The bloom half
+ * leaves no domain to inspect, so it is pinned where the harm shows up
instead, on the rows a real read returns:
+ * {@code
TestHudiEvolvedColumnPredicates.testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows}.
+ */
+ public static boolean hasComparableStatistics(Type domainType,
PrimitiveType fileType)
+ {
+ PrimitiveTypeName primitiveType = fileType.getPrimitiveTypeName();
+ LogicalTypeAnnotation annotation = fileType.getLogicalTypeAnnotation();
+
+ if (BOOLEAN.equals(domainType)) {
+ return primitiveType == PrimitiveTypeName.BOOLEAN;
+ }
+ if (TINYINT.equals(domainType) || SMALLINT.equals(domainType) ||
INTEGER.equals(domainType)
+ || BIGINT.equals(domainType) || DATE.equals(domainType)) {
+ // asLong takes any integral box the statistics can hold, so the
min/max side would read INT32 and INT64
+ // interchangeably. checkInBloomFilter would not: it hashes a
bigint lookup at eight bytes and every
+ // narrower integral one at four, while parquet-mr hashed the
column at its own width. A bigint over INT32
+ // then finds nothing in the filter and the row group is dropped,
so require the same width.
+ // A decimal column reports its UNSCALED value though, which is
the integer it stands for only at scale 0.
+ return (BIGINT.equals(domainType) ? primitiveType == INT64 :
primitiveType == INT32) && !isScaledDecimal(annotation);
+ }
+ if (domainType instanceof DecimalType domainDecimal) {
+ // getShortDecimal and getLongDecimal rescale against the column's
own annotation. Without one they read
+ // the raw value as an unscaled decimal at the DOMAIN's scale, so
an int or a string that evolved into a
+ // decimal would be compared a factor of ten-to-the-scale off, or
as raw UTF-8 bytes. The rescale itself
+ // throws NUMERIC_VALUE_OUT_OF_RANGE once a file bound no longer
fits the domain at the new scale, which
+ // the predicate reports as the very corrupt-statistics failure
this class exists to prevent. So keep
+ // exactly Hudi's own lossless widening, which cannot overflow:
neither the integer digits nor the scale
+ // may shrink.
+ return isDecimalPrimitive(primitiveType) && annotation instanceof
DecimalLogicalTypeAnnotation fileDecimal
+ && fileDecimal.getPrecision() - fileDecimal.getScale() <=
domainDecimal.getPrecision() - domainDecimal.getScale()
+ && fileDecimal.getScale() <= domainDecimal.getScale();
+ }
+ if (REAL.equals(domainType)) {
+ return primitiveType == FLOAT;
+ }
+ if (DOUBLE.equals(domainType)) {
+ return primitiveType == PrimitiveTypeName.DOUBLE;
+ }
+ if (domainType instanceof VarcharType) {
+ // Both sides compare raw bytes, which is varchar's own ordering -
unless the bytes are a decimal's
+ // big-endian two's complement, which orders nothing like the
digits it prints as.
+ return (primitiveType == BINARY || primitiveType ==
FIXED_LEN_BYTE_ARRAY)
+ && !(annotation instanceof DecimalLogicalTypeAnnotation);
+ }
+ if (domainType instanceof CharType || domainType instanceof
VarbinaryType || domainType instanceof UuidType) {
+ // getDomain has no branch for any of these, so it falls through
to a domain covering every value and only
+ // the null count prunes. checkInBloomFilter does have one for
varbinary and uuid, hashing the lookup as
+ // raw bytes, so the column still has to be one parquet-mr hashed
as bytes.
+ return primitiveType == BINARY || primitiveType ==
FIXED_LEN_BYTE_ARRAY;
+ }
+ if (domainType instanceof TimestampType) {
+ // INT96 is read from the binary statistics and INT64 from the
long ones, but an INT64 column has to say
+ // which unit it counts in before its bounds mean anything.
+ return primitiveType == INT96
+ || (primitiveType == INT64 && annotation instanceof
TimestampLogicalTypeAnnotation timestampAnnotation &&
timestampAnnotation.getUnit() != null);
+ }
+ return false;
+ }
+
+ private static boolean isDecimalPrimitive(PrimitiveTypeName primitiveType)
+ {
+ return primitiveType == INT32 || primitiveType == INT64 ||
primitiveType == BINARY || primitiveType == FIXED_LEN_BYTE_ARRAY;
+ }
+
+ private static boolean isScaledDecimal(LogicalTypeAnnotation annotation)
+ {
+ return annotation instanceof DecimalLogicalTypeAnnotation
decimalAnnotation && decimalAnnotation.getScale() != 0;
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
index fda63c8b4714..9b3ebc3836bc 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
@@ -16,6 +16,7 @@ package io.trino.plugin.hudi;
import io.trino.plugin.hudi.testing.CompositeHudiTablesInitializer;
import io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer;
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer;
import io.trino.testing.QueryRunner;
import org.junit.jupiter.api.Test;
@@ -35,10 +36,12 @@ public class TestHudiConnectorParquetColumnNamesTest
.addConnectorProperty("hudi.parquet.use-column-names", "false")
// The resource tables all register the Hudi meta columns in
the metastore, so their metastore
// ordinals already equal their physical ones and nothing here
resolves a stale ordinal. The
- // second fixture is the one whose metastore omits them.
+ // second fixture is the one whose metastore omits them, and
the third is the type-evolution table
+ // TestHudiSmokeTest reads, carried here so its predicates are
resolved positionally too.
.setDataLoader(new CompositeHudiTablesInitializer(
new ResourceHudiTablesInitializer(),
- new OmittedMetaColumnsHudiTablesInitializer()))
+ new OmittedMetaColumnsHudiTablesInitializer(),
+ new SchemaEvolutionHudiTablesInitializer()))
.build();
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java
new file mode 100644
index 000000000000..e609479092e7
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java
@@ -0,0 +1,387 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.airlift.slice.Slices;
+import io.trino.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.Range;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.predicate.ValueSet;
+import io.trino.spi.type.Type;
+import io.trino.testing.MaterializedResult;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Types;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.TestingBaseFilePageSource.dynamicFilterOn;
+import static io.trino.plugin.hudi.TestingBaseFilePageSource.read;
+import static io.trino.plugin.hudi.TestingBaseFilePageSource.writeBaseFile;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.DoubleType.DOUBLE;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Covers what a pushed-down predicate does to a base file written before the
column it constrains was evolved.
+ * <p>
+ * Hudi lets a column's type widen and hive sync then reports the NEW type,
while every base file written before the
+ * evolution keeps storing the old one. The parquet reader copes with that on
its own -- {@code ColumnReaderFactory}
+ * decodes {@code FLOAT} into {@code DOUBLE} and {@code INT32} into {@code
BIGINT}, and
+ * {@code ParquetTypeTranslator.createCoercer} handles the rest -- but the
statistics do not: {@code
+ * TupleDomainParquetPredicate.getDomain} reads them as whatever the DOMAIN's
type says, so a {@code DOUBLE} domain
+ * over a {@code FLOAT} column casts a {@code Float} to a {@code Double} and
fails the whole split with {@code
+ * HUDI_BAD_DATA}. See apache/hudi#19457.
+ * <p>
+ * Two files, because that is the state one insert after an evolution leaves a
table in: {@link #preEvolutionFile}
+ * stores every data column as the type it had BEFORE, {@link
#postEvolutionFile} as the type it has after, and the
+ * handles carry the metastore's post-evolution types for both. A split over
the first must drop its domain; a split
+ * over the second must keep it and go on pruning, which is what stops the
guard from being written as "give up on
+ * this column".
+ * <p>
+ * Values grow with the row index so pruning stays observable: a predicate
that survives pushdown reads fewer rows
+ * than the file holds, and one that was dropped reads all of them. Do not
"simplify" that to asserting the matching
+ * rows alone -- both a working pushdown and no pushdown at all produce the
same matching rows, since the connector's
+ * pushdown is an optimization and the engine re-applies the predicate above
the scan.
+ */
+class TestHudiEvolvedColumnPredicates
+{
+ private static final String STABLE_COLUMN = "stable_int";
+ /** Written as parquet FLOAT before the evolution, DOUBLE after, reported
by the metastore as double. */
+ private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double";
+ /** Written as parquet INT32 before the evolution, INT64 after, reported
by the metastore as bigint. */
+ private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint";
+ /** Written as parquet INT32 before the evolution, a BINARY string after,
reported by the metastore as string. */
+ private static final String INT_TO_VARCHAR_COLUMN = "evolved_varchar";
+
+ private static final int ROW_COUNT = 1000;
+ private static final long THRESHOLD = 900;
+ private static final int MATCHING_ROW_COUNT = (int) (ROW_COUNT - THRESHOLD
- 1);
+ /** A value that exists in both files, for the equality predicates that
reach the bloom filter. */
+ private static final long PRESENT_VALUE = 500;
+
+ @TempDir
+ static Path tempDir;
+
+ private static Path preEvolutionFile;
+ private static Path postEvolutionFile;
+
+ @BeforeAll
+ static void writeBaseFiles()
+ throws IOException
+ {
+ preEvolutionFile = tempDir.resolve("pre_evolution_base_file.parquet");
+ postEvolutionFile =
tempDir.resolve("post_evolution_base_file.parquet");
+
+ // Bloom filters on the two integral columns, as a Hudi writer builds
for every column set in
+ // parquet.bloom.filter.enabled#<column>. They are what makes an
equality predicate on an evolved column
+ // dangerous rather than merely unhelpful: see
testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows.
+ List<String> bloomFilterColumns = List.of(STABLE_COLUMN,
INT_TO_BIGINT_COLUMN);
+ int preEvolutionRowGroups = writeBaseFile(preEvolutionFile,
preEvolutionFileSchema(), ROW_COUNT, bloomFilterColumns,
+ (group, row) -> {
+ appendMetaColumns(group, row);
+ group.append(STABLE_COLUMN, row);
+ group.append(FLOAT_TO_DOUBLE_COLUMN, (float) row);
+ group.append(INT_TO_BIGINT_COLUMN, row);
+ group.append(INT_TO_VARCHAR_COLUMN, row);
+ });
+ int postEvolutionRowGroups = writeBaseFile(postEvolutionFile,
postEvolutionFileSchema(), ROW_COUNT, bloomFilterColumns,
+ (group, row) -> {
+ appendMetaColumns(group, row);
+ group.append(STABLE_COLUMN, row);
+ group.append(FLOAT_TO_DOUBLE_COLUMN, (double) row);
+ group.append(INT_TO_BIGINT_COLUMN, (long) row);
+ group.append(INT_TO_VARCHAR_COLUMN, Integer.toString(row));
+ });
+
+ // With a single row group there would be nothing to prune and every
"still prunes" assertion below would
+ // hold without proving anything, so assert the outcome rather than
the writer knobs that produce it.
+ assertThat(preEvolutionRowGroups).as("row groups in the pre-evolution
file").isGreaterThan(1);
+ assertThat(postEvolutionRowGroups).as("row groups in the
post-evolution file").isGreaterThan(1);
+ }
+
+ /**
+ * The both-modes anchor. {@code column()} builds every handle on its
physical ordinal, so the two values of
+ * {@code hudi.parquet.use-column-names} hand the guard the identical
descriptor and the axis cannot discriminate
+ * anywhere in this class -- it is carried here alone, to pin that the
guard runs after the resolution fork
+ * rather than inside one branch of it. Stale ordinals are {@code
TestHudiPageSourceProviderTest}'s subject.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testPredicateOnFloatColumnEvolvedToDouble(boolean
useParquetColumnNames)
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(FLOAT_TO_DOUBLE_COLUMN,
HiveType.HIVE_DOUBLE, DOUBLE);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
greaterThanThreshold(evolved, DOUBLE, (double) THRESHOLD),
+ useParquetColumnNames, DynamicFilter.EMPTY);
+
+ // The domain cannot be matched against FLOAT statistics, so it is
dropped and nothing is pruned. Before the
+ // fix this threw HUDI_BAD_DATA ("Corrupted statistics for column")
instead of reading anything at all.
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+ // The read itself promotes, so the rows the engine will filter carry
the widened values
+
assertThat(result.getMaterializedRows().get(7).getField(1)).as("promoted value
of row 7").isEqualTo(7.0d);
+ assertThat(valuesOver(result, 1)).as("rows matching %s > %s",
FLOAT_TO_DOUBLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testPredicateOnIntColumnEvolvedToVarchar()
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(INT_TO_VARCHAR_COLUMN,
HiveType.HIVE_STRING, VARCHAR);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
+ greaterThanThreshold(evolved, VARCHAR,
Slices.utf8Slice("900")), false, DynamicFilter.EMPTY);
+
+ // A varchar domain over an INT32 column would cast an Integer to a
Slice
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+
assertThat(result.getMaterializedRows().get(7).getField(1)).as("promoted value
of row 7").isEqualTo("7");
+ }
+
+ /**
+ * apache/hudi#19457 comment thread: {@code asLong} takes an Integer as
happily as a Long, so the min/max side
+ * reads this promotion perfectly -- and the guard still has to drop it,
because {@code checkInBloomFilter} does
+ * not. See {@link
#testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows} for what that costs
when the pair
+ * is kept. Losing row-group pruning on pre-evolution files is the price;
+ * {@link #testThePostEvolutionFileStillPrunesTheSamePredicate} pins that
files written after the evolution keep
+ * it.
+ */
+ @Test
+ public void testPredicateOnIntColumnEvolvedToBigintIsDropped()
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(INT_TO_BIGINT_COLUMN,
HiveType.HIVE_LONG, BIGINT);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
greaterThanThreshold(evolved, BIGINT, THRESHOLD),
+ false, DynamicFilter.EMPTY);
+
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+ assertThat(valuesOver(result, 1)).as("rows matching %s > %s",
INT_TO_BIGINT_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ /**
+ * The reason a bigint domain over an INT32 column has to be dropped even
though its statistics read correctly.
+ * <p>
+ * {@code checkInBloomFilter} hashes the looked-up value at the DOMAIN's
width, eight bytes for a bigint, while
+ * parquet-mr hashed the INT32 column at four. The lookup therefore finds
nothing,
+ * {@code TupleDomainParquetPredicate.matches(BloomFilterStore, int)}
reports no match, and every row group is
+ * dropped with the matching row still inside it -- no error, just missing
rows. Trino reads bloom filters by
+ * default, so any Hudi table written with {@code
parquet.bloom.filter.enabled#<column>=true} is exposed.
+ */
+ @Test
+ public void testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows()
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(INT_TO_BIGINT_COLUMN,
HiveType.HIVE_LONG, BIGINT);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
+ TupleDomain.withColumnDomains(Map.of(evolved,
Domain.singleValue(BIGINT, PRESENT_VALUE))),
+ false, DynamicFilter.EMPTY);
+
+ // Dropped, so nothing is pruned -- and the row is there. With the
pair kept the bloom filter answers "not
+ // present" for every row group and this reads 0 rows.
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+ assertThat(valuesEqualTo(result, 1)).as("rows holding %s = %s",
INT_TO_BIGINT_COLUMN, PRESENT_VALUE).isEqualTo(1);
+ }
+
+ /**
+ * The same predicate against a file written AFTER the evolution, where
the column really is an INT64 and both
+ * the statistics and the bloom filter answer it. Without this the guard
could be "never push down on a column
+ * the metastore widened" and every test above would still pass, while
every table would lose pruning forever.
+ */
+ @Test
+ public void testThePostEvolutionFileStillPrunesTheSamePredicate()
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(INT_TO_BIGINT_COLUMN,
HiveType.HIVE_LONG, BIGINT);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+ TupleDomain<HiveColumnHandle> predicate =
TupleDomain.withColumnDomains(
+ Map.of(evolved, Domain.singleValue(BIGINT, PRESENT_VALUE)));
+
+ MaterializedResult result = read(postEvolutionFile, projection,
predicate, false, DynamicFilter.EMPTY);
+
+ assertThat(result.getRowCount()).as("rows read out of %s",
ROW_COUNT).isLessThan(ROW_COUNT);
+ assertThat(valuesEqualTo(result, 1)).as("rows holding %s = %s",
INT_TO_BIGINT_COLUMN, PRESENT_VALUE).isEqualTo(1);
+
+ // ... and the range predicate the pre-evolution file could not prune
on at all
+ MaterializedResult rangeResult = read(postEvolutionFile, projection,
+ greaterThanThreshold(evolved, BIGINT, THRESHOLD), false,
DynamicFilter.EMPTY);
+ assertThat(rangeResult.getRowCount()).as("rows read out of %s",
ROW_COUNT).isLessThan(ROW_COUNT);
+ assertThat(valuesOver(rangeResult, 1)).as("rows matching %s > %s after
pruning", INT_TO_BIGINT_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testOnlyTheEvolvedColumnsDomainIsDropped()
+ throws Exception
+ {
+ HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT,
INTEGER);
+ HiveColumnHandle evolved = column(FLOAT_TO_DOUBLE_COLUMN,
HiveType.HIVE_DOUBLE, DOUBLE);
+ List<HiveColumnHandle> projection = List.of(stable, evolved);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
+ greaterThanThreshold(stable, INTEGER, THRESHOLD)
+ .intersect(greaterThanThreshold(evolved, DOUBLE,
(double) THRESHOLD)),
+ false, DynamicFilter.EMPTY);
+
+ // One unusable domain must not cost the whole predicate its pushdown:
the stable column's domain still
+ // prunes, which is only visible because reading everything and
reading nothing are both wrong here.
+ assertThat(result.getRowCount()).as("rows read out of %s",
ROW_COUNT).isLessThan(ROW_COUNT);
+ assertThat(valuesOver(result, 0)).as("rows matching %s > %s after
pruning", STABLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testEvolvedColumnArrivingThroughADynamicFilter()
+ throws Exception
+ {
+ HiveColumnHandle evolved = column(FLOAT_TO_DOUBLE_COLUMN,
HiveType.HIVE_DOUBLE, DOUBLE);
+ List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN,
HiveType.HIVE_INT, INTEGER), evolved);
+
+ // A dynamic filter reaches getCombinedPredicate by its own route and
its handles carry the metastore type
+ // just the same, so it has to be guarded on the same path
+ MaterializedResult result = read(preEvolutionFile, projection,
TupleDomain.all(), false,
+ dynamicFilterOn(greaterThanThreshold(evolved, DOUBLE, (double)
THRESHOLD)));
+
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+ }
+
+ /**
+ * The control for the test above, which on its own cannot tell a guarded
dynamic filter from one that never
+ * reached {@code getPushdownPredicate} at all -- both read every row.
Putting the same shape of filter on a
+ * column the guard keeps has to prune, so a dynamic filter that quietly
stopped being pushed down fails here.
+ */
+ @Test
+ public void testDynamicFilterOnTheStableColumnStillPrunes()
+ throws Exception
+ {
+ HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT,
INTEGER);
+ List<HiveColumnHandle> projection = List.of(stable);
+
+ MaterializedResult result = read(preEvolutionFile, projection,
TupleDomain.all(), false,
+ dynamicFilterOn(greaterThanThreshold(stable, INTEGER,
THRESHOLD)));
+
+ assertThat(result.getRowCount()).as("rows read out of %s",
ROW_COUNT).isLessThan(ROW_COUNT);
+ assertThat(valuesOver(result, 0)).as("rows matching %s > %s after
pruning", STABLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ private static void
appendMetaColumns(org.apache.parquet.example.data.Group group, int row)
+ {
+ for (String metaColumn : HOODIE_META_COLUMNS) {
+ group.append(metaColumn, metaColumn + "_" + row);
+ }
+ }
+
+ /**
+ * The base file as it was written BEFORE the evolution: the five {@code
_hoodie_*} meta columns followed by the
+ * data columns in their original types. The metastore column list the
handles below model reports the widened
+ * types instead, which is the whole point of the fixture.
+ */
+ private static MessageType preEvolutionFileSchema()
+ {
+ return fileSchema(
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FLOAT,
OPTIONAL).named(FLOAT_TO_DOUBLE_COLUMN),
+ Types.primitive(INT32, OPTIONAL).named(INT_TO_BIGINT_COLUMN),
+ Types.primitive(INT32, OPTIONAL).named(INT_TO_VARCHAR_COLUMN));
+ }
+
+ /** The same table one insert later: every column now written as the type
the metastore already reported. */
+ private static MessageType postEvolutionFileSchema()
+ {
+ return fileSchema(
+ Types.primitive(PrimitiveType.PrimitiveTypeName.DOUBLE,
OPTIONAL).named(FLOAT_TO_DOUBLE_COLUMN),
+ Types.primitive(INT64, OPTIONAL).named(INT_TO_BIGINT_COLUMN),
+ Types.primitive(BINARY,
OPTIONAL).as(LogicalTypeAnnotation.stringType()).named(INT_TO_VARCHAR_COLUMN));
+ }
+
+ private static MessageType fileSchema(org.apache.parquet.schema.Type...
evolvedColumns)
+ {
+ List<org.apache.parquet.schema.Type> fields = new ArrayList<>();
+ for (String metaColumn : HOODIE_META_COLUMNS) {
+ fields.add(Types.primitive(BINARY,
OPTIONAL).as(LogicalTypeAnnotation.stringType()).named(metaColumn));
+ }
+ fields.add(Types.primitive(INT32, OPTIONAL).named(STABLE_COLUMN));
+ fields.addAll(List.of(evolvedColumns));
+ return new MessageType("hudi_base_file", fields);
+ }
+
+ /**
+ * A handle as the metastore reports the column AFTER the evolution, on
its physical ordinal. Both files lay the
+ * columns out identically, so one handle serves either.
+ */
+ private static HiveColumnHandle column(String columnName, HiveType
hiveType, Type trinoType)
+ {
+ return createBaseColumn(columnName, physicalIndexOf(columnName),
hiveType, trinoType,
+ HiveColumnHandle.ColumnType.REGULAR, Optional.empty());
+ }
+
+ private static int physicalIndexOf(String columnName)
+ {
+ List<org.apache.parquet.schema.Type> fields =
preEvolutionFileSchema().getFields();
+ for (int i = 0; i < fields.size(); i++) {
+ if (fields.get(i).getName().equals(columnName)) {
+ return i;
+ }
+ }
+ throw new IllegalArgumentException("No such column in the fixture: " +
columnName);
+ }
+
+ private static TupleDomain<HiveColumnHandle>
greaterThanThreshold(HiveColumnHandle handle, Type type, Object threshold)
+ {
+ return TupleDomain.withColumnDomains(Map.of(handle,
+ Domain.create(ValueSet.ofRanges(Range.greaterThan(type,
threshold)), false)));
+ }
+
+ /** Counts the rows whose {@code fieldIndex}-th field is over {@link
#THRESHOLD}, whatever numeric type it read as. */
+ private static long valuesOver(MaterializedResult result, int fieldIndex)
+ {
+ return result.getMaterializedRows().stream()
+ .map(row -> row.getField(fieldIndex))
+ .filter(value -> value != null && ((Number)
value).doubleValue() > THRESHOLD)
+ .count();
+ }
+
+ private static long valuesEqualTo(MaterializedResult result, int
fieldIndex)
+ {
+ return result.getMaterializedRows().stream()
+ .map(row -> row.getField(fieldIndex))
+ .filter(value -> value != null && ((Number) value).longValue()
== PRESENT_VALUE)
+ .count();
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
index a224f2ffb6c2..c677e1c5b2ce 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
@@ -13,18 +13,9 @@
*/
package io.trino.plugin.hudi;
-import io.trino.filesystem.local.LocalInputFile;
import io.trino.metastore.HiveType;
-import io.trino.parquet.ParquetReaderOptions;
-import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.HiveColumnProjectionInfo;
-import io.trino.plugin.hive.parquet.ParquetReaderConfig;
-import io.trino.plugin.hudi.file.HudiBaseFile;
-import io.trino.spi.SplitWeight;
-import io.trino.spi.connector.ColumnHandle;
-import io.trino.spi.connector.ConnectorPageSource;
-import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.DynamicFilter;
import io.trino.spi.predicate.Domain;
import io.trino.spi.predicate.Range;
@@ -34,43 +25,30 @@ import io.trino.spi.type.BigintType;
import io.trino.spi.type.RowType;
import io.trino.spi.type.Type;
import io.trino.testing.MaterializedResult;
-import io.trino.testing.TestingConnectorSession;
-import org.apache.parquet.conf.PlainParquetConfiguration;
-import org.apache.parquet.example.data.Group;
-import org.apache.parquet.example.data.simple.SimpleGroupFactory;
-import org.apache.parquet.hadoop.ParquetFileReader;
-import org.apache.parquet.hadoop.ParquetWriter;
-import org.apache.parquet.hadoop.example.ExampleParquetWriter;
-import org.apache.parquet.io.LocalOutputFile;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Types;
-import org.joda.time.DateTimeZone;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
-import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.OptionalLong;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
-import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
import static
io.trino.plugin.hudi.HudiPageSourceProvider.remapColumnIndicesToPhysical;
import static
io.trino.plugin.hudi.HudiPageSourceProvider.remapPredicateColumnIndicesToPhysical;
+import static io.trino.plugin.hudi.TestingBaseFilePageSource.dynamicFilterOn;
+import static io.trino.plugin.hudi.TestingBaseFilePageSource.writeBaseFile;
import static io.trino.spi.type.DoubleType.DOUBLE;
import static io.trino.spi.type.IntegerType.INTEGER;
import static io.trino.spi.type.VarcharType.VARCHAR;
-import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
import static java.lang.Integer.parseInt;
import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS;
import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
@@ -113,35 +91,22 @@ class TestHudiPageSourceProviderTest
private static Path baseFile;
@BeforeAll
- static void writeBaseFile()
+ static void writeFixture()
throws IOException
{
- MessageType schema = hudiFileSchema(DATA_COLUMN_COUNT);
baseFile = tempDir.resolve("base_file.parquet");
- SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema);
- try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new
LocalOutputFile(baseFile))
- .withType(schema)
- .withConf(new PlainParquetConfiguration())
- .withRowGroupSize(1024L)
- .withPageSize(512)
- .build()) {
- for (int row = 0; row < ROW_COUNT; row++) {
- Group group = groupFactory.newGroup();
- for (String metaColumn : HOODIE_META_COLUMNS) {
- group.append(metaColumn, metaColumn + "_" + row);
- }
- for (int column = 0; column < DATA_COLUMN_COUNT; column++) {
- String columnName = "c" + column;
- group.append(columnName,
columnName.equals(PREDICATE_COLUMN) ? row : row % 10);
- }
- writer.write(group);
+ int rowGroups = writeBaseFile(baseFile,
hudiFileSchema(DATA_COLUMN_COUNT), ROW_COUNT, (group, row) -> {
+ for (String metaColumn : HOODIE_META_COLUMNS) {
+ group.append(metaColumn, metaColumn + "_" + row);
}
- }
- // The writer flushes a row group whenever the buffered size is over
withRowGroupSize, checked every
- // parquet.page.size.row.check.min records (100 by default), which is
what actually splits this file.
- // Assert the outcome rather than the knobs: with a single row group
there would be nothing to prune,
- // and every test below would pass without proving anything.
- assertThat(rowGroupCount(baseFile)).as("row groups
written").isGreaterThan(1);
+ for (int column = 0; column < DATA_COLUMN_COUNT; column++) {
+ String columnName = "c" + column;
+ group.append(columnName, columnName.equals(PREDICATE_COLUMN) ?
row : row % 10);
+ }
+ });
+ // Assert the outcome rather than the writer knobs: with a single row
group there would be nothing to prune,
+ // and every reading test below would pass without proving anything.
+ assertThat(rowGroups).as("row groups written").isGreaterThan(1);
}
@Test
@@ -579,10 +544,6 @@ class TestHudiPageSourceProviderTest
.isEqualTo(byName.getMaterializedRows());
}
- /**
- * Reads the whole base file through the page source the connector builds
for a split with no log files, which
- * is the only path on which it enables predicate pushdown.
- */
private static MaterializedResult read(
List<HiveColumnHandle> projection,
TupleDomain<HiveColumnHandle> predicate,
@@ -590,38 +551,7 @@ class TestHudiPageSourceProviderTest
DynamicFilter dynamicFilter)
throws Exception
{
- long fileSize = Files.size(baseFile);
- HudiSplit split = new HudiSplit(
- new HudiBaseFile(baseFile.toString(),
baseFile.getFileName().toString(), fileSize, 0, 0, fileSize),
- List.of(),
- "000",
- predicate,
- List.of(),
- SplitWeight.standard());
- HudiSessionProperties sessionProperties = new HudiSessionProperties(
- new
HudiConfig().setUseParquetColumnNames(useParquetColumnNames),
- new ParquetReaderConfig());
- ConnectorSession session = TestingConnectorSession.builder()
- .setPropertyMetadata(sessionProperties.getSessionProperties())
- .build();
-
- List<Type> types =
projection.stream().map(HiveColumnHandle::getType).toList();
- try (ConnectorPageSource pageSource = createPageSource(
- session,
- projection,
- split,
- new LocalInputFile(baseFile.toFile()),
- baseFile.toString(),
- 0L,
- fileSize,
- OptionalLong.of(fileSize),
- new FileFormatDataSourceStats(),
- ParquetReaderOptions.builder().build(),
- DateTimeZone.UTC,
- dynamicFilter,
- true)) {
- return materializeSourceDataStream(session, pageSource,
types).toTestTypes();
- }
+ return TestingBaseFilePageSource.read(baseFile, projection, predicate,
useParquetColumnNames, dynamicFilter);
}
/**
@@ -687,42 +617,6 @@ class TestHudiPageSourceProviderTest
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER,
THRESHOLD)), false)));
}
- private static DynamicFilter dynamicFilterOn(TupleDomain<HiveColumnHandle>
predicate)
- {
- return new DynamicFilter()
- {
- @Override
- public Set<ColumnHandle> getColumnsCovered()
- {
- return
Set.copyOf(predicate.getDomains().orElseThrow().keySet());
- }
-
- @Override
- public CompletableFuture<?> isBlocked()
- {
- return CompletableFuture.completedFuture(null);
- }
-
- @Override
- public boolean isComplete()
- {
- return true;
- }
-
- @Override
- public boolean isAwaitable()
- {
- return false;
- }
-
- @Override
- public TupleDomain<ColumnHandle> getCurrentPredicate()
- {
- return predicate.transformKeys(ColumnHandle.class::cast);
- }
- };
- }
-
private static long matchingRowCount(MaterializedResult result,
List<HiveColumnHandle> projection, String columnName)
{
int fieldIndex = projection.indexOf(dataColumn(columnName));
@@ -732,14 +626,6 @@ class TestHudiPageSourceProviderTest
.count();
}
- private static int rowGroupCount(Path path)
- throws IOException
- {
- try (ParquetFileReader reader = ParquetFileReader.open(new
org.apache.parquet.io.LocalInputFile(path))) {
- return reader.getRowGroups().size();
- }
- }
-
/**
* Returns the single remapped handle carrying the given base column name.
*/
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index 51d0ae1a375a..981191962f28 100644
--- a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -27,7 +27,9 @@ import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.hive.HiveTimestampPrecision;
import io.trino.plugin.hive.parquet.ParquetReaderConfig;
import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.plugin.hudi.testing.CompositeHudiTablesInitializer;
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer;
import io.trino.spi.SplitWeight;
import io.trino.spi.connector.ConnectorPageSource;
import io.trino.spi.connector.ConnectorSession;
@@ -71,6 +73,12 @@ import java.util.stream.Stream;
import static io.trino.metastore.HiveType.HIVE_TIMESTAMP;
import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.DOUBLE_THRESHOLD;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.FLOAT_TO_DOUBLE_COLUMN;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.INT_TO_BIGINT_COLUMN;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.INT_TO_VARCHAR_COLUMN;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.VARCHAR_THRESHOLD;
+import static
io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.expectedRowsFrom;
import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COMPREHENSIVE_TYPES_V6_MOR;
import static
io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable.HUDI_COMPREHENSIVE_TYPES_V8_MOR;
@@ -102,7 +110,14 @@ public class TestHudiSmokeTest
throws Exception
{
HudiQueryRunner.Builder builder = HudiQueryRunner.builder()
- .setDataLoader(new ResourceHudiTablesInitializer())
+ // The resource tables cover the connector's read surface; the
second fixture is a table whose base
+ // file predates a type widening, which is what
testPredicateOnColumnEvolvedFromFloatToDouble and its
+ // neighbours below read. Loading it here rather than from a
suite of its own is what gets it run in
+ // both hudi.parquet.use-column-names modes, since
TestHudiConnectorParquetColumnNamesTest reruns
+ // this class positionally.
+ .setDataLoader(new CompositeHudiTablesInitializer(
+ new ResourceHudiTablesInitializer(),
+ new SchemaEvolutionHudiTablesInitializer()))
.addConnectorProperties(getAdditionalHudiProperties());
getBlobCacheProperties().ifPresent(cacheProperties -> builder
.withPlugin(new AlluxioBlobCachePlugin())
@@ -1326,41 +1341,66 @@ public class TestHudiSmokeTest
assertQuery(session, actualQuery, expectedQuery);
}
+ /**
+ * apache/hudi#19457: a predicate on a column whose type was widened after
a base file was written used to fail
+ * the whole query with {@code Malformed Parquet file. Corrupted
statistics for column ...}, because the domain
+ * carries the metastore's widened type while the file's statistics are
still of the original one. The connector
+ * now leaves such a domain out of the parquet predicate, and the engine
applies it above the scan as it always
+ * did.
+ * <p>
+ * Selecting the evolved column without constraining it was never
affected, so the predicate has to be ON it --
+ * and the projection has to include it, or the domain would resolve to no
descriptor and be discarded for an
+ * unrelated reason, which is exactly the shape that passes against the
unfixed code.
+ * {@link #testReadingAnEvolvedColumnWithoutAPredicate} is the anchor
showing the read path itself was always
+ * fine.
+ * <p>
+ * {@link TestHudiConnectorParquetColumnNamesTest} reruns this class with
+ * {@code hudi.parquet.use-column-names=false}. The guard runs on
descriptor keys, after the resolution fork, so
+ * the two modes cannot diverge here; the second run is there because the
issue was reported against both.
+ * {@code TestHudiEvolvedColumnPredicates} is where the pruning these
queries cannot observe is asserted.
+ */
+ @Test
+ public void testPredicateOnColumnEvolvedFromFloatToDouble()
+ {
+ assertQuery(selectEvolvedColumnsWhere(FLOAT_TO_DOUBLE_COLUMN + " > " +
DOUBLE_THRESHOLD), expectedRowsFrom(3));
+ }
+
+ @Test
+ public void testPredicateOnColumnEvolvedFromIntToVarchar()
+ {
+ assertQuery(selectEvolvedColumnsWhere("%s >
'%s'".formatted(INT_TO_VARCHAR_COLUMN, VARCHAR_THRESHOLD)),
expectedRowsFrom(4));
+ }
+
+ @Test
+ public void testReadingAnEvolvedColumnWithoutAPredicate()
+ {
+ // The anchor: widening is a read-path feature that already worked, so
a regression here would mean the
+ // fixture stopped modelling an evolved table rather than that the
guard misbehaved
+ assertQuery(selectEvolvedColumnsWhere("true"), expectedRowsFrom(1));
+ }
+
+ private static String selectEvolvedColumnsWhere(String predicate)
+ {
+ return "SELECT key, %s, %s, %s FROM %s WHERE %s ORDER BY
key".formatted(
+ FLOAT_TO_DOUBLE_COLUMN, INT_TO_BIGINT_COLUMN,
INT_TO_VARCHAR_COLUMN,
+ SchemaEvolutionHudiTablesInitializer.TABLE_NAME, predicate);
+ }
+
private void testTimestampMicros(HiveTimestampPrecision
timestampPrecision, LocalDateTime expected)
throws Exception
{
File parquetFile = new
File(Resources.getResource("long_timestamp.parquet").toURI());
Type columnType =
createTimestampType(timestampPrecision.getPrecision());
- HudiSplit hudiSplit = new HudiSplit(
- new HudiBaseFile(parquetFile.getPath(), parquetFile.getName(),
parquetFile.length(), parquetFile.lastModified(), 0, parquetFile.length()),
- ImmutableList.of(),
- "000",
- TupleDomain.all(),
- ImmutableList.of(),
- SplitWeight.standard());
- HudiConfig config = new HudiConfig().setUseParquetColumnNames(false);
- HudiSessionProperties sessionProperties = new
HudiSessionProperties(config, new ParquetReaderConfig());
- ConnectorSession session = TestingConnectorSession.builder()
- .setPropertyMetadata(sessionProperties.getSessionProperties())
- .build();
-
- try (ConnectorPageSource pageSource = createPageSource(
- session,
+ MaterializedResult result = TestingBaseFilePageSource.read(
+ parquetFile.toPath(),
List.of(createBaseColumn("created", 0, HIVE_TIMESTAMP,
columnType, REGULAR, Optional.empty())),
- hudiSplit,
- new LocalInputFile(parquetFile),
- parquetFile.getPath(),
- 0L,
- parquetFile.length(),
- OptionalLong.of(parquetFile.length()),
- new FileFormatDataSourceStats(),
- ParquetReaderOptions.builder().build(),
- DateTimeZone.UTC, DynamicFilter.EMPTY, true)) {
- MaterializedResult result = materializeSourceDataStream(session,
pageSource, List.of(columnType)).toTestTypes();
- assertThat(result.getMaterializedRows())
- .containsOnly(new MaterializedRow(List.of(expected)));
- }
+ TupleDomain.all(),
+ false,
+ DynamicFilter.EMPTY);
+
+ assertThat(result.getMaterializedRows())
+ .containsOnly(new MaterializedRow(List.of(expected)));
}
private static Pattern getScanFilterInputRowsPattern(String
tableIdentifier)
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestingBaseFilePageSource.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestingBaseFilePageSource.java
new file mode 100644
index 000000000000..d0eefa4f57df
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestingBaseFilePageSource.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.trino.filesystem.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.type.Type;
+import io.trino.testing.MaterializedResult;
+import io.trino.testing.TestingConnectorSession;
+import org.apache.parquet.conf.PlainParquetConfiguration;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.schema.MessageType;
+import org.joda.time.DateTimeZone;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.ObjIntConsumer;
+
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+
+/**
+ * Drives {@link HudiPageSourceProvider#createPageSource} over a single base
file, which is the only path on which
+ * the connector enables predicate pushdown - a split carrying log files takes
the merge path instead.
+ * <p>
+ * {@code createPageSource} is package-private, so anything exercising it
directly has to live in this package. Three
+ * suites do: {@link TestHudiPageSourceProviderTest} for column resolution,
{@link TestHudiEvolvedColumnPredicates}
+ * for type evolution, and {@link TestHudiSmokeTest} for timestamp precision.
They had a writer, a reader, a dynamic
+ * filter and a row-group counter each, byte-for-byte the same; keeping one
copy is what stops the three from
+ * drifting into testing subtly different page sources.
+ */
+final class TestingBaseFilePageSource
+{
+ private TestingBaseFilePageSource() {}
+
+ /**
+ * Reads the whole file through the page source the connector builds for a
split with no log files.
+ * <p>
+ * The predicate is handed over as the split's, the dynamic filter
separately, exactly as
+ * {@code HudiPageSourceProvider} receives them, so both routes into
{@code getCombinedPredicate} stay reachable
+ * from a test.
+ */
+ static MaterializedResult read(
+ Path baseFile,
+ List<HiveColumnHandle> projection,
+ TupleDomain<HiveColumnHandle> predicate,
+ boolean useParquetColumnNames,
+ DynamicFilter dynamicFilter)
+ throws Exception
+ {
+ long fileSize = Files.size(baseFile);
+ HudiSplit split = new HudiSplit(
+ new HudiBaseFile(baseFile.toString(),
baseFile.getFileName().toString(), fileSize, 0, 0, fileSize),
+ List.of(),
+ "000",
+ predicate,
+ List.of(),
+ SplitWeight.standard());
+ HudiSessionProperties sessionProperties = new HudiSessionProperties(
+ new
HudiConfig().setUseParquetColumnNames(useParquetColumnNames),
+ new ParquetReaderConfig());
+ ConnectorSession session = TestingConnectorSession.builder()
+ .setPropertyMetadata(sessionProperties.getSessionProperties())
+ .build();
+
+ List<Type> types =
projection.stream().map(HiveColumnHandle::getType).toList();
+ try (ConnectorPageSource pageSource = createPageSource(
+ session,
+ projection,
+ split,
+ new LocalInputFile(baseFile.toFile()),
+ baseFile.toString(),
+ 0L,
+ fileSize,
+ OptionalLong.of(fileSize),
+ new FileFormatDataSourceStats(),
+ ParquetReaderOptions.builder().build(),
+ DateTimeZone.UTC,
+ dynamicFilter,
+ true)) {
+ return materializeSourceDataStream(session, pageSource,
types).toTestTypes();
+ }
+ }
+
+ /**
+ * Writes {@code rowCount} rows of {@code schema} to {@code baseFile} and
returns how many row groups that took.
+ * <p>
+ * The writer flushes a row group whenever the buffered size passes {@code
withRowGroupSize}, checked every
+ * {@code parquet.page.size.row.check.min} records (100 by default), so
the small sizes below are what actually
+ * split the file. Callers assert on the returned count rather than on
those knobs: with a single row group there
+ * would be nothing to prune and every "still prunes" assertion would hold
without proving anything.
+ * <p>
+ * {@code bloomFilterColumns} names the columns to build a parquet bloom
filter for, which is what a Hudi writer
+ * does for every column set in {@code
parquet.bloom.filter.enabled#<column>}. Trino reads bloom filters by
+ * default, so a column listed here is checked by {@code
TupleDomainParquetPredicate.matches(BloomFilterStore,
+ * int)} as well as by the statistics.
+ */
+ static int writeBaseFile(Path baseFile, MessageType schema, int rowCount,
List<String> bloomFilterColumns, ObjIntConsumer<Group> fillRow)
+ throws IOException
+ {
+ SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema);
+ ExampleParquetWriter.Builder builder =
ExampleParquetWriter.builder(new LocalOutputFile(baseFile))
+ .withType(schema)
+ .withConf(new PlainParquetConfiguration())
+ .withRowGroupSize(1024L)
+ .withPageSize(512);
+ for (String column : bloomFilterColumns) {
+ builder = builder.withBloomFilterEnabled(column, true);
+ }
+ try (ParquetWriter<Group> writer = builder.build()) {
+ for (int row = 0; row < rowCount; row++) {
+ Group group = groupFactory.newGroup();
+ fillRow.accept(group, row);
+ writer.write(group);
+ }
+ }
+ return rowGroupCount(baseFile);
+ }
+
+ static int writeBaseFile(Path baseFile, MessageType schema, int rowCount,
ObjIntConsumer<Group> fillRow)
+ throws IOException
+ {
+ return writeBaseFile(baseFile, schema, rowCount, List.of(), fillRow);
+ }
+
+ static int rowGroupCount(Path path)
+ throws IOException
+ {
+ try (ParquetFileReader reader = ParquetFileReader.open(new
org.apache.parquet.io.LocalInputFile(path))) {
+ return reader.getRowGroups().size();
+ }
+ }
+
+ /** A completed dynamic filter carrying {@code predicate}, the shape
{@code getCombinedPredicate} intersects in. */
+ static DynamicFilter dynamicFilterOn(TupleDomain<HiveColumnHandle>
predicate)
+ {
+ return new DynamicFilter()
+ {
+ @Override
+ public Set<ColumnHandle> getColumnsCovered()
+ {
+ return
Set.copyOf(predicate.getDomains().orElseThrow().keySet());
+ }
+
+ @Override
+ public CompletableFuture<?> isBlocked()
+ {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public boolean isComplete()
+ {
+ return true;
+ }
+
+ @Override
+ public boolean isAwaitable()
+ {
+ return false;
+ }
+
+ @Override
+ public TupleDomain<ColumnHandle> getCurrentPredicate()
+ {
+ return predicate.transformKeys(ColumnHandle.class::cast);
+ }
+ };
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SchemaEvolutionHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SchemaEvolutionHudiTablesInitializer.java
new file mode 100644
index 000000000000..3387c94f9474
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SchemaEvolutionHudiTablesInitializer.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.metastore.HiveType.HIVE_DOUBLE;
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+
+/**
+ * Creates a table whose base file was written BEFORE its columns were
widened, while the metastore reports the types
+ * they were widened TO -- the state every unrewritten base file is in after a
Hudi type evolution followed by a hive
+ * sync. The two halves are declared independently on purpose: {@link
#avroSchema()} is what the write client puts in
+ * the file, {@link #dataColumns()} is what the metastore hands the connector,
and only their disagreement is being
+ * modelled.
+ * <p>
+ * No schema-evolution write path is exercised, because none is needed to
reproduce apache/hudi#19457 and because
+ * neither route can run here. A real evolution is produced either by a
DataFrame append carrying a widened schema -
+ * schema-on-write, which is what apache/hudi#19457 was reported against - or
by
+ * {@code ALTER TABLE ... ALTER COLUMN ... TYPE}, which needs {@code
hoodie.schema.on.read.enable=true}:
+ * {@code AlterHoodieTableChangeColumnCommand} throws {@code
HoodieAnalysisException} for any type change. Both want
+ * a Spark writer, so the end-to-end version of this fixture belongs with the
docker-based integ tests rather than in
+ * a connector unit test; it is tracked in apache/hudi#19743.
+ * <p>
+ * The three widenings are the ones Hudi allows and the parquet reader can
serve: {@code float -> double},
+ * {@code int -> long} and {@code int -> string}. Reading them has always
worked. Putting a predicate on one is what
+ * used to fail the query, because the statistics in the file are still of the
original type while the pushed-down
+ * domain carries the widened one.
+ * <p>
+ * Unlike {@link OmittedMetaColumnsHudiTablesInitializer} this fixture
registers the Hudi meta columns, so every
+ * metastore ordinal equals its physical one and nothing here depends on how
columns are resolved. What the two
+ * {@code hudi.parquet.use-column-names} modes must agree on is the TYPE
handling alone; both run it, because
+ * {@code TestHudiSmokeTest} loads this fixture and {@code
TestHudiConnectorParquetColumnNamesTest} reruns that class
+ * positionally.
+ * <p>
+ * A single bulk-insert commit, so the file slice has no log files: predicate
pushdown is only enabled for
+ * base-file-only splits.
+ */
+public class SchemaEvolutionHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "schema_evolved_mor";
+
+ /** Written as Avro {@code float}, reported by the metastore as {@code
double}. */
+ public static final String FLOAT_TO_DOUBLE_COLUMN = "float_value";
+ /** Written as Avro {@code int}, reported by the metastore as {@code
bigint}. */
+ public static final String INT_TO_BIGINT_COLUMN = "int_value";
+ /** Written as Avro {@code int}, reported by the metastore as {@code
string}. */
+ public static final String INT_TO_VARCHAR_COLUMN = "string_value";
+
+ /** Sits between the third and the fourth row, so a predicate on it keeps
some rows and drops others. */
+ public static final String DOUBLE_THRESHOLD = "1003.0";
+ public static final long BIGINT_THRESHOLD = 1003;
+ public static final String VARCHAR_THRESHOLD = "1003";
+
+ private static final int ROW_COUNT = 5;
+ private static final int BASE_VALUE = 1000;
+
+ public SchemaEvolutionHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(FLOAT_TO_DOUBLE_COLUMN, HIVE_DOUBLE,
Optional.empty(), Map.of()),
+ new Column(INT_TO_BIGINT_COLUMN, HIVE_LONG, Optional.empty(),
Map.of()),
+ new Column(INT_TO_VARCHAR_COLUMN, HIVE_STRING,
Optional.empty(), Map.of()),
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(),
Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = new ArrayList<>();
+ fields.add(new Schema.Field(FLOAT_TO_DOUBLE_COLUMN,
Schema.create(Schema.Type.FLOAT)));
+ fields.add(new Schema.Field(INT_TO_BIGINT_COLUMN,
Schema.create(Schema.Type.INT)));
+ fields.add(new Schema.Field(INT_TO_VARCHAR_COLUMN,
Schema.create(Schema.Type.INT)));
+ fields.add(new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)));
+ fields.add(new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, fields);
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ List<HoodieRecord<HoodieAvroPayload>> records = new ArrayList<>();
+ for (int row = 1; row <= ROW_COUNT; row++) {
+ records.add(record(schema, row));
+ }
+ // One commit only: a file slice with log files would take the merge
path, which disables pushdown.
+ String commit = client.startCommit();
+ List<WriteStatus> statuses = client.bulkInsert(records, commit);
+ client.commit(commit, statuses);
+ }
+
+ /**
+ * The expected rows of {@code SELECT key, float_value, int_value,
string_value ... WHERE <column> > <threshold>}
+ * for the rows whose index is over {@code firstMatchingRow}.
+ * <p>
+ * Every float value is an exact binary fraction, so widening it to double
is lossless and the expected literals
+ * can be written out in full rather than compared with a tolerance.
+ */
+ public static String expectedRowsFrom(int firstMatchingRow)
+ {
+ List<String> rows = new ArrayList<>();
+ for (int row = firstMatchingRow; row <= ROW_COUNT; row++) {
+ rows.add("('k%s', CAST(%s AS DOUBLE), CAST(%s AS BIGINT),
'%s')".formatted(
+ row, floatValue(row), BASE_VALUE + row, BASE_VALUE + row));
+ }
+ return "VALUES " + String.join(", ", rows);
+ }
+
+ private static float floatValue(int row)
+ {
+ return BASE_VALUE + row + 0.5f;
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema, int
row)
+ {
+ String key = "k" + row;
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(FLOAT_TO_DOUBLE_COLUMN, floatValue(row));
+ record.put(INT_TO_BIGINT_COLUMN, BASE_VALUE + row);
+ record.put(INT_TO_VARCHAR_COLUMN, BASE_VALUE + row);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put(ORDERING_FIELD, 100L);
+ return avroRecord(record, key);
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java
new file mode 100644
index 000000000000..34c65e5c751a
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java
@@ -0,0 +1,390 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import io.airlift.slice.Slices;
+import io.trino.parquet.ParquetDataSourceId;
+import io.trino.parquet.predicate.TupleDomainParquetPredicate;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.Range;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.predicate.ValueSet;
+import io.trino.spi.type.CharType;
+import io.trino.spi.type.DecimalType;
+import io.trino.spi.type.Int128;
+import io.trino.spi.type.Type;
+import io.trino.spi.type.UuidType;
+import io.trino.spi.type.VarbinaryType;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import
org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Types;
+import org.joda.time.DateTimeZone;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static
io.trino.plugin.hudi.util.ParquetStatisticsDomains.dropIncomparableDomains;
+import static
io.trino.plugin.hudi.util.ParquetStatisticsDomains.hasComparableStatistics;
+import static
io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.ALL;
+import static
io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.NARROW;
+import static
io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.THROWS;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.BooleanType.BOOLEAN;
+import static io.trino.spi.type.CharType.createCharType;
+import static io.trino.spi.type.DateType.DATE;
+import static io.trino.spi.type.DoubleType.DOUBLE;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.RealType.REAL;
+import static io.trino.spi.type.TimestampType.TIMESTAMP_MILLIS;
+import static io.trino.spi.type.TinyintType.TINYINT;
+import static io.trino.spi.type.UuidType.UUID;
+import static io.trino.spi.type.VarbinaryType.VARBINARY;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
+import static
org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Pins {@link ParquetStatisticsDomains#hasComparableStatistics} against the
two methods it exists to protect. Every
+ * case below states BOTH what the guard decides and what the library actually
does with the same pair, and the check
+ * is run against the real {@code TupleDomainParquetPredicate}, not a
description of it. A Trino upgrade that moves a
+ * branch therefore fails here, where the mismatch is a line of test output,
instead of in a query.
+ * <p>
+ * The three library outcomes are worth telling apart, because the guard
exists for more than one reason:
+ * <ul>
+ * <li>{@link LibraryOutcome#THROWS} - the cast fails and the whole split
dies with {@code HUDI_BAD_DATA}. This
+ * is apache/hudi#19457 as reported.</li>
+ * <li>{@link LibraryOutcome#NARROW} on a pair the guard drops - far
worse: a domain IS produced, from bytes that
+ * mean something else entirely, and row groups get pruned on a comparison
that is simply false. Nothing fails,
+ * rows just go missing.</li>
+ * <li>{@link LibraryOutcome#ALL} - {@code getDomain} declines the pair
itself. Keeping such a pair is still
+ * worth it when the bloom filter can answer it, and safe either way,
because the declined domain covers every
+ * value and only the column's null count prunes.</li>
+ * </ul>
+ * A kept pair must never be one the library reads out of a representation
that means something else, so every
+ * cross-type kept pair states the exact domain {@code getDomain} has to come
back with. Asserting only that the
+ * result is narrower than "any value" would pass a rescale that silently
landed a factor of a hundred out.
+ * <p>
+ * {@code getDomain} is only half of what the guard has to mirror. The other
half is {@code checkInBloomFilter},
+ * which dispatches on the domain type of its own accord and hashes the lookup
at that type's width - the reason the
+ * integral arm insists on a matching physical width and the reason the
byte-typed arm exists at all. Its bounds are
+ * not visible in a domain, so it is pinned where the harm shows up instead,
on the row groups a real read returns:
+ * {@code
TestHudiEvolvedColumnPredicates.testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows}.
+ */
+class TestParquetStatisticsDomains
+{
+ private static final ParquetDataSourceId DATA_SOURCE_ID = new
ParquetDataSourceId("test");
+ private static final long VALUE_COUNT = 10;
+ private static final int MIN_VALUE = 1;
+ private static final int MAX_VALUE = 2;
+
+ enum LibraryOutcome
+ {
+ /** getDomain read the statistics and returned a range narrower than
"any value". */
+ NARROW,
+ /** getDomain declined to use the statistics and returned a domain
covering every value. */
+ ALL,
+ /** getDomain failed, which the connector reports as a
corrupt-statistics error over the whole split. */
+ THROWS,
+ }
+
+ private record TypePair(String description, Type domainType, PrimitiveType
fileType, boolean comparable,
+ LibraryOutcome outcome, Optional<Domain>
expectedDomain)
+ {
+ @Override
+ public String toString()
+ {
+ return description;
+ }
+ }
+
+ private static List<TypePair> typePairs()
+ {
+ return List.of(
+ // A column that never evolved: the domain's type is the one
the file was written with
+ kept("boolean over BOOLEAN", BOOLEAN,
plain(PrimitiveTypeName.BOOLEAN), NARROW),
+ kept("integer over INT32", INTEGER, plain(INT32), NARROW),
+ kept("bigint over INT64", BIGINT, plain(INT64), NARROW),
+ kept("tinyint over INT32", TINYINT, plain(INT32), NARROW),
+ kept("date over INT32 date", DATE, annotated(INT32,
LogicalTypeAnnotation.dateType()), NARROW),
+ kept("real over FLOAT", REAL, plain(FLOAT), NARROW),
+ kept("double over DOUBLE", DOUBLE,
plain(PrimitiveTypeName.DOUBLE), NARROW),
+ kept("varchar over BINARY string", VARCHAR, annotated(BINARY,
LogicalTypeAnnotation.stringType()), NARROW),
+ kept("decimal(9,2) over INT32 decimal(9,2)",
DecimalType.createDecimalType(9, 2), decimal(INT32, 9, 2), NARROW),
+ kept("timestamp over INT64 timestamp", TIMESTAMP_MILLIS,
annotated(INT64, LogicalTypeAnnotation.timestampType(false, TimeUnit.MILLIS)),
NARROW),
+ kept("timestamp over INT96", TIMESTAMP_MILLIS, plain(INT96),
NARROW),
+ // A char, a varbinary and a uuid all fall past every
getDomain branch, so it declines them and only
+ // the null count prunes. Keeping them costs nothing and buys
back the bloom filter, which does have
+ // a branch for the last two.
+ kept("char over BINARY", createCharType(4), plain(BINARY),
ALL),
+ kept("varbinary over BINARY", VARBINARY, plain(BINARY), ALL),
+ kept("uuid over FIXED_LEN_BYTE_ARRAY", UUID,
plain(FIXED_LEN_BYTE_ARRAY), ALL),
+
+ // Promotions the statistics can answer, so pushdown must
survive them. The domain each one has to
+ // come back with is spelled out: a rescale that lands a
factor of ten out is still "narrow".
+ keptReading("decimal(20,2) -> decimal(38,4)",
DecimalType.createDecimalType(38, 4), decimal(FIXED_LEN_BYTE_ARRAY, 20, 2),
+ rangeOf(DecimalType.createDecimalType(38, 4),
Int128.valueOf(100), Int128.valueOf(200))),
+ keptReading("decimal(9,2) over BINARY decimal(9,2)",
DecimalType.createDecimalType(9, 2), decimal(BINARY, 9, 2),
+ rangeOf(DecimalType.createDecimalType(9, 2), (long)
MIN_VALUE, (long) MAX_VALUE)),
+ keptReading("integer over a zero-scale INT32 decimal",
INTEGER, decimal(INT32, 9, 0),
+ rangeOf(INTEGER, (long) MIN_VALUE, (long) MAX_VALUE)),
+ // Every table synced with hive_sync.support_timestamp=false
carries this pair: HiveSchemaUtil maps
+ // TIMESTAMP to BIGINT, so the metastore says bigint while the
file stays an annotated INT64.
+ keptReading("bigint over INT64 timestamp(micros)", BIGINT,
annotated(INT64, LogicalTypeAnnotation.timestampType(true, TimeUnit.MICROS)),
+ rangeOf(BIGINT, (long) MIN_VALUE, (long) MAX_VALUE)),
+ keptReading("varchar over a plain FIXED_LEN_BYTE_ARRAY",
VARCHAR, plain(FIXED_LEN_BYTE_ARRAY),
+ rangeOf(VARCHAR, sliceOf(plain(FIXED_LEN_BYTE_ARRAY),
MIN_VALUE), sliceOf(plain(FIXED_LEN_BYTE_ARRAY), MAX_VALUE))),
+
+ // Promotions the min/max side reads correctly but the BLOOM
side does not. checkInBloomFilter hashes
+ // the lookup at the domain's width, parquet-mr hashed the
column at the file's, so the filter misses
+ // and the row group is dropped with its rows still in it. See
apache/hudi#19457 and trinodb/trino#30544;
+ //
TestHudiEvolvedColumnPredicates.testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows
pins the mechanism.
+ dropped("int -> long", BIGINT, plain(INT32), NARROW),
+ dropped("long -> int", INTEGER, plain(INT64), NARROW),
+ // Hudi rejects this evolution as lossy
(HoodieSchemaCompatibilityChecker lets neither the integer
+ // digits nor the scale shrink) and so does the guard: the
rescale in getShortDecimal multiplies by a
+ // hundred, which throws NUMERIC_VALUE_OUT_OF_RANGE for any
file bound over 9999999 -- read as
+ // "Corrupted statistics", the very failure this class
removes. Small bounds hide that, hence the drop
+ // is on the type pair rather than on the values.
+ dropped("decimal(9,2) -> decimal(9,4), lossy",
DecimalType.createDecimalType(9, 4), decimal(INT32, 9, 2), NARROW),
+
+ // Promotions that fail the split today: apache/hudi#19457 and
its neighbours
+ dropped("float -> double", DOUBLE, plain(FLOAT), THROWS),
+ dropped("int -> double", DOUBLE, plain(INT32), THROWS),
+ dropped("long -> double", DOUBLE, plain(INT64), THROWS),
+ dropped("int -> float", REAL, plain(INT32), THROWS),
+ dropped("int -> string", VARCHAR, plain(INT32), THROWS),
+ dropped("long -> string", VARCHAR, plain(INT64), THROWS),
+ dropped("float -> string", VARCHAR, plain(FLOAT), THROWS),
+ dropped("double -> string", VARCHAR,
plain(PrimitiveTypeName.DOUBLE), THROWS),
+ dropped("string -> date", DATE, annotated(BINARY,
LogicalTypeAnnotation.stringType()), THROWS),
+
+ // Promotions that silently prune on a comparison that means
nothing, which is why the guard cannot
+ // be a try/catch around the cast
+ dropped("decimal -> string", VARCHAR,
decimal(FIXED_LEN_BYTE_ARRAY, 20, 2), NARROW),
+ dropped("string -> decimal", DecimalType.createDecimalType(9,
2), annotated(BINARY, LogicalTypeAnnotation.stringType()), NARROW),
+ dropped("int -> decimal", DecimalType.createDecimalType(9, 2),
plain(INT32), NARROW),
+ dropped("integer over a scaled INT32 decimal", INTEGER,
decimal(INT32, 9, 2), NARROW),
+
+ // Pairs getDomain declines on its own. Dropping costs nothing
on the min/max side, but a varbinary
+ // over an INT32 column would still be hashed as raw bytes
against a filter built from four-byte
+ // integer hashes, so the guard has to drop it rather than
lean on getDomain declining.
+ dropped("timestamp over an unannotated INT64",
TIMESTAMP_MILLIS, plain(INT64), ALL),
+ dropped("varbinary over INT32", VARBINARY, plain(INT32), ALL));
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testGuardMatchesTheParquetPredicate(TypePair pair)
+ throws Exception
+ {
+ assertThat(hasComparableStatistics(pair.domainType(), pair.fileType()))
+ .as("guard verdict for %s", pair)
+ .isEqualTo(pair.comparable());
+
+ ColumnDescriptor descriptor = descriptorOf(pair.fileType());
+ Statistics<?> statistics = statisticsOf(pair.fileType());
+ if (pair.outcome() == THROWS) {
+ assertThatThrownBy(() ->
TupleDomainParquetPredicate.getDomain(descriptor, pair.domainType(),
VALUE_COUNT, statistics, DATA_SOURCE_ID, DateTimeZone.UTC))
+ .as("getDomain for %s", pair)
+ .hasMessageContaining("Corrupted statistics");
+ return;
+ }
+
+ Domain domain = TupleDomainParquetPredicate.getDomain(descriptor,
pair.domainType(), VALUE_COUNT, statistics, DATA_SOURCE_ID, DateTimeZone.UTC);
+ assertThat(domain.getValues().isAll())
+ .as("getDomain for %s returned %s", pair, domain)
+ .isEqualTo(pair.outcome() == ALL);
+ pair.expectedDomain().ifPresent(expected -> assertThat(domain)
+ .as("bounds getDomain read for %s", pair)
+ .isEqualTo(expected));
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testNoKeptPairIsMisread(TypePair pair)
+ {
+ // The invariant behind the whole table: a false negative only costs
pruning, but a false positive is either a
+ // failed query or a wrong one. So nothing may be kept that the
library cannot read -- either it produces a
+ // real range out of the file's own bytes, or it declines the
statistics and leaves the null count to prune.
+ if (!hasComparableStatistics(pair.domainType(), pair.fileType())) {
+ return;
+ }
+ assertThat(pair.outcome()).as("library outcome for the kept pair %s",
pair).isIn(NARROW, ALL);
+ // A pair getDomain declines is worth keeping only when something else
pays for it: the bloom filter, which
+ // has a branch for varbinary and uuid, and the null count, which
every type gets. Keeping a declined pair
+ // anywhere else would be a guard that mirrors getDomain's dispatch
wrongly rather than deliberately.
+ if (pair.outcome() == ALL) {
+ assertThat(pair.domainType())
+ .as("declined pair %s is kept only for its bloom filter
and null count", pair)
+ .isInstanceOfAny(CharType.class, VarbinaryType.class,
UuidType.class);
+ }
+ }
+
+ @Test
+ public void testAllAndNonePassThroughUntouched()
+ {
+
assertThat(dropIncomparableDomains(TupleDomain.all())).isEqualTo(TupleDomain.all());
+
assertThat(dropIncomparableDomains(TupleDomain.none())).isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testOnlyTheIncomparableDomainIsDropped()
+ {
+ // Distinct names on purpose: a ColumnDescriptor is keyed by its path,
so two columns sharing one name would
+ // collapse into a single map entry and the test would pass without
ever exercising the filtering
+ ColumnDescriptor evolved = descriptorNamed("evolved", FLOAT);
+ ColumnDescriptor stable = descriptorNamed("stable", INT32);
+ Domain doubleDomain = Domain.singleValue(DOUBLE, 1.0d);
+ Domain intDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<ColumnDescriptor> filtered = dropIncomparableDomains(
+ TupleDomain.withColumnDomains(Map.of(evolved, doubleDomain,
stable, intDomain)));
+
+
assertThat(filtered.getDomains().orElseThrow()).containsExactly(Map.entry(stable,
intDomain));
+ }
+
+ @Test
+ public void testAPredicateWithNothingToDropIsReturnedAsIs()
+ {
+ TupleDomain<ColumnDescriptor> predicate =
TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(INT32)), Domain.singleValue(INTEGER,
1L)));
+
+ assertThat(dropIncomparableDomains(predicate)).isSameAs(predicate);
+ }
+
+ @Test
+ public void testDroppingEveryDomainLeavesAnUnconstrainedPredicate()
+ {
+ TupleDomain<ColumnDescriptor> predicate =
TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(FLOAT)), Domain.singleValue(DOUBLE,
1.0d)));
+
+ // Not TupleDomain.none(): dropping means "do not prune on this",
never "this matches nothing"
+ assertThat(dropIncomparableDomains(predicate).isAll()).as("everything
dropped").isTrue();
+ }
+
+ private static TypePair kept(String description, Type domainType,
PrimitiveType fileType, LibraryOutcome outcome)
+ {
+ return new TypePair(description, domainType, fileType, true, outcome,
Optional.empty());
+ }
+
+ /** A kept pair whose domain type is not the file's own, so the exact
bounds it reads are worth pinning. */
+ private static TypePair keptReading(String description, Type domainType,
PrimitiveType fileType, Domain expectedDomain)
+ {
+ return new TypePair(description, domainType, fileType, true, NARROW,
Optional.of(expectedDomain));
+ }
+
+ private static TypePair dropped(String description, Type domainType,
PrimitiveType fileType, LibraryOutcome outcome)
+ {
+ return new TypePair(description, domainType, fileType, false, outcome,
Optional.empty());
+ }
+
+ /** The statistics fixture holds {@link #MIN_VALUE} and {@link
#MAX_VALUE}, never null, so nulls are not allowed. */
+ private static Domain rangeOf(Type type, Object min, Object max)
+ {
+ return Domain.create(ValueSet.ofRanges(Range.range(type, min, true,
max, true)), false);
+ }
+
+ private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName)
+ {
+ if (primitiveTypeName == FIXED_LEN_BYTE_ARRAY) {
+ return Types.primitive(primitiveTypeName,
OPTIONAL).length(16).named("c");
+ }
+ return Types.primitive(primitiveTypeName, OPTIONAL).named("c");
+ }
+
+ private static PrimitiveType annotated(PrimitiveTypeName
primitiveTypeName, LogicalTypeAnnotation annotation)
+ {
+ return Types.primitive(primitiveTypeName,
OPTIONAL).as(annotation).named("c");
+ }
+
+ private static PrimitiveType decimal(PrimitiveTypeName primitiveTypeName,
int precision, int scale)
+ {
+ Types.PrimitiveBuilder<PrimitiveType> builder =
Types.primitive(primitiveTypeName, OPTIONAL);
+ if (primitiveTypeName == FIXED_LEN_BYTE_ARRAY) {
+ builder = builder.length(16);
+ }
+ return builder.as(LogicalTypeAnnotation.decimalType(scale,
precision)).named("c");
+ }
+
+ private static ColumnDescriptor descriptorOf(PrimitiveType fileType)
+ {
+ return new ColumnDescriptor(new String[] {fileType.getName()},
fileType, 0, 1);
+ }
+
+ private static ColumnDescriptor descriptorNamed(String name,
PrimitiveTypeName primitiveTypeName)
+ {
+ return descriptorOf(Types.primitive(primitiveTypeName,
OPTIONAL).named(name));
+ }
+
+ private static io.airlift.slice.Slice sliceOf(PrimitiveType fileType, int
value)
+ {
+ return Slices.wrappedBuffer(statisticsBytes(fileType, value));
+ }
+
+ /**
+ * Statistics holding a small, non-degenerate range, so that a pair the
library CAN read produces a domain
+ * narrower than "any value" and the {@link LibraryOutcome#NARROW} cases
stay distinguishable from
+ * {@link LibraryOutcome#ALL}. Two nearby values also keep every integral
type clear of
+ * {@code isStatisticsOverflow}, which would otherwise widen tinyint back
to everything.
+ */
+ private static Statistics<?> statisticsOf(PrimitiveType fileType)
+ {
+ return Statistics.getBuilderForReading(fileType)
+ .withMin(statisticsBytes(fileType, MIN_VALUE))
+ .withMax(statisticsBytes(fileType, MAX_VALUE))
+ .withNumNulls(0)
+ .build();
+ }
+
+ private static byte[] statisticsBytes(PrimitiveType fileType, int value)
+ {
+ return switch (fileType.getPrimitiveTypeName()) {
+ // Both bounds false, so the boolean branch reports "only false"
rather than "true and false", which it
+ // would report as every value
+ case BOOLEAN -> new byte[] {0};
+ case INT32 ->
ByteBuffer.allocate(4).order(LITTLE_ENDIAN).putInt(value).array();
+ case INT64 ->
ByteBuffer.allocate(8).order(LITTLE_ENDIAN).putLong(value).array();
+ case FLOAT ->
ByteBuffer.allocate(4).order(LITTLE_ENDIAN).putFloat(value).array();
+ case DOUBLE ->
ByteBuffer.allocate(8).order(LITTLE_ENDIAN).putDouble(value).array();
+ // INT96 statistics are only usable when the bounds are equal
(PARQUET-1065), so ignore the value:
+ // 8 bytes of nanos-within-the-day followed by the julian day of
the epoch
+ case INT96 ->
ByteBuffer.allocate(12).order(LITTLE_ENDIAN).putLong(0).putInt(2440588).array();
+ // A decimal's unscaled value is big-endian two's complement
whichever physical type carries it, so a
+ // BINARY decimal cannot reuse the digits a BINARY string is
written as
+ case BINARY -> fileType.getLogicalTypeAnnotation() instanceof
DecimalLogicalTypeAnnotation
+ ? new byte[] {(byte) value}
+ : Integer.toString(value).getBytes(UTF_8);
+ case FIXED_LEN_BYTE_ARRAY ->
ByteBuffer.allocate(fileType.getTypeLength()).put(fileType.getTypeLength() - 1,
(byte) value).array();
+ };
+ }
+}
