This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 64afac03908b fix(trino): remap pushed-down predicate columns to
physical file ordinals (#19456)
64afac03908b is described below
commit 64afac03908bba1f32df05982aee388d9011b714
Author: Vova Kolmakov <[email protected]>
AuthorDate: Sun Aug 2 21:03:12 2026 +0700
fix(trino): remap pushed-down predicate columns to physical file ordinals
(#19456)
* fix(trino): remap pushed-down predicate columns to physical file ordinals
* fix(trino): dedupe pushed-down predicate domains per projected field
* test(trino): cover stale metastore ordinals and merge-path pushdown
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../trino/plugin/hudi/HudiPageSourceProvider.java | 189 ++++++-
.../TestHudiConnectorParquetColumnNamesTest.java | 41 +-
.../plugin/hudi/TestHudiMorMergeModeSemantics.java | 21 +
.../hudi/TestHudiPageSourceProviderTest.java | 558 +++++++++++++++++++++
.../AbstractMergerHudiTablesInitializer.java | 12 +-
.../OmittedMetaColumnsHudiTablesInitializer.java | 157 ++++++
6 files changed, 949 insertions(+), 29 deletions(-)
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
index 7ffca56717ef..2803f8ae231a 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
@@ -34,6 +34,7 @@ import io.trino.parquet.reader.ParquetReader;
import io.trino.parquet.reader.RowGroupInfo;
import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HiveColumnProjectionInfo;
import io.trino.plugin.hive.parquet.ParquetReaderConfig;
import io.trino.plugin.hudi.file.HudiBaseFile;
import io.trino.plugin.hudi.reader.HudiTrinoReaderContext;
@@ -48,6 +49,7 @@ import io.trino.spi.connector.ConnectorTableHandle;
import io.trino.spi.connector.ConnectorTransactionHandle;
import io.trino.spi.connector.DynamicFilter;
import io.trino.spi.connector.EmptyPageSource;
+import io.trino.spi.predicate.Domain;
import io.trino.spi.predicate.TupleDomain;
import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
@@ -70,11 +72,14 @@ import org.joda.time.DateTimeZone;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
+import java.util.Set;
import java.util.stream.Collectors;
import static
io.trino.memory.context.AggregatedMemoryContext.newSimpleAggregatedMemoryContext;
@@ -383,9 +388,15 @@ public class HudiPageSourceProvider
// When not using columnNames, physical indexes are used and there
could be cases when the physical index in HiveColumnHandle is different from
the fileSchema of the
// parquet files. This could happen when schema evolution
happened. In such a case, we will need to remap the column indices in the
HiveColumnHandles.
+ // The projection and the predicate resolve the same names against
the same file, so the name-to-position
+ // map is built once per split and shared: one lookup table means
the two can never disagree about which
+ // physical column a name denotes, and a wide table pays for the
lower-casing pass only once.
+ // HiveColumnHandle names are in lower case, case-insensitive
+ Optional<Map<String, Integer>> physicalIndexMap = Optional.empty();
if (!useColumnNames) {
- // HiveColumnHandle names are in lower case, case-insensitive
- columns = remapColumnIndicesToPhysical(fileSchema, columns,
false);
+ Map<String, Integer> indexMap =
buildPhysicalIndexMap(fileSchema, false);
+ columns = remapColumnIndicesToPhysical(fileSchema, columns,
indexMap, false);
+ physicalIndexMap = Optional.of(indexMap);
}
Optional<MessageType> message = getParquetMessageType(columns,
useColumnNames, fileSchema);
@@ -397,7 +408,7 @@ public class HudiPageSourceProvider
TupleDomain<ColumnDescriptor> parquetTupleDomain =
options.isIgnoreStatistics() || !enablePredicatePushDown
? TupleDomain.all()
- : getParquetTupleDomain(descriptorsByPath,
getCombinedPredicate(hudiSplit, dynamicFilter), fileSchema, useColumnNames);
+ : getParquetTupleDomain(descriptorsByPath,
getPushdownPredicate(hudiSplit, dynamicFilter, physicalIndexMap), fileSchema,
useColumnNames);
TupleDomainParquetPredicate parquetPredicate =
buildPredicate(requestedSchema, parquetTupleDomain, descriptorsByPath,
timeZone);
@@ -482,41 +493,165 @@ public class HudiPageSourceProvider
boolean caseSensitive)
{
// Create a map from column name to its physical index in the
fileSchema.
- Map<String, Integer> physicalIndexMap = new HashMap<>();
- List<Type> fileFields = fileSchema.getFields();
- for (int i = 0; i < fileFields.size(); i++) {
- Type field = fileFields.get(i);
- String fieldName = field.getName();
- String mapKey = caseSensitive ? fieldName :
fieldName.toLowerCase(Locale.ROOT);
- physicalIndexMap.put(mapKey, i);
- }
+ return remapColumnIndicesToPhysical(fileSchema, requestedColumns,
buildPhysicalIndexMap(fileSchema, caseSensitive), caseSensitive);
+ }
+ /**
+ * {@link #remapColumnIndicesToPhysical(MessageType, List, boolean)}
against a {@code physicalIndexMap} the caller
+ * already built, so a split that remaps both its projection and its
predicate builds the map once.
+ * {@code caseSensitive} must be the one the map was built with, or the
lookups miss.
+ */
+ private static List<HiveColumnHandle> remapColumnIndicesToPhysical(
+ MessageType fileSchema,
+ List<HiveColumnHandle> requestedColumns,
+ Map<String, Integer> physicalIndexMap,
+ boolean caseSensitive)
+ {
// Iterate through the columns requested by Trino IN ORDER.
List<HiveColumnHandle> remappedHandles = new
ArrayList<>(requestedColumns.size());
for (HiveColumnHandle originalHandle : requestedColumns) {
- String requestedName = originalHandle.getBaseColumnName();
-
- // Determine the key to use for looking up the physical index
- String lookupKey = caseSensitive ? requestedName :
requestedName.toLowerCase(Locale.ROOT);
-
// Find the physical index from the file schema map constructed
from fileSchema. A column the file
// does not carry keeps an index one past the last field, which
the parquet reader null-fills.
- Integer physicalIndex = physicalIndexMap.get(lookupKey);
-
- HiveColumnHandle remappedHandle = new HiveColumnHandle(
- requestedName,
- physicalIndex == null ? fileFields.size() : physicalIndex,
- originalHandle.getBaseHiveType(),
- originalHandle.getType(),
- originalHandle.getHiveColumnProjectionInfo(),
- originalHandle.getColumnType(),
- originalHandle.getComment());
- remappedHandles.add(remappedHandle);
+ Integer physicalIndex =
physicalIndexMap.get(normalizeColumnName(originalHandle.getBaseColumnName(),
caseSensitive));
+ remappedHandles.add(withPhysicalIndex(originalHandle,
physicalIndex == null ? fileSchema.getFieldCount() : physicalIndex));
}
return remappedHandles;
}
+ /**
+ * Rebuilds a predicate's column handles on physical file ordinals, the
predicate-side counterpart of
+ * {@link #remapColumnIndicesToPhysical}. With {@code
hudi.parquet.use-column-names=false},
+ * {@code ParquetPageSourceFactory.getParquetTupleDomain} resolves a
predicate column positionally, as
+ * {@code fileSchema.getType(handle.getBaseHiveColumnIndex())}, but the
handles reaching it carry METASTORE
+ * ordinals: a metastore that omits the Hudi meta fields (hive sync with
{@code omit_metadata_fields=true})
+ * shifts every data column, and so does reordering or dropping one. Left
unremapped, the domain attaches to
+ * whichever column happens to sit at the stale ordinal and row groups are
pruned on that column's statistics,
+ * silently dropping rows.
+ * <p>
+ * Resolution is by name, so the predicate ends up bound to exactly the
column the projection reads - which is
+ * the property that matters, since the two are compared against each
other. It is not a defence against a
+ * column being dropped and re-added under full schema evolution:
name-based binding will match the new column
+ * to the old one, exactly as the projection remap and the whole {@code
use-column-names=true} mode already do.
+ * <p>
+ * A column the file does not carry is dropped from the predicate rather
than mapped to the
+ * {@link #remapColumnIndicesToPhysical} sentinel, which every absent
column would share. Dropping loses row
+ * group pruning but never a row: the static half of the predicate is
handed back to the engine in full as
+ * {@code HudiMetadata.applyFilter}'s remaining filter, and the dynamic
half is by construction redundant with
+ * the join above the scan. It is also what already happens today for a
predicate column the query does not
+ * read, since {@code descriptorsByPath} is derived from the projection and
+ * {@code getParquetTupleDomain} skips any column it cannot resolve.
+ *
+ * @param fileSchema The MessageType representing the physical schema of
the Parquet file.
+ * @param predicate The predicate to push down, keyed on handles carrying
metastore ordinals.
+ * @param caseSensitive Whether the lookup between Trino column names
(from handles) and Parquet field names (from fileSchema) should be
case-sensitive.
+ * @return The same domains, keyed on handles carrying physical ordinals,
minus the columns the file lacks.
+ */
+ @VisibleForTesting
+ public static TupleDomain<HiveColumnHandle>
remapPredicateColumnIndicesToPhysical(
+ MessageType fileSchema,
+ TupleDomain<HiveColumnHandle> predicate,
+ boolean caseSensitive)
+ {
+ return remapPredicateColumnIndicesToPhysical(predicate,
buildPhysicalIndexMap(fileSchema, caseSensitive), caseSensitive);
+ }
+
+ /**
+ * {@link #remapPredicateColumnIndicesToPhysical(MessageType, TupleDomain,
boolean)} against a
+ * {@code physicalIndexMap} the caller already built, so a split that
remaps both its projection and its predicate
+ * builds the map once. {@code caseSensitive} must be the one the map was
built with, or the lookups miss.
+ */
+ private static TupleDomain<HiveColumnHandle>
remapPredicateColumnIndicesToPhysical(
+ TupleDomain<HiveColumnHandle> predicate,
+ Map<String, Integer> physicalIndexMap,
+ boolean caseSensitive)
+ {
+ if (predicate.isAll() || predicate.isNone()) {
+ return predicate;
+ }
+
+ Set<Map.Entry<Integer, Optional<HiveColumnProjectionInfo>>>
pushedFields = new HashSet<>();
+ Map<HiveColumnHandle, Domain> remappedDomains = new LinkedHashMap<>();
+ for (Map.Entry<HiveColumnHandle, Domain> entry :
predicate.getDomains().orElseThrow().entrySet()) {
+ Integer physicalIndex =
physicalIndexMap.get(normalizeColumnName(entry.getKey().getBaseColumnName(),
caseSensitive));
+ if (physicalIndex == null) {
+ continue;
+ }
+ // Deduplicate on what getParquetTupleDomain resolves the handle
to rather than on the handle itself: two
+ // handles whose names differ only by case resolve to one file
column while staying unequal to each other,
+ // and pushing both down would hand it the same ColumnDescriptor
twice, which it rejects by failing the
+ // split. The base column alone is too coarse a key, because a
dereference handle carries its subfield
+ // path into the descriptor, so the projection is part of the key
and two projections of one base column
+ // both survive. Neither collision is reachable today - the case
one needs a metastore holding two such
+ // columns, which Hive's name normalisation rules out, and
trino-parquet lower-cases every field name when
+ // it builds the MessageType from the footer anyway - but keeping
only the first domain is a cheap
+ // guarantee that the read can never be made worse than pushing
nothing down.
+ if (pushedFields.add(Map.entry(physicalIndex,
entry.getKey().getHiveColumnProjectionInfo()))) {
+ remappedDomains.put(withPhysicalIndex(entry.getKey(),
physicalIndex), entry.getValue());
+ }
+ }
+ return TupleDomain.withColumnDomains(remappedDomains);
+ }
+
+ /**
+ * Maps each of {@code fileSchema}'s top-level field names to its physical
position.
+ */
+ private static Map<String, Integer> buildPhysicalIndexMap(MessageType
fileSchema, boolean caseSensitive)
+ {
+ Map<String, Integer> physicalIndexMap = new HashMap<>();
+ List<Type> fileFields = fileSchema.getFields();
+ for (int i = 0; i < fileFields.size(); i++) {
+
physicalIndexMap.put(normalizeColumnName(fileFields.get(i).getName(),
caseSensitive), i);
+ }
+ return physicalIndexMap;
+ }
+
+ private static String normalizeColumnName(String columnName, boolean
caseSensitive)
+ {
+ return caseSensitive ? columnName :
columnName.toLowerCase(Locale.ROOT);
+ }
+
+ /**
+ * Copies {@code handle} with its base column index replaced by a physical
one, every other attribute carried
+ * over unchanged. Note that the constructor's fourth argument is the BASE
type: it differs from
+ * {@code getType()} only for a dereference handle, whose {@code
getType()} is the projected subfield's type
+ * rather than the column's, and {@code createParquetPageSource} reads the
base type throughout.
+ * <p>
+ * Copying a dereference handle's projection across matters: {@code
createParquetPageSource} branches on
+ * {@code isBaseColumn()} and dereferences through {@code
getHiveColumnProjectionInfo}, and it reads the base
+ * column's stored {@code baseType} on the way. The connector never
produces such a handle today, because
+ * {@code HudiMetadata} does not implement {@code applyProjection}.
+ */
+ private static HiveColumnHandle withPhysicalIndex(HiveColumnHandle handle,
int physicalIndex)
+ {
+ return new HiveColumnHandle(
+ handle.getBaseColumnName(),
+ physicalIndex,
+ handle.getBaseHiveType(),
+ handle.getBaseType(),
+ handle.getHiveColumnProjectionInfo(),
+ handle.getColumnType(),
+ handle.getComment());
+ }
+
+ /**
+ * Resolves the predicate handed to {@code
ParquetPageSourceFactory.getParquetTupleDomain}. Only the
+ * positional mode needs the handles rebuilt; when columns are resolved by
name the metastore ordinals are
+ * never read, which is exactly when {@code physicalIndexMap} is empty.
Being handed the very map the projection
+ * was remapped with is what makes it structural, rather than a
convention, that the two agree about which
+ * physical column a name denotes.
+ */
+ private static TupleDomain<HiveColumnHandle> getPushdownPredicate(
+ HudiSplit hudiSplit,
+ DynamicFilter dynamicFilter,
+ Optional<Map<String, Integer>> physicalIndexMap)
+ {
+ TupleDomain<HiveColumnHandle> combinedPredicate =
getCombinedPredicate(hudiSplit, dynamicFilter);
+ return physicalIndexMap
+ .map(indexMap ->
remapPredicateColumnIndicesToPhysical(combinedPredicate, indexMap, false))
+ .orElse(combinedPredicate);
+ }
+
private static TupleDomain<HiveColumnHandle>
getCombinedPredicate(HudiSplit hudiSplit, DynamicFilter dynamicFilter)
{
// Combine static and dynamic predicates
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
index 7e938ba3aed1..fda63c8b4714 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiConnectorParquetColumnNamesTest.java
@@ -13,8 +13,16 @@
*/
package io.trino.plugin.hudi;
+import io.trino.plugin.hudi.testing.CompositeHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer;
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static
io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer.LATE_COLUMN;
+import static
io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer.SHADOWED_COLUMN;
+import static
io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer.THRESHOLD;
+import static
io.trino.plugin.hudi.testing.OmittedMetaColumnsHudiTablesInitializer.expectedRowsAboveThreshold;
public class TestHudiConnectorParquetColumnNamesTest
extends TestHudiSmokeTest
@@ -25,7 +33,38 @@ public class TestHudiConnectorParquetColumnNamesTest
{
return HudiQueryRunner.builder()
.addConnectorProperty("hudi.parquet.use-column-names", "false")
- .setDataLoader(new ResourceHudiTablesInitializer())
+ // The resource tables all register the Hudi meta columns in
the metastore, so their metastore
+ // ordinals already equal their physical ones and nothing here
resolves a stale ordinal. The
+ // second fixture is the one whose metastore omits them.
+ .setDataLoader(new CompositeHudiTablesInitializer(
+ new ResourceHudiTablesInitializer(),
+ new OmittedMetaColumnsHudiTablesInitializer()))
.build();
}
+
+ /**
+ * apache/hudi#19387: with columns resolved positionally, a predicate
handle carrying a metastore ordinal has to
+ * be rebuilt on the file's physical ordinal before it is pushed into the
parquet reader. Left unremapped, the
+ * domain lands on whichever column physically sits at that ordinal --
here {@code shadowed_value}, whose values
+ * are far below the threshold -- and the only row group is pruned, so the
query returns nothing at all.
+ * <p>
+ * {@code shadowed_value} has to stay in the SELECT list: {@code
descriptorsByPath} is derived from the
+ * projection, so a domain resolving to a column the query does not read
finds no descriptor and is dropped
+ * instead of being misapplied. Narrowing this projection turns the test
green against the unfixed code.
+ * <p>
+ * Only the {@code hudi.parquet.use-column-names=false} suite runs this;
the name-based parent resolves the
+ * predicate by name and was never affected.
+ */
+ @Test
+ public void testPredicateOnColumnWithStaleMetastoreOrdinal()
+ {
+ assertQuery(
+ "SELECT key, %s, %s FROM %s WHERE %s > %s ORDER BY
key".formatted(
+ SHADOWED_COLUMN,
+ LATE_COLUMN,
+ OmittedMetaColumnsHudiTablesInitializer.TABLE_NAME,
+ LATE_COLUMN,
+ THRESHOLD),
+ expectedRowsAboveThreshold());
+ }
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
index ba21f2b34a67..a3e62db9de0d 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMorMergeModeSemantics.java
@@ -132,4 +132,25 @@ public class TestHudiMorMergeModeSemantics
"SELECT key, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " ORDER BY key",
"VALUES ('k1', CAST(11 AS BIGINT)), ('k4', 40), ('k5', 50),
('k6', 60)");
}
+
+ @Test
+ public void testPredicateIsNotPushedIntoTheBaseReadOfAMergedSplit()
+ {
+ // HudiPageSourceProvider enables parquet predicate pushdown for
base-file-only splits ONLY; the merge
+ // path builds its base page source with it off. That invariant is
load-bearing and nothing else pins it:
+ // pruning happens on the BASE row group's statistics, before the log
records the merge needs are seen.
+ //
+ // 65 is above every base value (10..60) and below the obsolete k6
update (66), which loses on event time
+ // and must not surface. With pushdown enabled on this path the whole
row group is pruned, the base side
+ // comes back empty, and every log record is then emitted as an insert
-- so k6 appears with 66 and the
+ // merge is silently skipped. Note the naive shape does NOT
discriminate: for a row whose log record wins
+ // and carries the full record, dropping the base row still yields the
right answer.
+ assertQueryReturnsEmptyResult(
+ "SELECT key, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " WHERE value > 65");
+ // Anchor: the same predicate one step lower does return the rows it
should, so the query above is empty
+ // because of the merge, not because nothing ever matches.
+ assertQuery(
+ "SELECT key, value FROM " +
EventTimeDeletesHudiTablesInitializer.RT_TABLE_NAME + " WHERE value > 45 ORDER
BY key",
+ "VALUES ('k5', CAST(50 AS BIGINT)), ('k6', 60)");
+ }
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
index 4bce0ba0cfc5..a224f2ffb6c2 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java
@@ -13,28 +13,137 @@
*/
package io.trino.plugin.hudi;
+import io.trino.filesystem.local.LocalInputFile;
import io.trino.metastore.HiveType;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HiveColumnProjectionInfo;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.Range;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.predicate.ValueSet;
import io.trino.spi.type.BigintType;
+import io.trino.spi.type.RowType;
import io.trino.spi.type.Type;
+import io.trino.testing.MaterializedResult;
+import io.trino.testing.TestingConnectorSession;
+import org.apache.parquet.conf.PlainParquetConfiguration;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.io.LocalOutputFile;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Types;
+import org.joda.time.DateTimeZone;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
import static
io.trino.plugin.hudi.HudiPageSourceProvider.remapColumnIndicesToPhysical;
+import static
io.trino.plugin.hudi.HudiPageSourceProvider.remapPredicateColumnIndicesToPhysical;
import static io.trino.spi.type.DoubleType.DOUBLE;
import static io.trino.spi.type.IntegerType.INTEGER;
import static io.trino.spi.type.VarcharType.VARCHAR;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS;
import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
import static org.assertj.core.api.Assertions.assertThat;
+/**
+ * Covers {@link HudiPageSourceProvider}'s column remapping from both sides:
the index arithmetic of
+ * {@code remapColumnIndicesToPhysical} and {@code
remapPredicateColumnIndicesToPhysical} on their own, and reads of
+ * a real base file through {@code createPageSource}, which are what prove a
remapped predicate actually reaches the
+ * parquet reader and prunes the column it was written for.
+ * <p>
+ * The base file the reading tests share has a physical column order the
metastore does not: it carries the five
+ * {@code _hoodie_*} meta columns, which hive sync with
+ * {@code hoodie.datasource.hive_sync.omit_metadata_fields=true} leaves out,
so every data column's metastore ordinal
+ * is five below its physical position. With {@code
hudi.parquet.use-column-names=false} the parquet page source
+ * resolves columns positionally, so a predicate whose handle still carries
the metastore ordinal lands on whichever
+ * column physically sits there and row groups get pruned on that column's
statistics. The fixture makes that
+ * observable: {@link #PREDICATE_COLUMN} grows with the row index while every
other data column stays in 0..9, so a
+ * domain meant for it but applied to any other column excludes every row
group and the read returns nothing.
+ * <p>
+ * Note that the shadowed column has to be part of the PROJECTION for the
damage to appear: {@code
+ * descriptorsByPath} is derived from the projection, so a domain resolving to
a column the query does not read
+ * finds no descriptor and is discarded instead. Do not "simplify" the
projections below to the predicate column
+ * alone - that turns those tests green against the unfixed code.
+ */
class TestHudiPageSourceProviderTest
{
+ private static final int DATA_COLUMN_COUNT = 10;
+ /** The column the predicate is on: physically at 12, but numbered 7 by a
metastore without the meta columns. */
+ private static final String PREDICATE_COLUMN = "c7";
+ /** The column physically sitting at {@code c7}'s stale ordinal, and
therefore the one that shadows it. */
+ private static final String SHADOWED_COLUMN = "c2";
+ 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 = hudiFileSchema(DATA_COLUMN_COUNT);
+ baseFile = tempDir.resolve("base_file.parquet");
+ SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema);
+ try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new
LocalOutputFile(baseFile))
+ .withType(schema)
+ .withConf(new PlainParquetConfiguration())
+ .withRowGroupSize(1024L)
+ .withPageSize(512)
+ .build()) {
+ for (int row = 0; row < ROW_COUNT; row++) {
+ Group group = groupFactory.newGroup();
+ for (String metaColumn : HOODIE_META_COLUMNS) {
+ group.append(metaColumn, metaColumn + "_" + row);
+ }
+ for (int column = 0; column < DATA_COLUMN_COUNT; column++) {
+ String columnName = "c" + column;
+ group.append(columnName,
columnName.equals(PREDICATE_COLUMN) ? row : row % 10);
+ }
+ writer.write(group);
+ }
+ }
+ // The writer flushes a row group whenever the buffered size is over
withRowGroupSize, checked every
+ // parquet.page.size.row.check.min records (100 by default), which is
what actually splits this file.
+ // Assert the outcome rather than the knobs: with a single row group
there would be nothing to prune,
+ // and every test below would pass without proving anything.
+ assertThat(rowGroupCount(baseFile)).as("row groups
written").isGreaterThan(1);
+ }
+
@Test
public void testRemapSimpleMatchCaseInsensitive()
{
@@ -193,6 +302,455 @@ class TestHudiPageSourceProviderTest
assertHandle(remapped.get(1), "col_x", fileSchema.getFieldCount(),
HiveType.HIVE_STRING, VARCHAR);
}
+ @Test
+ public void testRemapPredicateStaleMetastoreOrdinals()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0, c1, c2]
+ MessageType fileSchema = hudiFileSchema(3);
+
+ // A metastore synced with omit_metadata_fields=true carries no meta
columns, so "c2" is numbered 2
+ // while it physically sits at 7, and "c0" is numbered 0 while it
physically sits at 5.
+ HiveColumnHandle staleC2 = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle staleC0 = createDummyHandle("c0", 0,
HiveType.HIVE_INT, INTEGER);
+ Domain c2Domain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 900L)), false);
+ Domain c0Domain = Domain.singleValue(INTEGER, 7L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(staleC2, c2Domain,
staleC0, c0Domain)),
+ false);
+
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(2);
+ // Each domain now keys off the column's physical position, so it is
matched against that column's statistics
+ assertThat(handleOf(domains,
"c2").getBaseHiveColumnIndex()).isEqualTo(7);
+ assertThat(domains.get(handleOf(domains, "c2"))).isEqualTo(c2Domain);
+ assertThat(handleOf(domains,
"c0").getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(handleOf(domains, "c0"))).isEqualTo(c0Domain);
+ }
+
+ @Test
+ public void testRemapPredicateDropsColumnsAbsentFromFile()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle present = createDummyHandle("c0", 0,
HiveType.HIVE_INT, INTEGER);
+ // Added after this base file was written, so the file does not carry
them. Two of them, because the
+ // projection remap's out-of-range sentinel is one value that every
absent column would share.
+ HiveColumnHandle firstAbsent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle secondAbsent = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+ Domain presentDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(
+ present, presentDomain,
+ firstAbsent, Domain.singleValue(INTEGER, 2L),
+ secondAbsent, Domain.singleValue(INTEGER, 3L))),
+ false);
+
+ // Both absent columns are dropped rather than mapped to that shared
sentinel, which would have collided.
+ // Dropping them costs row group pruning only; the engine still
applies the filter itself.
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(1);
+ assertThat(handleOf(domains,
"c0").getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(handleOf(domains,
"c0"))).isEqualTo(presentDomain);
+ }
+
+ @Test
+ public void testRemapPredicateKeepsOneDomainPerPhysicalColumn()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ // Two handles whose names differ only by case resolve to the same
file field, so both land on physical
+ // index 5 while remaining unequal to each other. The connector cannot
produce this - Hive normalises
+ // column names to lower case - but pushing both down would hand
getParquetTupleDomain one
+ // ColumnDescriptor twice, which it rejects by failing the whole split.
+ HiveColumnHandle upperCase = createDummyHandle("C0", 0,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle lowerCase = createDummyHandle("c0", 3,
HiveType.HIVE_INT, INTEGER);
+ Domain firstDomain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 10L)), false);
+ // Insertion-ordered so that "first wins" is a deterministic assertion
+ Map<HiveColumnHandle, Domain> predicate = new LinkedHashMap<>();
+ predicate.put(upperCase, firstDomain);
+ predicate.put(lowerCase,
Domain.create(ValueSet.ofRanges(Range.lessThan(INTEGER, 20L)), false));
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema, TupleDomain.withColumnDomains(predicate), false);
+
+ // Only the first is pushed down, and no IllegalArgumentException
escapes
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(1);
+ HiveColumnHandle survivor = handleOf(domains, "C0");
+ assertThat(survivor.getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(survivor)).isEqualTo(firstDomain);
+ }
+
+ @Test
+ public void testRemapPredicateKeepsBothProjectionsOfOneBaseColumn()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ // Two dereference handles projecting DIFFERENT subfields of the same
struct column. Both resolve to base
+ // physical index 5, but getParquetTupleDomain builds a descriptor per
subfield path, so it would accept
+ // both; deduplicating on the base index alone would silently discard
one of the two domains.
+ HiveType structType = HiveType.valueOf("struct<f:int,g:int>");
+ RowType baseType = RowType.rowType(RowType.field("f", INTEGER),
RowType.field("g", INTEGER));
+ HiveColumnHandle onF = dereferenceHandle(structType, baseType, 0, "f");
+ HiveColumnHandle onG = dereferenceHandle(structType, baseType, 1, "g");
+ Domain fDomain = Domain.singleValue(INTEGER, 5L);
+ Domain gDomain = Domain.singleValue(INTEGER, 3L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(onF, fDomain, onG,
gDomain)),
+ false);
+
+ assertThat(remapped.getDomains().orElseThrow())
+ .as("both subfield domains survive, each on the base column's
physical index")
+ .isEqualTo(Map.of(
+ withBaseIndex(onF, 5), fDomain,
+ withBaseIndex(onG, 5), gDomain));
+ }
+
+ @Test
+ public void testRemapPreservesTheBaseTypeOfADereferenceHandle()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ // A handle projecting one field out of a struct column. HudiMetadata
does not implement applyProjection,
+ // so the connector never builds one today, but the remap has to
rebuild it without corrupting it: the
+ // constructor's type argument is the BASE column's type, while
getType() is the projected field's.
+ RowType baseType = RowType.rowType(RowType.field("f", INTEGER));
+ HiveColumnHandle dereference =
dereferenceHandle(HiveType.valueOf("struct<f:int>"), baseType, 0, "f");
+
+ HiveColumnHandle remapped = remapColumnIndicesToPhysical(fileSchema,
List.of(dereference), false).get(0);
+
+ assertThat(remapped.getBaseHiveColumnIndex())
+ .as("physical index")
+ .isEqualTo(5);
+ assertThat(remapped.getBaseType())
+ .as("base type, which is what the parquet page source reads")
+ .isEqualTo(baseType);
+ assertThat(remapped.getType())
+ .as("projected field type")
+ .isEqualTo(INTEGER);
+ assertThat(remapped.getHiveColumnProjectionInfo())
+ .as("projection info")
+ .isEqualTo(dereference.getHiveColumnProjectionInfo());
+ }
+
+ @Test
+ public void testRemapPredicateAllAndNonePassThrough()
+ {
+ MessageType fileSchema = hudiFileSchema(1);
+
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
TupleDomain.<HiveColumnHandle>all(), false))
+ .isEqualTo(TupleDomain.all());
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
TupleDomain.<HiveColumnHandle>none(), false))
+ .isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testRemapPredicateCaseSensitivity()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle upperCase = createDummyHandle("C0", 0,
HiveType.HIVE_INT, INTEGER);
+ TupleDomain<HiveColumnHandle> predicate =
TupleDomain.withColumnDomains(Map.of(upperCase, Domain.singleValue(INTEGER,
1L)));
+
+ // Case-insensitive: "C0" resolves to the file's "c0" at physical
index 5
+ Map<HiveColumnHandle, Domain> insensitive =
remapPredicateColumnIndicesToPhysical(fileSchema, predicate, false)
+ .getDomains().orElseThrow();
+ assertThat(handleOf(insensitive,
"C0").getBaseHiveColumnIndex()).isEqualTo(5);
+
+ // Case-sensitive: no match, so the domain is dropped instead of being
left on a stale ordinal
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
predicate, true).isAll()).isTrue();
+ }
+
+ @Test
+ public void testRemapPredicatePreservesEveryOtherHandleAttribute()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle original = new HiveColumnHandle(
+ "c0",
+ 0,
+ HiveType.HIVE_INT,
+ INTEGER,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.of("a comment"));
+ Domain domain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 900L)), true);
+
+ Map<HiveColumnHandle, Domain> domains =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(original, domain)),
+ false)
+ .getDomains().orElseThrow();
+
+ HiveColumnHandle remapped = handleOf(domains, "c0");
+ assertHandle(remapped, "c0", 5, HiveType.HIVE_INT, INTEGER);
+ assertThat(remapped.getComment())
+ .as("Comment mismatch for c0")
+ .isEqualTo(Optional.of("a comment"));
+ assertThat(domains.get(remapped))
+ .as("Domain mismatch for c0")
+ .isEqualTo(domain);
+ }
+
+ @Test
+ public void testPredicateOnStaleOrdinalStillPrunesRowGroups()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // Correct results alone would also be produced by pushing nothing
down; reading fewer rows than the file
+ // holds is only possible if the domain reached the column it was
written for, and the matching rows must
+ // survive that pruning. The shadowed column never leaves 0..9, so a
domain of "> 900" applied to it would
+ // prune every row group instead.
+ assertThat(result.getRowCount())
+ .as("rows read out of %s", ROW_COUNT)
+ .isLessThan(ROW_COUNT);
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s after pruning", PREDICATE_COLUMN,
THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testStaleOrdinalArrivingThroughADynamicFilter()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ // A dynamic filter reaches getCombinedPredicate by its own route, and
its handles carry the same stale
+ // metastore ordinals the split's predicate does
+ MaterializedResult result = read(projection, TupleDomain.all(), false,
+ dynamicFilterOn(greaterThanThreshold(PREDICATE_COLUMN)));
+
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching a dynamic filter of %s > %s",
PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testPredicateOnColumnAddedAfterBaseFileWasWritten()
+ throws Exception
+ {
+ // The metastore carries one column more than this base file does,
numbered 10 - an ordinal that is still
+ // in range physically, where it picks out "c5"
+ String addedColumn = "c" + DATA_COLUMN_COUNT;
+ List<HiveColumnHandle> projection = List.of(dataColumn("c5"),
dataColumn(PREDICATE_COLUMN), dataColumn(addedColumn));
+
+ // IS NULL, not a range: the added column is null in every row of this
base file, so this predicate is
+ // satisfied by all of them. A range predicate would be unsatisfiable
here and the buggy read's empty
+ // result would be the right answer by accident.
+ MaterializedResult result = read(projection,
+ TupleDomain.withColumnDomains(Map.of(dataColumn(addedColumn),
Domain.onlyNull(INTEGER))),
+ false, DynamicFilter.EMPTY);
+
+ // The added column must not stay on its stale metastore ordinal:
pushed positionally it would land on
+ // "c5", which has no nulls at all, and every row group would be
pruned.
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+
assertThat(result.getMaterializedRows().getFirst().getField(2)).as("value of
%s", addedColumn).isNull();
+ }
+
+ @Test
+ public void testPositionalAndNameBasedResolutionAgree()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+ TupleDomain<HiveColumnHandle> predicate =
greaterThanThreshold(PREDICATE_COLUMN);
+
+ MaterializedResult positional = read(projection, predicate, false,
DynamicFilter.EMPTY);
+ MaterializedResult byName = read(projection, predicate, true,
DynamicFilter.EMPTY);
+
+ // Anchor the comparison: both modes regressing to no pushdown at all
would otherwise agree happily
+ assertThat(byName.getRowCount()).as("rows read with
use-column-names=true").isLessThan(ROW_COUNT);
+ assertThat(positional.getMaterializedRows())
+ .as("hudi.parquet.use-column-names=false must read what
use-column-names=true reads")
+ .isEqualTo(byName.getMaterializedRows());
+ }
+
+ /**
+ * Reads the whole base file through the page source the connector builds
for a split with no log files, which
+ * is the only path on which it enables predicate pushdown.
+ */
+ private static MaterializedResult read(
+ List<HiveColumnHandle> projection,
+ TupleDomain<HiveColumnHandle> predicate,
+ boolean useParquetColumnNames,
+ DynamicFilter dynamicFilter)
+ throws Exception
+ {
+ long fileSize = Files.size(baseFile);
+ HudiSplit split = new HudiSplit(
+ new HudiBaseFile(baseFile.toString(),
baseFile.getFileName().toString(), fileSize, 0, 0, fileSize),
+ List.of(),
+ "000",
+ predicate,
+ List.of(),
+ SplitWeight.standard());
+ HudiSessionProperties sessionProperties = new HudiSessionProperties(
+ new
HudiConfig().setUseParquetColumnNames(useParquetColumnNames),
+ new ParquetReaderConfig());
+ ConnectorSession session = TestingConnectorSession.builder()
+ .setPropertyMetadata(sessionProperties.getSessionProperties())
+ .build();
+
+ List<Type> types =
projection.stream().map(HiveColumnHandle::getType).toList();
+ try (ConnectorPageSource pageSource = createPageSource(
+ session,
+ projection,
+ split,
+ new LocalInputFile(baseFile.toFile()),
+ baseFile.toString(),
+ 0L,
+ fileSize,
+ OptionalLong.of(fileSize),
+ new FileFormatDataSourceStats(),
+ ParquetReaderOptions.builder().build(),
+ DateTimeZone.UTC,
+ dynamicFilter,
+ true)) {
+ return materializeSourceDataStream(session, pageSource,
types).toTestTypes();
+ }
+ }
+
+ /**
+ * Builds a file schema laid out like a Hudi base file: the five {@code
_hoodie_*} meta columns followed by
+ * {@code dataColumnCount} int columns named {@code c0..cN}. A metastore
synced with
+ * {@code hoodie.datasource.hive_sync.omit_metadata_fields=true} omits the
meta columns, so a data column's
+ * metastore ordinal is its physical ordinal minus five.
+ */
+ private static MessageType hudiFileSchema(int dataColumnCount)
+ {
+ List<org.apache.parquet.schema.Type> fields = new ArrayList<>();
+ for (String metaColumn : HOODIE_META_COLUMNS) {
+ fields.add(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY,
OPTIONAL).as(LogicalTypeAnnotation.stringType()).named(metaColumn));
+ }
+ for (int i = 0; i < dataColumnCount; i++) {
+ fields.add(Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
OPTIONAL).named("c" + i));
+ }
+ return new MessageType("hudi_base_file", fields);
+ }
+
+ /**
+ * Builds the handle a metastore without the Hudi meta columns produces:
numbered by its position among the
+ * data columns alone, which is {@code HOODIE_META_COLUMNS.size()} short
of its physical position. Only
+ * {@code c0..cN} data column names are accepted - the numeric suffix IS
the metastore ordinal - so a meta
+ * column name passed here would fail to parse rather than produce a
meaningful handle.
+ */
+ private static HiveColumnHandle dataColumn(String columnName)
+ {
+ return createBaseColumn(columnName, parseInt(columnName.substring(1)),
HiveType.HIVE_INT, INTEGER,
+ HiveColumnHandle.ColumnType.REGULAR, Optional.empty());
+ }
+
+ /** A handle projecting the {@code fieldIndex}-th field, named {@code
fieldName}, out of the struct column {@code c0}. */
+ private static HiveColumnHandle dereferenceHandle(HiveType structType,
RowType baseType, int fieldIndex, String fieldName)
+ {
+ return new HiveColumnHandle(
+ "c0",
+ 0,
+ structType,
+ baseType,
+ Optional.of(new HiveColumnProjectionInfo(List.of(fieldIndex),
List.of(fieldName), HiveType.HIVE_INT, INTEGER)),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /** The handle {@code remapColumnIndicesToPhysical} is expected to rebuild
from {@code handle}. */
+ private static HiveColumnHandle withBaseIndex(HiveColumnHandle handle, int
baseHiveColumnIndex)
+ {
+ return new HiveColumnHandle(
+ handle.getBaseColumnName(),
+ baseHiveColumnIndex,
+ handle.getBaseHiveType(),
+ handle.getBaseType(),
+ handle.getHiveColumnProjectionInfo(),
+ handle.getColumnType(),
+ handle.getComment());
+ }
+
+ private static TupleDomain<HiveColumnHandle> greaterThanThreshold(String
columnName)
+ {
+ return TupleDomain.withColumnDomains(Map.of(
+ dataColumn(columnName),
+ Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER,
THRESHOLD)), false)));
+ }
+
+ private static DynamicFilter dynamicFilterOn(TupleDomain<HiveColumnHandle>
predicate)
+ {
+ return new DynamicFilter()
+ {
+ @Override
+ public Set<ColumnHandle> getColumnsCovered()
+ {
+ return
Set.copyOf(predicate.getDomains().orElseThrow().keySet());
+ }
+
+ @Override
+ public CompletableFuture<?> isBlocked()
+ {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public boolean isComplete()
+ {
+ return true;
+ }
+
+ @Override
+ public boolean isAwaitable()
+ {
+ return false;
+ }
+
+ @Override
+ public TupleDomain<ColumnHandle> getCurrentPredicate()
+ {
+ return predicate.transformKeys(ColumnHandle.class::cast);
+ }
+ };
+ }
+
+ private static long matchingRowCount(MaterializedResult result,
List<HiveColumnHandle> projection, String columnName)
+ {
+ int fieldIndex = projection.indexOf(dataColumn(columnName));
+ return result.getMaterializedRows().stream()
+ .map(row -> row.getField(fieldIndex))
+ .filter(value -> value != null && ((Number) value).longValue()
> THRESHOLD)
+ .count();
+ }
+
+ private static int rowGroupCount(Path path)
+ throws IOException
+ {
+ try (ParquetFileReader reader = ParquetFileReader.open(new
org.apache.parquet.io.LocalInputFile(path))) {
+ return reader.getRowGroups().size();
+ }
+ }
+
+ /**
+ * Returns the single remapped handle carrying the given base column name.
+ */
+ private static HiveColumnHandle handleOf(Map<HiveColumnHandle, Domain>
domains, String baseColumnName)
+ {
+ return domains.keySet().stream()
+ .filter(handle ->
handle.getBaseColumnName().equals(baseColumnName))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No domain was kept for
column " + baseColumnName));
+ }
+
/**
* Creates a basic HiveColumnHandle for testing.
* Assumes REGULAR column type and no projection info or comments.
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
index 2eac0a195f87..9bcf0d6bcac4 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java
@@ -148,6 +148,16 @@ public abstract class AbstractMergerHudiTablesInitializer
/** The table's data columns, in schema order; the Hudi metadata columns
are prepended by this class. */
protected abstract List<Column> dataColumns();
+ /**
+ * Whether the metastore definition prepends {@link #HUDI_META_COLUMNS}.
The base file always carries them, so
+ * returning {@code false} models hive sync with {@code
hoodie.datasource.hive_sync.omit_metadata_fields=true}:
+ * every data column's metastore ordinal then sits five below its physical
position in the file.
+ */
+ protected boolean includeMetaColumnsInMetastore()
+ {
+ return true;
+ }
+
/** The Avro schema the write client writes, matching {@link
#dataColumns()}. */
protected abstract Schema avroSchema();
@@ -285,7 +295,7 @@ public abstract class AbstractMergerHudiTablesInitializer
.setTableType(EXTERNAL_TABLE.name())
.setOwner(Optional.of("public"))
.setDataColumns(ImmutableList.<Column>builder()
- .addAll(HUDI_META_COLUMNS)
+ .addAll(includeMetaColumnsInMetastore() ?
HUDI_META_COLUMNS : ImmutableList.<Column>of())
.addAll(dataColumns())
.build())
.setParameters(ImmutableMap.of("serialization.format", "1",
"EXTERNAL", "TRUE"))
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OmittedMetaColumnsHudiTablesInitializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OmittedMetaColumnsHudiTablesInitializer.java
new file mode 100644
index 000000000000..1be2c44cf65d
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/testing/OmittedMetaColumnsHudiTablesInitializer.java
@@ -0,0 +1,157 @@
+/*
+ * 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_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+
+/**
+ * Creates a non-partitioned table whose METASTORE column list omits the five
{@code _hoodie_*} meta columns the
+ * base file itself carries -- the layout hive sync leaves behind with
+ * {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}. Every data
column's metastore ordinal is therefore
+ * five below its physical position, which is the shape {@code
hudi.parquet.use-column-names=false} resolves
+ * positionally and the one apache/hudi#19387 was about. Every other fixture
in this module registers the meta
+ * columns ({@code ResourceHudiTablesInitializer.TestingTable.getDataColumns}
prepends them unconditionally), so
+ * metastore ordinal equals physical ordinal there and the bug cannot appear.
+ * <p>
+ * The column list looks padded on purpose. Physical fields 0..4 are always
the meta columns, so a data column at
+ * metastore ordinal {@code i} is read positionally at physical field {@code
i}, which is a meta column for any
+ * {@code i < 5}. A meta column can never be part of a projection here -- the
metastore does not expose it -- so
+ * {@code descriptorsByPath} has no entry for it and a stray domain is
silently discarded rather than misapplied.
+ * Only from the SIXTH data column on does the stale ordinal land on a real,
projectable column. Hence
+ * {@code late_value} at metastore ordinal 5, shadowed by {@code
shadowed_value} at physical field 5, and the four
+ * fillers in between.
+ * <p>
+ * The values are disjoint so the damage is unambiguous: {@code
shadowed_value} stays in 1..5 while
+ * {@code late_value} is above 1000, so a domain meant for {@code late_value}
but matched against
+ * {@code shadowed_value}'s statistics prunes the only row group and the query
returns nothing.
+ * <p>
+ * A single bulk-insert commit, so the file slice has no log files: predicate
pushdown is only enabled for
+ * base-file-only splits. See {@code TestHudiConnectorParquetColumnNamesTest}.
+ */
+public class OmittedMetaColumnsHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "omitted_meta_columns_mor";
+
+ /** The column the predicate goes on: metastore ordinal 5, physical field
10. */
+ public static final String LATE_COLUMN = "late_value";
+ /** The column physically sitting at {@link #LATE_COLUMN}'s stale ordinal,
and therefore the one that shadows it. */
+ public static final String SHADOWED_COLUMN = "shadowed_value";
+ /** Below every {@link #LATE_COLUMN} value and above every {@link
#SHADOWED_COLUMN} one. */
+ public static final long THRESHOLD = 900;
+
+ private static final int ROW_COUNT = 5;
+ private static final List<String> FILLER_COLUMNS =
ImmutableList.of("filler_1", "filler_2", "filler_3", "filler_4");
+
+ public OmittedMetaColumnsHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected boolean includeMetaColumnsInMetastore()
+ {
+ return false;
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ ImmutableList.Builder<Column> columns = ImmutableList.builder();
+ columns.add(new Column(SHADOWED_COLUMN, HIVE_LONG, Optional.empty(),
Map.of()));
+ FILLER_COLUMNS.forEach(name -> columns.add(new Column(name, HIVE_LONG,
Optional.empty(), Map.of())));
+ columns.add(new Column(LATE_COLUMN, HIVE_LONG, Optional.empty(),
Map.of()));
+ columns.add(new Column(RECORD_KEY_FIELD, HIVE_STRING,
Optional.empty(), Map.of()));
+ columns.add(new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(),
Map.of()));
+ return columns.build();
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = new ArrayList<>();
+ fields.add(new Schema.Field(SHADOWED_COLUMN,
Schema.create(Schema.Type.LONG)));
+ FILLER_COLUMNS.forEach(name -> fields.add(new Schema.Field(name,
Schema.create(Schema.Type.LONG))));
+ fields.add(new Schema.Field(LATE_COLUMN,
Schema.create(Schema.Type.LONG)));
+ fields.add(new Schema.Field(RECORD_KEY_FIELD,
Schema.create(Schema.Type.STRING)));
+ fields.add(new Schema.Field(ORDERING_FIELD,
Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, fields);
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder
tableBuilder)
+ {
+ tableBuilder.setRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder
writeConfigBuilder)
+ {
+
writeConfigBuilder.withRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING);
+ }
+
+ @Override
+ protected void
writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ {
+ Schema schema = avroSchema();
+ List<HoodieRecord<HoodieAvroPayload>> records = new ArrayList<>();
+ for (int row = 1; row <= ROW_COUNT; row++) {
+ records.add(record(schema, "k" + row, row, 1000L + row));
+ }
+ // One commit only: a file slice with log files would take the merge
path, which disables pushdown.
+ String commit = client.startCommit();
+ List<WriteStatus> statuses = client.bulkInsert(records, commit);
+ client.commit(commit, statuses);
+ }
+
+ /** The expected rows of {@code SELECT key, shadowed_value, late_value ...
WHERE late_value > THRESHOLD}. */
+ public static String expectedRowsAboveThreshold()
+ {
+ List<String> rows = new ArrayList<>();
+ for (int row = 1; row <= ROW_COUNT; row++) {
+ rows.add("('k%s', CAST(%s AS BIGINT), CAST(%s AS
BIGINT))".formatted(row, row, 1000 + row));
+ }
+ return "VALUES " + String.join(", ", rows);
+ }
+
+ private static HoodieRecord<HoodieAvroPayload> record(Schema schema,
String key, long shadowedValue, long lateValue)
+ {
+ GenericRecord record = new GenericData.Record(schema);
+ record.put(SHADOWED_COLUMN, shadowedValue);
+ FILLER_COLUMNS.forEach(name -> record.put(name, 0L));
+ record.put(LATE_COLUMN, lateValue);
+ record.put(RECORD_KEY_FIELD, key);
+ record.put(ORDERING_FIELD, 100L);
+ return avroRecord(record, key);
+ }
+}
