You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "wombatu-kun (via GitHub)" <gi...@apache.org> on 2026/08/03 06:13:56 UTC
[PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
wombatu-kun opened a new pull request, #19467:
URL: https://github.com/apache/hudi/pull/19467
### Describe the issue this Pull Request addresses
Closes #19457
A predicate on a column whose type was widened after a base file was written fails the query, for as long as any base file still stores the original physical type:
```
Malformed Parquet file. Corrupted statistics for column "[c7] optional float c7"
Caused by: java.lang.ClassCastException: class java.lang.Float cannot be cast to class java.lang.Double
```
`TupleDomainParquetPredicate.getDomain` picks its branch from the pushed-down domain's type and reads the parquet statistics as that type. The domain's type comes from the metastore, the statistics come from the file, and a type evolution is what makes the two disagree. Reading such a column always worked; only the statistics side was blind.
This is schema-on-write, not schema-on-read: `HoodieSchemaCompatibilityChecker.getCompatibility` permits `double <- float`, `float <- int|long`, `long <- int` and `string <- any numeric`, and `hoodie.avro.schema.validate` defaults to `false`.
### Summary and Changelog
The connector now leaves a domain out of the parquet predicate when the file's physical type cannot answer it. The engine still applies that predicate above the scan, so failing queries return their rows, and the only loss is row group pruning on the evolved column, for files written before the evolution.
- `ParquetStatisticsDomains` (new): `hasComparableStatistics` mirrors `getDomain`'s dispatch, `dropIncomparableDomains` filters the descriptor-keyed tuple domain. Filtering on the descriptor rather than on the handle is what stops the check and the evaluation disagreeing about which column and type is meant, and covers both `hudi.parquet.use-column-names` modes in one pass.
- `HudiPageSourceProvider.createPageSource`: wraps the existing `getParquetTupleDomain` call, which feeds both `buildPredicate` and `getFilteredRowGroups`.
- Dropping is sound because `HudiMetadata.applyFilter` returns the whole regular predicate as the remaining filter, so pushdown here is an optimization only. It is the trade `remapPredicateColumnIndicesToPhysical` and `HudiColumnStatsIndexSupport.getDomainFromColumnStats` already make.
- Two rejected pairs are silent wrong answers rather than crashes, a decimal column read as varchar and a string column read as decimal, which is why this is a type check and not a try/catch.
Tests, where `hudi-trino` had no schema evolution coverage at all: `TestParquetStatisticsDomains` (30 type pairs, each checked against the real `getDomain`), `TestHudiEvolvedColumnPredicates` (reads a pre-evolution base file through `createPageSource`), and `TestHudiSchemaEvolutionPredicates` with its positional subclass at SQL level. Both column-resolution modes throughout.
### Impact
Queries with a predicate on a type-evolved column succeed instead of failing with `HUDI_BAD_DATA`. Pruning is unchanged everywhere else, including `int -> long`, which the statistics can answer and the guard keeps. Pushdown is only enabled for base-file-only splits, so the merge path is untouched. No config or public API change.
### Risk Level
low
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5199946004
@voonhous #19217 is merged now and this branch is rebased on top of it. Could you take a look when you get a chance?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5201259583
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5377640438
Sorry, i will prioritize this next week.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859248401
##########
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:
Done 44c2344b18e4. Pinned end to end rather than on `checkInBloomFilter` directly, so the test survives the pin advancing past trinodb/trino#30545: with the pair kept, an equality predicate on the evolved column reads 0 of 1000 rows.
##########
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:
Done 44c2344b18e4.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3702095179
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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())
+ .as("getDomain for %s returned %s", pair, domain)
+ .isEqualTo(pair.outcome() == ALL);
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testEveryKeptPairYieldsUsableStatistics(TypePair pair)
+ {
+ // The invariant behind the whole table: a false negative only costs pruning, but a false positive is either
+ // a failed query or a wrong one, so nothing may be kept that the library does not read a real range out of.
+ if (hasComparableStatistics(pair.domainType(), pair.fileType())) {
+ assertThat(pair.outcome()).as("library outcome for the kept pair %s", pair).isEqualTo(NARROW);
+ }
+ }
+
+ @Test
+ public void testAllAndNonePassThroughUntouched()
+ {
+ assertThat(dropIncomparableDomains(TupleDomain.all())).isEqualTo(TupleDomain.all());
+ assertThat(dropIncomparableDomains(TupleDomain.none())).isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testOnlyTheIncomparableDomainIsDropped()
+ {
+ // Distinct names on purpose: a ColumnDescriptor is keyed by its path, so two columns sharing one name would
+ // collapse into a single map entry and the test would pass without ever exercising the filtering
+ ColumnDescriptor evolved = descriptorNamed("evolved", FLOAT);
+ ColumnDescriptor stable = descriptorNamed("stable", INT32);
+ Domain doubleDomain = Domain.singleValue(DOUBLE, 1.0d);
+ Domain intDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<ColumnDescriptor> filtered = dropIncomparableDomains(
+ TupleDomain.withColumnDomains(Map.of(evolved, doubleDomain, stable, intDomain)));
+
+ assertThat(filtered.getDomains().orElseThrow()).containsExactly(Map.entry(stable, intDomain));
+ }
+
+ @Test
+ public void testAPredicateWithNothingToDropIsReturnedAsIs()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(INT32)), Domain.singleValue(INTEGER, 1L)));
+
+ assertThat(dropIncomparableDomains(predicate)).isSameAs(predicate);
+ }
+
+ @Test
+ public void testDroppingEveryDomainLeavesAnUnconstrainedPredicate()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(FLOAT)), Domain.singleValue(DOUBLE, 1.0d)));
+
+ // Not TupleDomain.none(): dropping means "do not prune on this", never "this matches nothing"
+ assertThat(dropIncomparableDomains(predicate).isAll()).as("everything dropped").isTrue();
+ }
+
+ private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName)
+ {
+ if (primitiveTypeName == INT96) {
+ return Types.primitive(primitiveTypeName, OPTIONAL).named("c");
+ }
+ if (primitiveTypeName == FIXED_LEN_BYTE_ARRAY) {
+ return Types.primitive(primitiveTypeName, OPTIONAL).length(16).named("c");
+ }
+ return Types.primitive(primitiveTypeName, OPTIONAL).named("c");
+ }
+
+ private static PrimitiveType annotated(PrimitiveTypeName primitiveTypeName, LogicalTypeAnnotation annotation)
+ {
+ return Types.primitive(primitiveTypeName, OPTIONAL).as(annotation).named("c");
+ }
+
+ private static PrimitiveType decimal(PrimitiveTypeName primitiveTypeName, int precision, int scale)
Review Comment:
π€ nit: the `INT96` branch in `plain()` returns exactly the same expression as the fallthrough, so it doesn't change behaviour but does invite a reader to wonder whether INT96 needs special treatment here. Could you remove it, leaving only the `FIXED_LEN_BYTE_ARRAY` guard and the default?
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859251985
##########
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:
Done 44c2344b18e4.
##########
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:
Done 44c2344b18e4. Kept the boolean parameter on `testPredicateOnFloatColumnEvolvedToDouble` as the both-modes anchor.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5199796605
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
* dac69a5aeff9c2a1b349ca908b9398493a197019 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859253807
##########
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:
Done 44c2344b18e4.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859250209
##########
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:
Done 44c2344b18e4. The second file also carries the bloom filters, so it doubles as the control that a post-evolution column still prunes on the same predicate the pre-evolution one cannot.
##########
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:
Done 44c2344b18e4.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous merged PR #19467:
URL: https://github.com/apache/hudi/pull/19467
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5421745371
Pushed 31ba4e27cd15 onto the branch: the type-table comment named a test that does not exist, it now points at `testEqualityOnAnIntColumnEvolvedToBigintStillFindsItsRows`. All threads addressed, E2E tracked in #19743.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3725570245
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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())
+ .as("getDomain for %s returned %s", pair, domain)
+ .isEqualTo(pair.outcome() == ALL);
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testEveryKeptPairYieldsUsableStatistics(TypePair pair)
+ {
+ // The invariant behind the whole table: a false negative only costs pruning, but a false positive is either
+ // a failed query or a wrong one, so nothing may be kept that the library does not read a real range out of.
+ if (hasComparableStatistics(pair.domainType(), pair.fileType())) {
+ assertThat(pair.outcome()).as("library outcome for the kept pair %s", pair).isEqualTo(NARROW);
+ }
+ }
+
+ @Test
+ public void testAllAndNonePassThroughUntouched()
+ {
+ assertThat(dropIncomparableDomains(TupleDomain.all())).isEqualTo(TupleDomain.all());
+ assertThat(dropIncomparableDomains(TupleDomain.none())).isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testOnlyTheIncomparableDomainIsDropped()
+ {
+ // Distinct names on purpose: a ColumnDescriptor is keyed by its path, so two columns sharing one name would
+ // collapse into a single map entry and the test would pass without ever exercising the filtering
+ ColumnDescriptor evolved = descriptorNamed("evolved", FLOAT);
+ ColumnDescriptor stable = descriptorNamed("stable", INT32);
+ Domain doubleDomain = Domain.singleValue(DOUBLE, 1.0d);
+ Domain intDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<ColumnDescriptor> filtered = dropIncomparableDomains(
+ TupleDomain.withColumnDomains(Map.of(evolved, doubleDomain, stable, intDomain)));
+
+ assertThat(filtered.getDomains().orElseThrow()).containsExactly(Map.entry(stable, intDomain));
+ }
+
+ @Test
+ public void testAPredicateWithNothingToDropIsReturnedAsIs()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(INT32)), Domain.singleValue(INTEGER, 1L)));
+
+ assertThat(dropIncomparableDomains(predicate)).isSameAs(predicate);
+ }
+
+ @Test
+ public void testDroppingEveryDomainLeavesAnUnconstrainedPredicate()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(FLOAT)), Domain.singleValue(DOUBLE, 1.0d)));
+
+ // Not TupleDomain.none(): dropping means "do not prune on this", never "this matches nothing"
+ assertThat(dropIncomparableDomains(predicate).isAll()).as("everything dropped").isTrue();
+ }
+
+ private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName)
+ {
+ if (primitiveTypeName == INT96) {
Review Comment:
π€ nit: the `INT96` branch returns the same expression as the fallthrough, so it's dead code β could you remove it? A future reader will wonder what was supposed to be different about INT96.
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag quality.</i></sub>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3725626859
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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())
+ .as("getDomain for %s returned %s", pair, domain)
+ .isEqualTo(pair.outcome() == ALL);
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testEveryKeptPairYieldsUsableStatistics(TypePair pair)
+ {
+ // The invariant behind the whole table: a false negative only costs pruning, but a false positive is either
+ // a failed query or a wrong one, so nothing may be kept that the library does not read a real range out of.
+ if (hasComparableStatistics(pair.domainType(), pair.fileType())) {
+ assertThat(pair.outcome()).as("library outcome for the kept pair %s", pair).isEqualTo(NARROW);
+ }
+ }
+
+ @Test
+ public void testAllAndNonePassThroughUntouched()
+ {
+ assertThat(dropIncomparableDomains(TupleDomain.all())).isEqualTo(TupleDomain.all());
+ assertThat(dropIncomparableDomains(TupleDomain.none())).isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testOnlyTheIncomparableDomainIsDropped()
+ {
+ // Distinct names on purpose: a ColumnDescriptor is keyed by its path, so two columns sharing one name would
+ // collapse into a single map entry and the test would pass without ever exercising the filtering
+ ColumnDescriptor evolved = descriptorNamed("evolved", FLOAT);
+ ColumnDescriptor stable = descriptorNamed("stable", INT32);
+ Domain doubleDomain = Domain.singleValue(DOUBLE, 1.0d);
+ Domain intDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<ColumnDescriptor> filtered = dropIncomparableDomains(
+ TupleDomain.withColumnDomains(Map.of(evolved, doubleDomain, stable, intDomain)));
+
+ assertThat(filtered.getDomains().orElseThrow()).containsExactly(Map.entry(stable, intDomain));
+ }
+
+ @Test
+ public void testAPredicateWithNothingToDropIsReturnedAsIs()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(INT32)), Domain.singleValue(INTEGER, 1L)));
+
+ assertThat(dropIncomparableDomains(predicate)).isSameAs(predicate);
+ }
+
+ @Test
+ public void testDroppingEveryDomainLeavesAnUnconstrainedPredicate()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(FLOAT)), Domain.singleValue(DOUBLE, 1.0d)));
+
+ // Not TupleDomain.none(): dropping means "do not prune on this", never "this matches nothing"
+ assertThat(dropIncomparableDomains(predicate).isAll()).as("everything dropped").isTrue();
+ }
+
+ private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName)
+ {
+ if (primitiveTypeName == INT96) {
+ return Types.primitive(primitiveTypeName, OPTIONAL).named("c");
+ }
+ if (primitiveTypeName == FIXED_LEN_BYTE_ARRAY) {
+ return Types.primitive(primitiveTypeName, OPTIONAL).length(16).named("c");
+ }
+ return Types.primitive(primitiveTypeName, OPTIONAL).named("c");
+ }
+
+ private static PrimitiveType annotated(PrimitiveTypeName primitiveTypeName, LogicalTypeAnnotation annotation)
+ {
+ return Types.primitive(primitiveTypeName, OPTIONAL).as(annotation).named("c");
+ }
+
+ private static PrimitiveType decimal(PrimitiveTypeName primitiveTypeName, int precision, int scale)
Review Comment:
Done cd48bb14dbe4
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5200082345
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
* dac69a5aeff9c2a1b349ca908b9398493a197019 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116)
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5166655917
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3725628080
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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())
+ .as("getDomain for %s returned %s", pair, domain)
+ .isEqualTo(pair.outcome() == ALL);
+ }
+
+ @ParameterizedTest
+ @MethodSource("typePairs")
+ public void testEveryKeptPairYieldsUsableStatistics(TypePair pair)
+ {
+ // The invariant behind the whole table: a false negative only costs pruning, but a false positive is either
+ // a failed query or a wrong one, so nothing may be kept that the library does not read a real range out of.
+ if (hasComparableStatistics(pair.domainType(), pair.fileType())) {
+ assertThat(pair.outcome()).as("library outcome for the kept pair %s", pair).isEqualTo(NARROW);
+ }
+ }
+
+ @Test
+ public void testAllAndNonePassThroughUntouched()
+ {
+ assertThat(dropIncomparableDomains(TupleDomain.all())).isEqualTo(TupleDomain.all());
+ assertThat(dropIncomparableDomains(TupleDomain.none())).isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testOnlyTheIncomparableDomainIsDropped()
+ {
+ // Distinct names on purpose: a ColumnDescriptor is keyed by its path, so two columns sharing one name would
+ // collapse into a single map entry and the test would pass without ever exercising the filtering
+ ColumnDescriptor evolved = descriptorNamed("evolved", FLOAT);
+ ColumnDescriptor stable = descriptorNamed("stable", INT32);
+ Domain doubleDomain = Domain.singleValue(DOUBLE, 1.0d);
+ Domain intDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<ColumnDescriptor> filtered = dropIncomparableDomains(
+ TupleDomain.withColumnDomains(Map.of(evolved, doubleDomain, stable, intDomain)));
+
+ assertThat(filtered.getDomains().orElseThrow()).containsExactly(Map.entry(stable, intDomain));
+ }
+
+ @Test
+ public void testAPredicateWithNothingToDropIsReturnedAsIs()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(INT32)), Domain.singleValue(INTEGER, 1L)));
+
+ assertThat(dropIncomparableDomains(predicate)).isSameAs(predicate);
+ }
+
+ @Test
+ public void testDroppingEveryDomainLeavesAnUnconstrainedPredicate()
+ {
+ TupleDomain<ColumnDescriptor> predicate = TupleDomain.withColumnDomains(
+ Map.of(descriptorOf(plain(FLOAT)), Domain.singleValue(DOUBLE, 1.0d)));
+
+ // Not TupleDomain.none(): dropping means "do not prune on this", never "this matches nothing"
+ assertThat(dropIncomparableDomains(predicate).isAll()).as("everything dropped").isTrue();
+ }
+
+ private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName)
+ {
+ if (primitiveTypeName == INT96) {
Review Comment:
Done cd48bb14dbe4
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5421904980
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
}, {
"hash" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559",
"triggerID" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"triggerType" : "PUSH"
}, {
"hash" : "31ba4e27cd15ca00c222b2054254dfb126dac2c9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16567",
"triggerID" : "31ba4e27cd15ca00c222b2054254dfb126dac2c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44c2344b18e42b4e3b45f0296ec709ca714a1ede Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559)
* 31ba4e27cd15ca00c222b2054254dfb126dac2c9 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16567)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859249348
##########
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:
Done 44c2344b18e4.
##########
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:
Filed as #19743, with the `.commands` fixture and the `ITTestTrino*` skeleton sketched out; `hudi_trino_e2e.yml` already selects it by pattern, so no workflow change. The fixture Javadoc now names both producing routes and points there 44c2344b18e4.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5199764193
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
* dac69a5aeff9c2a1b349ca908b9398493a197019 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5200051903
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
* dac69a5aeff9c2a1b349ca908b9398493a197019 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116)
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5162995038
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5414776878
Apologies, I'll review this. tomorrow morning!
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859251154
##########
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:
Done 44c2344b18e4, as `TestingBaseFilePageSource` shared by all three call sites.
##########
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:
Done 44c2344b18e4. Left the initializer itself in place: `configureTableConfig` and `configureWriteConfig` are abstract in `AbstractMergerHudiTablesInitializer` with 12 identical overrides, so folding them up is a module-wide change rather than this PR's.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5420000470
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
}, {
"hash" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120)
* 44c2344b18e42b4e3b45f0296ec709ca714a1ede UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5420039372
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
}, {
"hash" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559",
"triggerID" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120)
* 44c2344b18e42b4e3b45f0296ec709ca714a1ede Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19467:
URL: https://github.com/apache/hudi/pull/19467#discussion_r3859252892
##########
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:
Done 44c2344b18e4, as a sibling on `stable_int` asserting fewer rows read.
##########
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:
Done 44c2344b18e4.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5421845767
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
}, {
"hash" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559",
"triggerID" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"triggerType" : "PUSH"
}, {
"hash" : "31ba4e27cd15ca00c222b2054254dfb126dac2c9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "31ba4e27cd15ca00c222b2054254dfb126dac2c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44c2344b18e42b4e3b45f0296ec709ca714a1ede Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559)
* 31ba4e27cd15ca00c222b2054254dfb126dac2c9 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5200963365
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dac69a5aeff9c2a1b349ca908b9398493a197019 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116)
* cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5163034998
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0b33546c5dc491de57cc7265a46b8ec6e98a5b03 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5163258400
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/19467?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 71.82%. Comparing base ([`64afac0`](https://app.codecov.io/gh/apache/hudi/commit/64afac03908bba1f32df05982aee388d9011b714?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0b33546`](https://app.codecov.io/gh/apache/hudi/commit/0b33546c5dc491de57cc7265a46b8ec6e98a5b03?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 4 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19467 +/- ##
============================================
- Coverage 76.81% 71.82% -5.00%
+ Complexity 33770 32005 -1765
============================================
Files 2576 2576
Lines 143457 143457
Branches 17589 17589
============================================
- Hits 110201 103036 -7165
- Misses 25010 32787 +7777
+ Partials 8246 7634 -612
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.52% <ΓΈ> (-0.75%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `80.70% <ΓΈ> (-1.15%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.96% <ΓΈ> (ΓΈ)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `48.96% <ΓΈ> (-25.27%)` | :arrow_down: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.66% <ΓΈ> (ΓΈ)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `15.32% <ΓΈ> (ΓΈ)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.80% <ΓΈ> (-0.70%)` | :arrow_down: |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.82% <ΓΈ> (-0.06%)` | :arrow_down: |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.60% <ΓΈ> (ΓΈ)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.23% <ΓΈ> (ΓΈ)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `64.00% <ΓΈ> (ΓΈ)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/19467/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ΓΈ> (ΓΈ)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [flink-integration-tests](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `48.80% <ΓΈ> (+<0.01%)` | :arrow_up: |
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.42% <ΓΈ> (-0.01%)` | :arrow_down: |
| [integration-tests](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `13.58% <ΓΈ> (-0.01%)` | :arrow_down: |
| [spark-client-hadoop-common](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `48.68% <ΓΈ> (-0.01%)` | :arrow_down: |
| [spark-scala-tests](https://app.codecov.io/gh/apache/hudi/pull/19467/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `32.84% <ΓΈ> (-14.38%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 288 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/19467/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5163579102
I'll hold off on review for this until the E2E branch is merged in. I'm trying to think of a way where we can clearly segregate the schema-on-write and schema-on-read tests, while also testing this E2E with a Spark-write and a Trino-read.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(trino): skip predicate pushdown on type-evolved parquet columns [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19467:
URL: https://github.com/apache/hudi/pull/19467#issuecomment-5420976790
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15962",
"triggerID" : "0b33546c5dc491de57cc7265a46b8ec6e98a5b03",
"triggerType" : "PUSH"
}, {
"hash" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16116",
"triggerID" : "dac69a5aeff9c2a1b349ca908b9398493a197019",
"triggerType" : "PUSH"
}, {
"hash" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16120",
"triggerID" : "cd48bb14dbe4a917c82e768ad8364b64e0c3ff4f",
"triggerType" : "PUSH"
}, {
"hash" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559",
"triggerID" : "44c2344b18e42b4e3b45f0296ec709ca714a1ede",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44c2344b18e42b4e3b45f0296ec709ca714a1ede Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=16559)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org