fix(trino): skip predicate pushdown on type-evolved parquet columns - #19467
fix(trino): skip predicate pushdown on type-evolved parquet columns#19467wombatu-kun wants to merge 2 commits into
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19467 +/- ##
============================================
+ Coverage 76.39% 76.40% +0.01%
- Complexity 32101 32109 +8
============================================
Files 2520 2520
Lines 138985 138995 +10
Branches 16695 16699 +4
============================================
+ Hits 106171 106195 +24
+ Misses 25178 25168 -10
+ Partials 7636 7632 -4
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR fixes a Trino query failure (#19457) where a predicate on a type-widened column crashed the whole split, by dropping domains from the parquet statistics pushdown when the file's physical type cannot answer them, while the engine still re-applies the predicate above the scan. I verified the load-bearing soundness assumption directly in the code — HudiMetadata.applyFilter returns the full regular predicate as the unenforced remaining filter, so dropping a domain only costs row-group pruning and never a row — and the new guard is pinned against the real TupleDomainParquetPredicate.getDomain in unit tests. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here.
. Code looks clean overall — one minor readability nit below.
cc @yihua
| return Types.primitive(primitiveTypeName, OPTIONAL).as(annotation).named("c"); | ||
| } | ||
|
|
||
| private static PrimitiveType decimal(PrimitiveTypeName primitiveTypeName, int precision, int scale) |
There was a problem hiding this 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?
|
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. |
0b33546 to
dac69a5
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR adds a guard (ParquetStatisticsDomains.hasComparableStatistics / dropIncomparableDomains) that keeps a metastore-typed pushdown domain from being matched against parquet statistics of a column still physically stored under its pre-evolution type, which previously failed queries with Corrupted statistics for column .... I verified the guard branch-by-branch against Trino 481's actual TupleDomainParquetPredicate.getDomain dispatch (extracted from the trino-parquet sources): every kept (domainType, fileType) pair is one getDomain reads correctly without throwing (including the long-decimal-over-INT32/INT64 rescale path, which uses asLong and does not throw), and every dropped pair is one getDomain would either throw or misprune on — so there are no false positives and no pruning regression on unevolved data. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One dead-code branch in a test helper worth cleaning up; otherwise the code is clean and well-documented.
cc @yihua
|
|
||
| private static PrimitiveType plain(PrimitiveTypeName primitiveTypeName) | ||
| { | ||
| if (primitiveTypeName == INT96) { |
There was a problem hiding this 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.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR adds a connector-side guard (ParquetStatisticsDomains.dropIncomparableDomains) that removes pushed-down domains whose metastore type can't be compared against a base file's pre-evolution physical statistics, fixing the Corrupted statistics/ClassCastException failure on type-evolved columns. The change is correctness-safe by construction: it only ever returns a subset of the existing parquet tuple domain, so it can never prune more row groups or introduce new data loss, and the hasComparableStatistics dispatch is pinned against the real TupleDomainParquetPredicate.getDomain for the risky pairs. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
5 participants
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:
TupleDomainParquetPredicate.getDomainpicks 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.getCompatibilitypermitsdouble <- float,float <- int|long,long <- intandstring <- any numeric, andhoodie.avro.schema.validatedefaults tofalse.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):hasComparableStatisticsmirrorsgetDomain's dispatch,dropIncomparableDomainsfilters 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 bothhudi.parquet.use-column-namesmodes in one pass.HudiPageSourceProvider.createPageSource: wraps the existinggetParquetTupleDomaincall, which feeds bothbuildPredicateandgetFilteredRowGroups.HudiMetadata.applyFilterreturns the whole regular predicate as the remaining filter, so pushdown here is an optimization only. It is the traderemapPredicateColumnIndicesToPhysicalandHudiColumnStatsIndexSupport.getDomainFromColumnStatsalready make.Tests, where
hudi-trinohad no schema evolution coverage at all:TestParquetStatisticsDomains(30 type pairs, each checked against the realgetDomain),TestHudiEvolvedColumnPredicates(reads a pre-evolution base file throughcreatePageSource), andTestHudiSchemaEvolutionPredicateswith 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, includingint -> 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