voonhous commented on code in PR #19467: URL: https://github.com/apache/hudi/pull/19467#discussion_r3856305067
########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.DecimalType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.Type; +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. + */ +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 {@code TupleDomainParquetPredicate.getDomain} builds a meaningful domain out of a {@code fileType} + * column's statistics when asked for {@code domainType}, which is the case only when the two describe the same + * physical values. + * <p> + * The accepted pairs mirror that method's dispatch branch for branch. Everything else is rejected, which for a + * type it has no branch for - {@code CHAR}, {@code VARBINARY}, {@code UUID}, {@code TIME}, a timestamp with time + * zone - costs nothing at all: its fallthrough returns a domain covering every value, which prunes exactly as + * much as pushing nothing down. Only the accepted pairs can be wrong, and + * {@code TestParquetStatisticsDomains} pins each of them against the real {@code getDomain}. + */ + 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 INT32 and INT64 are interchangeable here. + // A decimal column reports its UNSCALED value though, which is the integer it stands for only at scale 0. + return (primitiveType == INT32 || primitiveType == INT64) && !isScaledDecimal(annotation); + } + if (domainType instanceof DecimalType) { + // 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. + return isDecimalPrimitive(primitiveType) && annotation instanceof DecimalLogicalTypeAnnotation; + } Review Comment: The decimal arm keeps any precision/scale mismatch, but the rescale in `getShortDecimal`/`getLongDecimal` throws `NUMERIC_VALUE_OUT_OF_RANGE` when a file bound does not fit the domain at the new scale, and that surfaces as the same `Corrupted statistics` / `HUDI_BAD_DATA` this PR removes: `decimal(9,4)` over INT32 `decimal(9,2)` with max=999999999 throws; the test row at `TestParquetStatisticsDomains:125` passes only because its bound is 2. Hudi itself rejects that pair (`HoodieSchemaCompatibilityChecker`: neither the integer digits nor the scale may shrink), so could we keep exactly Hudi's lossless widening, which is provably overflow-free? ```suggestion 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 when a file bound does not fit the domain at the new scale, so only Hudi's lossless widening // is kept: 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(); } ``` ########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.DecimalType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.Type; +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. + */ +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 {@code TupleDomainParquetPredicate.getDomain} builds a meaningful domain out of a {@code fileType} + * column's statistics when asked for {@code domainType}, which is the case only when the two describe the same + * physical values. + * <p> + * The accepted pairs mirror that method's dispatch branch for branch. Everything else is rejected, which for a + * type it has no branch for - {@code CHAR}, {@code VARBINARY}, {@code UUID}, {@code TIME}, a timestamp with time + * zone - costs nothing at all: its fallthrough returns a domain covering every value, which prunes exactly as + * much as pushing nothing down. Only the accepted pairs can be wrong, and + * {@code TestParquetStatisticsDomains} pins each of them against the real {@code getDomain}. + */ + 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 INT32 and INT64 are interchangeable here. + // A decimal column reports its UNSCALED value though, which is the integer it stands for only at scale 0. + return (primitiveType == INT32 || primitiveType == INT64) && !isScaledDecimal(annotation); Review Comment: The kept `int -> long` pair is safe for min/max but not for the bloom matcher: `checkInBloomFilter` hashes a BIGINT lookup at 8 bytes, parquet-mr built the INT32 column's filter from 4-byte hashes, so `findHash` is false and `PredicateUtils.predicateMatches` drops the row group silently (INTEGER/DATE over INT64 fails the same way via `toIntExact`). Reachable whenever the writer set `parquet.bloom.filter.enabled#<col>` (`HoodieBaseParquetWriter.handleParquetBloomFilters`); it is the residual trinodb/trino#30544 describes. Pre-existing, but this PR pins the pair as safe, so could we require the same physical width here (losing only stats pruning for `int -> long` on pre-evolution files) and flip `testPredicateOnIntColumnEvolvedToBigintStillPrunes`? If keeping that pruning matters more, would a Javadoc line naming the bloom residual plus a follow-up issue do? ```suggestion // asLong takes any integral box, but checkInBloomFilter hashes the lookup at the DOMAIN's width while // parquet-mr hashed the column at the FILE's, so a bigint over INT32 (or an int over INT64) never finds // its own value in a bloom filter and the row group is dropped. Same width only. // 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); ``` ########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.DecimalType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.Type; +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. + */ +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 {@code TupleDomainParquetPredicate.getDomain} builds a meaningful domain out of a {@code fileType} + * column's statistics when asked for {@code domainType}, which is the case only when the two describe the same + * physical values. + * <p> + * The accepted pairs mirror that method's dispatch branch for branch. Everything else is rejected, which for a + * type it has no branch for - {@code CHAR}, {@code VARBINARY}, {@code UUID}, {@code TIME}, a timestamp with time + * zone - costs nothing at all: its fallthrough returns a domain covering every value, which prunes exactly as Review Comment: The "costs nothing at all" claim does not hold: `getDomain`'s null-count early returns (`onlyNull`, `hasNullValue`) run before the type dispatch, so an `IS NULL` / `IS NOT NULL` predicate on an unevolved `binary` column loses row-group pruning once its domain is dropped here, and VARBINARY/UUID also lose bloom pruning (`checkInBloomFilter` has branches for both). Could we add an arm keeping `VarbinaryType`/`CharType`/`UuidType` over BINARY or FIXED_LEN_BYTE_ARRAY, and reword this sentence? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/testing/SchemaEvolutionHudiTablesInitializer.java: ########## @@ -0,0 +1,164 @@ +/* + * 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. No schema-evolution write path is exercised, because none is needed to reproduce apache/hudi#19457. Review Comment: With #19217 merged, could this land with the Spark-write / Trino-read E2E that was the condition for reviewing? This fixture never evolves anything, so it cannot show that hive sync reports the widened type, nor which files a real evolution leaves unrewritten. The producing routes are a DataFrame append with a widened schema (schema-on-write) or `ALTER COLUMN TYPE` under `hoodie.schema.on.read.enable=true` (`AlterHoodieTableChangeColumnCommand` rejects it otherwise), so this sentence could name them too. Sketch below. <details> <summary>Proposed test</summary> - `docker/demo/sparksql-schema-evolution-trino.commands`, modelled on `sparksql-blob-type-df.commands`: v1 schema with `metric FloatType`, 3 rows into `dt='2024-01-01'` (Overwrite); v2 with `metric DoubleType`, 3 rows into `dt='2024-01-02'` (Append); hive sync through the same `applyWriteOpts`; print a `..._SETUP_SUCCESS` marker. - `hudi-integ-test/.../integ2/testcontainers/trino/ITTestTrinoSchemaEvolution.java` on the `ITTestTrinoCustomType` skeleton: `SELECT count(*)` = 6 as the anchor; `SELECT id FROM evolved_test WHERE metric > 1.5 ORDER BY id` returns ids from both partitions (pre-fix: `HUDI_BAD_DATA`); `WHERE id > 4` as the positive control that pushdown is alive. </details> ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java: ########## @@ -0,0 +1,388 @@ +/* + * 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.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.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; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.predicate.ValueSet; +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 org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +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.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 io.trino.testing.MaterializedResult.materializeSourceDataStream; +import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS; +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> + * The fixture writes every data column as the type it had BEFORE the evolution and every handle carries the type the + * metastore reports AFTER it, which is exactly the state an unrewritten base file is in. 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, reported by the metastore as double. */ + private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double"; + /** Written as parquet INT32, reported by the metastore as bigint. */ + private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint"; + /** Written as parquet INT32, 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); + + @TempDir + static Path tempDir; + + private static Path baseFile; + + @BeforeAll + static void writeBaseFile() Review Comment: Every fixture in the PR holds exactly one base file, all of it pre-evolution, so no test covers the state a table is actually in after an evolution plus one insert: one split must drop the domain while the next must keep it and prune. Could `@BeforeAll` write a second file with `evolved_double` as DOUBLE and `evolved_bigint` as INT64, and a test read both splits, asserting the post-evolution one still prunes while the pre-evolution one reads everything? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSchemaEvolutionPredicates.java: ########## @@ -0,0 +1,98 @@ +/* + * 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.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.QueryRunner; +import org.junit.jupiter.api.Test; + +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.BIGINT_THRESHOLD; +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.TABLE_NAME; +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.VARCHAR_THRESHOLD; +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.expectedRowsFrom; + +/** + * 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 every test here has to put a predicate + * ON the evolved column -- and every projection has to include it, otherwise 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 that shows the read path itself was + * always fine. + * + * @see TestHudiSchemaEvolutionPredicatesPositional for the same suite with columns resolved by ordinal + */ +public class TestHudiSchemaEvolutionPredicates + extends AbstractTestQueryFramework +{ + @Override + protected QueryRunner createQueryRunner() Review Comment: `SchemaEvolutionHudiTablesInitializer`'s config and write methods are 24 of 25 lines identical to `OmittedMetaColumnsHudiTablesInitializer`, and this suite plus its `Positional` twin each start a query runner where the module's idiom is the composite loader: #19387's regression is one test on `TestHudiSmokeTest`, which `TestHudiConnectorParquetColumnNamesTest` already reruns positionally. The guard runs after the `useColumnNames` fork on descriptor keys, so the two modes cannot diverge here (with the fix reverted both suites fail the same three tests). Could we add the fixture to the smoke loader, keep the `float -> double` and `int -> varchar` cases there, and drop both `TestHudiSchemaEvolutionPredicates*` classes? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java: ########## @@ -0,0 +1,388 @@ +/* + * 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.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.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; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.predicate.ValueSet; +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 org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +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.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 io.trino.testing.MaterializedResult.materializeSourceDataStream; +import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS; +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> + * The fixture writes every data column as the type it had BEFORE the evolution and every handle carries the type the + * metastore reports AFTER it, which is exactly the state an unrewritten base file is in. 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, reported by the metastore as double. */ + private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double"; + /** Written as parquet INT32, reported by the metastore as bigint. */ + private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint"; + /** Written as parquet INT32, 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); + + @TempDir + static Path tempDir; + + private static Path baseFile; + + @BeforeAll + static void writeBaseFile() + throws IOException + { + MessageType schema = preEvolutionFileSchema(); + baseFile = tempDir.resolve("evolved_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); + } + 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); + writer.write(group); + } + } + // 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(rowGroupCount(baseFile)).as("row groups written").isGreaterThan(1); + } + + @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(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); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToVarchar(boolean useParquetColumnNames) + 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(projection, + greaterThanThreshold(evolved, VARCHAR, Slices.utf8Slice("900")), + useParquetColumnNames, 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"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToBigintStillPrunes(boolean useParquetColumnNames) + 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(projection, greaterThanThreshold(evolved, BIGINT, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + // asLong takes an Integer as happily as a Long, so this promotion is one the statistics CAN answer and the + // guard must leave it alone. This is what catches a check that drops more than it should. + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 1)).as("rows matching %s > %s after pruning", INT_TO_BIGINT_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnUnevolvedColumnStillPrunes(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER); + List<HiveColumnHandle> projection = List.of(stable); + + MaterializedResult result = read(projection, greaterThanThreshold(stable, INTEGER, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + 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 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(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(projection, TupleDomain.all(), false, + dynamicFilterOn(greaterThanThreshold(evolved, DOUBLE, (double) THRESHOLD))); + + assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT); + } + + /** + * 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( Review Comment: `read`, `dynamicFilterOn`, `rowGroupCount`, the constants and the `@BeforeAll` writer block are byte-for-byte the ones in `TestHudiPageSourceProviderTest` (about 90 identical lines), and `TestHudiSmokeTest.testTimestampMicros` hand-rolls the same `HudiSplit` + `createPageSource` block a third time. `createPageSource` is package-private, so a helper has to live in this package. Could we extract a `TestingBaseFilePageSource` (`read`/`dynamicFilterOn`/`rowGroupCount`/`writeBaseFile`) shared by all three, or fold these tests into `TestHudiPageSourceProviderTest` as a `@Nested` class? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java: ########## @@ -0,0 +1,388 @@ +/* + * 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.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.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; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.predicate.ValueSet; +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 org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +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.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 io.trino.testing.MaterializedResult.materializeSourceDataStream; +import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS; +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> + * The fixture writes every data column as the type it had BEFORE the evolution and every handle carries the type the + * metastore reports AFTER it, which is exactly the state an unrewritten base file is in. 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, reported by the metastore as double. */ + private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double"; + /** Written as parquet INT32, reported by the metastore as bigint. */ + private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint"; + /** Written as parquet INT32, 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); + + @TempDir + static Path tempDir; + + private static Path baseFile; + + @BeforeAll + static void writeBaseFile() + throws IOException + { + MessageType schema = preEvolutionFileSchema(); + baseFile = tempDir.resolve("evolved_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); + } + 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); + writer.write(group); + } + } + // 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(rowGroupCount(baseFile)).as("row groups written").isGreaterThan(1); + } + + @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(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); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToVarchar(boolean useParquetColumnNames) + 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(projection, + greaterThanThreshold(evolved, VARCHAR, Slices.utf8Slice("900")), + useParquetColumnNames, 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"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToBigintStillPrunes(boolean useParquetColumnNames) + 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(projection, greaterThanThreshold(evolved, BIGINT, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + // asLong takes an Integer as happily as a Long, so this promotion is one the statistics CAN answer and the + // guard must leave it alone. This is what catches a check that drops more than it should. + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 1)).as("rows matching %s > %s after pruning", INT_TO_BIGINT_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnUnevolvedColumnStillPrunes(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER); + List<HiveColumnHandle> projection = List.of(stable); + + MaterializedResult result = read(projection, greaterThanThreshold(stable, INTEGER, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + 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 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(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() Review Comment: nit, feel free to ignore: `rowCount == ROW_COUNT` also holds if the dynamic filter never reaches `getPushdownPredicate` at all, and `TupleDomain.all().intersect(df)` is the same input `testPredicateOnFloatColumnEvolvedToDouble` already sends. Could a sibling put the dynamic filter on `stable_int` and assert `rowCount < ROW_COUNT`, so this one is pinned to the filter actually being pushed down, or else be dropped? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java: ########## @@ -0,0 +1,294 @@ +/* + * 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.trino.parquet.ParquetDataSourceId; +import io.trino.parquet.predicate.TupleDomainParquetPredicate; +import io.trino.spi.predicate.Domain; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.type.DecimalType; +import io.trino.spi.type.Type; +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.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 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.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.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 method it exists to protect. Every case + * below states BOTH what the guard decides and what {@code TupleDomainParquetPredicate.getDomain} actually does with + * the same pair, and the check is run against the real {@code getDomain}, 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 two different reasons: + * <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} - the library declines the pair itself, so dropping it changes nothing.</li> + * </ul> + * The invariant that ties them together is asserted for every case: whatever the guard keeps must be a pair the + * library reads a real range out of. + */ +class TestParquetStatisticsDomains +{ + private static final ParquetDataSourceId DATA_SOURCE_ID = new ParquetDataSourceId("test"); + private static final long VALUE_COUNT = 10; + + 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) + { + @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 + new TypePair("boolean over BOOLEAN", BOOLEAN, plain(PrimitiveTypeName.BOOLEAN), true, NARROW), + new TypePair("integer over INT32", INTEGER, plain(INT32), true, NARROW), + new TypePair("bigint over INT64", BIGINT, plain(INT64), true, NARROW), + new TypePair("tinyint over INT32", TINYINT, plain(INT32), true, NARROW), + new TypePair("date over INT32 date", DATE, annotated(INT32, LogicalTypeAnnotation.dateType()), true, NARROW), + new TypePair("real over FLOAT", REAL, plain(FLOAT), true, NARROW), + new TypePair("double over DOUBLE", DOUBLE, plain(PrimitiveTypeName.DOUBLE), true, NARROW), + new TypePair("varchar over BINARY string", VARCHAR, annotated(BINARY, LogicalTypeAnnotation.stringType()), true, NARROW), + new TypePair("decimal(9,2) over INT32 decimal(9,2)", DecimalType.createDecimalType(9, 2), decimal(INT32, 9, 2), true, NARROW), + new TypePair("timestamp over INT64 timestamp", TIMESTAMP_MILLIS, annotated(INT64, LogicalTypeAnnotation.timestampType(false, TimeUnit.MILLIS)), true, NARROW), + new TypePair("timestamp over INT96", TIMESTAMP_MILLIS, plain(INT96), true, NARROW), + + // Promotions the statistics can answer, so pushdown must survive them + new TypePair("int -> long", BIGINT, plain(INT32), true, NARROW), + new TypePair("decimal(9,2) -> decimal(9,4)", DecimalType.createDecimalType(9, 4), decimal(INT32, 9, 2), true, NARROW), + new TypePair("decimal(20,2) -> decimal(38,4)", DecimalType.createDecimalType(38, 4), decimal(FIXED_LEN_BYTE_ARRAY, 20, 2), true, NARROW), + new TypePair("integer over a zero-scale INT32 decimal", INTEGER, decimal(INT32, 9, 0), true, NARROW), + + // Promotions that fail the split today: apache/hudi#19457 and its neighbours + new TypePair("float -> double", DOUBLE, plain(FLOAT), false, THROWS), + new TypePair("int -> double", DOUBLE, plain(INT32), false, THROWS), + new TypePair("long -> double", DOUBLE, plain(INT64), false, THROWS), + new TypePair("int -> float", REAL, plain(INT32), false, THROWS), + new TypePair("int -> string", VARCHAR, plain(INT32), false, THROWS), + new TypePair("long -> string", VARCHAR, plain(INT64), false, THROWS), Review Comment: nit, feel free to ignore: the four `* -> string` and three `* -> double` rows all pin the same two cast lines, while three pairs the guard keeps are unpinned and are the ones a tightening would silently break: `varchar` over a plain FIXED_LEN_BYTE_ARRAY, `decimal` over a BINARY decimal, and `bigint` over INT64 TIMESTAMP(MICROS), which is what every table synced with `hive_sync.support_timestamp=false` carries (`HiveSchemaUtil.convertField` maps TIMESTAMP to BIGINT). Could those three join the table, and the `decimal(9,2) -> decimal(9,4)` row be relabelled, since Hudi rejects that evolution as lossy? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java: ########## @@ -0,0 +1,294 @@ +/* + * 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.trino.parquet.ParquetDataSourceId; +import io.trino.parquet.predicate.TupleDomainParquetPredicate; +import io.trino.spi.predicate.Domain; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.type.DecimalType; +import io.trino.spi.type.Type; +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.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 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.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.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 method it exists to protect. Every case + * below states BOTH what the guard decides and what {@code TupleDomainParquetPredicate.getDomain} actually does with + * the same pair, and the check is run against the real {@code getDomain}, 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 two different reasons: + * <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} - the library declines the pair itself, so dropping it changes nothing.</li> + * </ul> + * The invariant that ties them together is asserted for every case: whatever the guard keeps must be a pair the + * library reads a real range out of. + */ +class TestParquetStatisticsDomains +{ + private static final ParquetDataSourceId DATA_SOURCE_ID = new ParquetDataSourceId("test"); + private static final long VALUE_COUNT = 10; + + 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) + { + @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 + new TypePair("boolean over BOOLEAN", BOOLEAN, plain(PrimitiveTypeName.BOOLEAN), true, NARROW), + new TypePair("integer over INT32", INTEGER, plain(INT32), true, NARROW), + new TypePair("bigint over INT64", BIGINT, plain(INT64), true, NARROW), + new TypePair("tinyint over INT32", TINYINT, plain(INT32), true, NARROW), + new TypePair("date over INT32 date", DATE, annotated(INT32, LogicalTypeAnnotation.dateType()), true, NARROW), + new TypePair("real over FLOAT", REAL, plain(FLOAT), true, NARROW), + new TypePair("double over DOUBLE", DOUBLE, plain(PrimitiveTypeName.DOUBLE), true, NARROW), + new TypePair("varchar over BINARY string", VARCHAR, annotated(BINARY, LogicalTypeAnnotation.stringType()), true, NARROW), + new TypePair("decimal(9,2) over INT32 decimal(9,2)", DecimalType.createDecimalType(9, 2), decimal(INT32, 9, 2), true, NARROW), + new TypePair("timestamp over INT64 timestamp", TIMESTAMP_MILLIS, annotated(INT64, LogicalTypeAnnotation.timestampType(false, TimeUnit.MILLIS)), true, NARROW), + new TypePair("timestamp over INT96", TIMESTAMP_MILLIS, plain(INT96), true, NARROW), + + // Promotions the statistics can answer, so pushdown must survive them + new TypePair("int -> long", BIGINT, plain(INT32), true, NARROW), + new TypePair("decimal(9,2) -> decimal(9,4)", DecimalType.createDecimalType(9, 4), decimal(INT32, 9, 2), true, NARROW), + new TypePair("decimal(20,2) -> decimal(38,4)", DecimalType.createDecimalType(38, 4), decimal(FIXED_LEN_BYTE_ARRAY, 20, 2), true, NARROW), + new TypePair("integer over a zero-scale INT32 decimal", INTEGER, decimal(INT32, 9, 0), true, NARROW), + + // Promotions that fail the split today: apache/hudi#19457 and its neighbours + new TypePair("float -> double", DOUBLE, plain(FLOAT), false, THROWS), + new TypePair("int -> double", DOUBLE, plain(INT32), false, THROWS), + new TypePair("long -> double", DOUBLE, plain(INT64), false, THROWS), + new TypePair("int -> float", REAL, plain(INT32), false, THROWS), + new TypePair("int -> string", VARCHAR, plain(INT32), false, THROWS), + new TypePair("long -> string", VARCHAR, plain(INT64), false, THROWS), + new TypePair("float -> string", VARCHAR, plain(FLOAT), false, THROWS), + new TypePair("double -> string", VARCHAR, plain(PrimitiveTypeName.DOUBLE), false, THROWS), + new TypePair("string -> date", DATE, annotated(BINARY, LogicalTypeAnnotation.stringType()), false, THROWS), + + // Promotions that silently prune on a comparison that means nothing, which is why the guard cannot + // be a try/catch around the cast + new TypePair("decimal -> string", VARCHAR, decimal(FIXED_LEN_BYTE_ARRAY, 20, 2), false, NARROW), + new TypePair("string -> decimal", DecimalType.createDecimalType(9, 2), annotated(BINARY, LogicalTypeAnnotation.stringType()), false, NARROW), + new TypePair("int -> decimal", DecimalType.createDecimalType(9, 2), plain(INT32), false, NARROW), + new TypePair("integer over a scaled INT32 decimal", INTEGER, decimal(INT32, 9, 2), false, NARROW), + + // Pairs the library declines on its own, where dropping costs nothing + new TypePair("timestamp over an unannotated INT64", TIMESTAMP_MILLIS, plain(INT64), false, ALL), + new TypePair("varbinary over BINARY", VARBINARY, plain(BINARY), false, 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()) Review Comment: nit, feel free to ignore: `isAll()` is false for a range that is narrow but wrong, which is the failure mode the class Javadoc calls worse than a crash. For the cross-type kept rows (the two decimal rescales, `int -> long`, integer over a zero-scale decimal) could the test assert the actual bounds, e.g. `decimal(20,2)` min=1 read as `decimal(38,4)` must be `100`, so a rescale regression fails instead of passing as "narrow"? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/ParquetStatisticsDomains.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.DecimalType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.Type; +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. Review Comment: trinodb/trino#30545 (open) moves this same check into `lib/trino-parquet`, bloom path included, and keeps `float -> double` pruning instead of dropping it. hudi-trino now tracks a trino master SHA (#19642), so the day the pin passes that PR every `THROWS` row in `TestParquetStatisticsDomains` becomes `NARROW` and the build goes red with no pointer to why. Could the Javadoc say this guard is a stopgap for the pinned Trino and name #30545 as its removal trigger? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java: ########## @@ -0,0 +1,388 @@ +/* + * 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.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.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; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.predicate.ValueSet; +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 org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +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.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 io.trino.testing.MaterializedResult.materializeSourceDataStream; +import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS; +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> + * The fixture writes every data column as the type it had BEFORE the evolution and every handle carries the type the + * metastore reports AFTER it, which is exactly the state an unrewritten base file is in. 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, reported by the metastore as double. */ + private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double"; + /** Written as parquet INT32, reported by the metastore as bigint. */ + private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint"; + /** Written as parquet INT32, 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); + + @TempDir + static Path tempDir; + + private static Path baseFile; + + @BeforeAll + static void writeBaseFile() + throws IOException + { + MessageType schema = preEvolutionFileSchema(); + baseFile = tempDir.resolve("evolved_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); + } + 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); + writer.write(group); + } + } + // 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(rowGroupCount(baseFile)).as("row groups written").isGreaterThan(1); + } + + @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(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); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToVarchar(boolean useParquetColumnNames) + 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(projection, + greaterThanThreshold(evolved, VARCHAR, Slices.utf8Slice("900")), + useParquetColumnNames, 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"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToBigintStillPrunes(boolean useParquetColumnNames) + 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(projection, greaterThanThreshold(evolved, BIGINT, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + // asLong takes an Integer as happily as a Long, so this promotion is one the statistics CAN answer and the + // guard must leave it alone. This is what catches a check that drops more than it should. + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 1)).as("rows matching %s > %s after pruning", INT_TO_BIGINT_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnUnevolvedColumnStillPrunes(boolean useParquetColumnNames) Review Comment: nit, feel free to ignore: this method's two assertions are the same two `testOnlyTheEvolvedColumnsDomainIsDropped` makes at lines 226-227 on a strictly stronger setup (stable domain kept alongside a dropped one), and the `@ValueSource` axis on these four tests is inert: `column()` builds every handle on its physical ordinal, so both modes hand the guard the identical descriptor. Could this test go, and the boolean parameter stay on one test as the both-modes anchor? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSchemaEvolutionPredicates.java: ########## @@ -0,0 +1,98 @@ +/* + * 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.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.QueryRunner; +import org.junit.jupiter.api.Test; + +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.BIGINT_THRESHOLD; +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.TABLE_NAME; +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.VARCHAR_THRESHOLD; +import static io.trino.plugin.hudi.testing.SchemaEvolutionHudiTablesInitializer.expectedRowsFrom; + +/** + * 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 every test here has to put a predicate + * ON the evolved column -- and every projection has to include it, otherwise 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 that shows the read path itself was + * always fine. + * + * @see TestHudiSchemaEvolutionPredicatesPositional for the same suite with columns resolved by ordinal + */ +public class TestHudiSchemaEvolutionPredicates + extends AbstractTestQueryFramework +{ + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .addConnectorProperty("hudi.parquet.use-column-names", "true") + .setDataLoader(new SchemaEvolutionHudiTablesInitializer()) + .build(); + } + + @Test + public void testPredicateOnColumnEvolvedFromFloatToDouble() + { + assertQuery(selectWhere(FLOAT_TO_DOUBLE_COLUMN + " > " + DOUBLE_THRESHOLD), expectedRowsFrom(3)); + } + + @Test + public void testPredicateOnColumnEvolvedFromIntToBigint() Review Comment: nit, feel free to ignore: this test and `testPredicatesOnEvolvedAndUnevolvedColumnsTogether` pass whether the domain is kept or dropped. `applyFilter` returns the whole regular predicate as the remaining filter and the fixture is 5 rows in one row group, so `assertQuery` cannot observe pruning; the discriminating half of the "together" case is the `float_value` conjunct already covered at line 58. Could these two be dropped, leaving `TestHudiEvolvedColumnPredicates:181` as the one that pins `int -> bigint` staying kept? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
