voonhous commented on code in PR #19456:
URL: https://github.com/apache/hudi/pull/19456#discussion_r3698764299
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -482,41 +486,135 @@ public static List<HiveColumnHandle>
remapColumnIndicesToPhysical(
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);
- }
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, 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(lookupKey(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)
+ {
+ if (predicate.isAll() || predicate.isNone()) {
+ return predicate;
+ }
+
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, caseSensitive);
+ Set<Integer> pushedPhysicalIndices = new HashSet<>();
+ Map<HiveColumnHandle, Domain> remappedDomains = new LinkedHashMap<>();
+ for (Map.Entry<HiveColumnHandle, Domain> entry :
predicate.getDomains().orElseThrow().entrySet()) {
+ Integer physicalIndex =
physicalIndexMap.get(lookupKey(entry.getKey().getBaseColumnName(),
caseSensitive));
+ if (physicalIndex == null) {
+ continue;
+ }
+ // Deduplicate on the physical index rather than on the rebuilt
handle: 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 getParquetTupleDomain the same ColumnDescriptor
twice, which it rejects by failing the
+ // split. That needs a metastore holding two such columns, which
Hive's name normalisation rules out,
+ // but keeping only the first domain is a cheap guarantee that the
read can never be made worse than
+ // pushing nothing down.
+ if (pushedPhysicalIndices.add(physicalIndex)) {
+ 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(lookupKey(fileFields.get(i).getName(),
caseSensitive), i);
+ }
+ return physicalIndexMap;
+ }
+
+ private static String lookupKey(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 is not a claim that
such a handle would be read correctly;
+ * {@code createParquetPageSource} ignores {@code
getHiveColumnProjectionInfo} entirely. The connector never
+ * produces one, because {@code HudiMetadata} does not implement {@code
applyProjection}.
Review Comment:
**Correctness (javadoc).** This says `createParquetPageSource` ignores
`getHiveColumnProjectionInfo` entirely. It does not. In Trino 481:
```java
//
trino-hive-481-sources.jar!io/trino/plugin/hive/parquet/ParquetPageSourceFactory.java:524-533
if (column.isBaseColumn()) {
transforms.column(ordinal, coercer.map(Function.identity()));
}
else {
transforms.dereferenceField(
ImmutableList.<Integer>builder().add(ordinal).addAll(getProjection(column,
baseColumn)).build(),
coercer.map(Function.identity()));
}
```
and the same method reads `baseColumn.getBaseType()` at `:509` and `:511` to
pick the coercer and the read type. That is exactly *why* the `getType()` ->
`getBaseType()` swap in `withPhysicalIndex` is right. As written, the comment
tells the next maintainer the projection copy is inert, which is an invitation
to revert the very fix this hunk makes.
```suggestion
* 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}.
```
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPredicatePushdownColumnOrdinals.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi;
+
+import io.trino.filesystem.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.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 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.metastore.HiveType.HIVE_INT;
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads a base file whose physical column order does not match the
metastore's, the layout hive sync produces
+ * with {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}: the
five {@code _hoodie_*} meta columns are
+ * absent from the metastore, so every data column's metastore ordinal is five
below its physical position.
+ * <p>
+ * 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: {@code c7} grows with the
+ * row index while every other data column stays in 0..9, so a domain meant
for {@code c7} 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 these tests green against the unfixed code.
+ */
+class TestHudiPredicatePushdownColumnOrdinals
Review Comment:
**Coverage.** Nothing at the connector level pins this fix, and that is
exactly the blind spot that let the bug ship in the first place.
`TestHudiConnectorParquetColumnNamesTest` already reruns the entire smoke
suite -- including the five dynamic-filter pushdown tests at
`TestHudiSmokeTest:899,935,963,996,1034` -- with
`hudi.parquet.use-column-names=false`, and it never caught this. The reason is
the fixture: `ResourceHudiTablesInitializer.java:365` registers
`_hoodie_commit_time` and friends in the metastore `Column` list, so every
table's metastore ordinals already equal its physical ones.
#19288 established the pattern that would catch it --
`io/trino/plugin/hudi/testing/OmittedOrderingFieldHudiTablesInitializer.java:52`
(`omitted_ordering_field_mor`), a metastore whose column list deliberately
omits a column the Avro schema carries.
The tests in this file call the package-private `createPageSource` directly,
so they pin the helper but not the wiring above it (`createBaseFilePageSource`
-> `createPageSource`, and `shouldUseParquetColumnNames` reaching it from the
session).
Please add one initializer whose metastore `Column` list omits the five
`_hoodie_*` columns, register it alongside the existing
`use-column-names=false` suite, and assert that a single `WHERE <late_column> >
k` returns the right rows. That is the one test that survives a future refactor
of `createPageSource`'s signature.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -482,41 +486,135 @@ public static List<HiveColumnHandle>
remapColumnIndicesToPhysical(
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);
- }
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, 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(lookupKey(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)
+ {
+ if (predicate.isAll() || predicate.isNone()) {
+ return predicate;
+ }
+
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, caseSensitive);
Review Comment:
**Performance, minor.** Under `use-column-names=false`
`buildPhysicalIndexMap` now runs twice per split -- once at `:489` for the
projection remap and once here for the predicate remap -- and each pass does a
`toLowerCase(Locale.ROOT)` per top-level file field. On a wide table that is a
needless per-split allocation plus O(fields) string work.
Build it once in `createPageSource` and pass the map into both remaps;
`remapPredicateColumnIndicesToPhysical` only takes `fileSchema` in order to
build it. If you would rather keep the two helpers self-contained for the
`@VisibleForTesting` entry points, an overload taking the prebuilt map with the
current signature delegating to it keeps both properties.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPredicatePushdownColumnOrdinals.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi;
+
+import io.trino.filesystem.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.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 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.metastore.HiveType.HIVE_INT;
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads a base file whose physical column order does not match the
metastore's, the layout hive sync produces
+ * with {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}: the
five {@code _hoodie_*} meta columns are
+ * absent from the metastore, so every data column's metastore ordinal is five
below its physical position.
+ * <p>
+ * 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: {@code c7} grows with the
+ * row index while every other data column stays in 0..9, so a domain meant
for {@code c7} 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 these tests green against the unfixed code.
+ */
+class TestHudiPredicatePushdownColumnOrdinals
+{
+ private static final List<String> META_COLUMNS = List.of(
+ "_hoodie_commit_time",
+ "_hoodie_commit_seqno",
+ "_hoodie_record_key",
+ "_hoodie_partition_path",
+ "_hoodie_file_name");
+ 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 = fileSchema();
+ 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 : 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 testPredicateOnStaleOrdinalKeepsMatchingRows()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // The shadowed column never leaves 0..9, so a domain of "> 900"
applied to it prunes every row group
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s", PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
Review Comment:
**Cleanliness.** `testPredicateOnStaleOrdinalKeepsMatchingRows` is strictly
subsumed by `testPredicateOnStaleOrdinalStillPrunesRowGroups` directly below
it: same `projection`, same `read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY)`, and
`:169-171` repeats this test's only assertion verbatim (just a different
`.as()` label) and adds `rowCount < ROW_COUNT` on top.
Confirmed by A/B -- with the fix's call site reverted, both fail identically:
```
testPredicateOnStaleOrdinalKeepsMatchingRows:152 expected: 99L but was: 0L
testPredicateOnStaleOrdinalStillPrunesRowGroups:171 expected: 99L but was: 0L
```
Please delete it; the superset test loses nothing.
```suggestion
```
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -482,41 +486,135 @@ public static List<HiveColumnHandle>
remapColumnIndicesToPhysical(
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);
- }
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, 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(lookupKey(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.
Review Comment:
**Follow-up issue, not a change request for this PR.** The PR description
says positional mode now "inherits whatever the name-based mode already does"
with a promoted metastore type. Worth recording what that inheritance actually
is: a hard query failure, in both modes.
With a metastore that says `c7` is DOUBLE after a float -> double evolution
while the base file still holds FLOAT:
```
PR HEAD, use-column-names=true -> TrinoException: Malformed Parquet file.
Corrupted statistics for column "[c7] optional float c7"
PR HEAD, use-column-names=false -> TrinoException: ... column "[c7]
optional float c7"
pre-PR, use-column-names=false -> TrinoException: ... column "[c2]
optional int32 c2"
```
`TupleDomainParquetPredicate.getDomain` (trino-parquet 481, `:420-424`)
casts a `Float` to `Double`; the CCE is wrapped as `ParquetCorruptionException`
at `:658` and mapped to `HUDI_BAD_DATA` by `createPageSource:448`. `int ->
long` is benign (`asLong` at `:684` takes any `Number`); it is the float/double
family that breaks.
So this is not a regression -- the PR strictly improves it, from failing
while naming the wrong column to failing while naming the right one. But
hudi-trino has no schema-evolution test at all today. Could you file a GitHub
issue with this repro and link it from #18780? No change needed here.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPredicatePushdownColumnOrdinals.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi;
+
+import io.trino.filesystem.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.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 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.metastore.HiveType.HIVE_INT;
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads a base file whose physical column order does not match the
metastore's, the layout hive sync produces
+ * with {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}: the
five {@code _hoodie_*} meta columns are
+ * absent from the metastore, so every data column's metastore ordinal is five
below its physical position.
+ * <p>
+ * 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: {@code c7} grows with the
+ * row index while every other data column stays in 0..9, so a domain meant
for {@code c7} 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 these tests green against the unfixed code.
+ */
+class TestHudiPredicatePushdownColumnOrdinals
+{
+ private static final List<String> META_COLUMNS = List.of(
+ "_hoodie_commit_time",
+ "_hoodie_commit_seqno",
+ "_hoodie_record_key",
+ "_hoodie_partition_path",
+ "_hoodie_file_name");
+ 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 = fileSchema();
+ 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 : 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 testPredicateOnStaleOrdinalKeepsMatchingRows()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // The shadowed column never leaves 0..9, so a domain of "> 900"
applied to it prunes every row group
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s", PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @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
+ 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);
+
+ // A column the file does not carry has to be dropped from the
pushed-down predicate. 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();
+ }
+ }
+
+ private static MessageType fileSchema()
Review Comment:
**Cleanliness.** This `fileSchema()` and
`TestHudiPageSourceProviderTest.hudiFileSchema(int)` are the same builder --
same five meta columns, same `c0..cN` loop -- duplicated only because the tests
live in two files. Same package, same class under test
(`HudiPageSourceProvider`), and a flat merge of this file's five tests plus its
helpers into `TestHudiPageSourceProviderTest` runs the whole class in 0.457s,
so there is no fixture-cost argument for the split.
This repo's PMCs have flagged test-file proliferation before, so please fold
this file into `TestHudiPageSourceProviderTest` and collapse the two schema
builders into one. (Note if you try it: `@Nested` silently discovers zero tests
here because `@BeforeAll` and `@TempDir` are static -- use a flat merge.)
`TestHudiSmokeTest` is *not* the right home despite already calling
`createPageSource` at `:1303` with `setUseParquetColumnNames(false)`; it
extends `AbstractTestQueryFramework` and would pay a full `QueryRunner` fixture.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPredicatePushdownColumnOrdinals.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi;
+
+import io.trino.filesystem.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.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 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.metastore.HiveType.HIVE_INT;
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads a base file whose physical column order does not match the
metastore's, the layout hive sync produces
+ * with {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}: the
five {@code _hoodie_*} meta columns are
+ * absent from the metastore, so every data column's metastore ordinal is five
below its physical position.
+ * <p>
+ * 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: {@code c7} grows with the
+ * row index while every other data column stays in 0..9, so a domain meant
for {@code c7} 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 these tests green against the unfixed code.
+ */
+class TestHudiPredicatePushdownColumnOrdinals
+{
+ private static final List<String> META_COLUMNS = List.of(
+ "_hoodie_commit_time",
+ "_hoodie_commit_seqno",
+ "_hoodie_record_key",
+ "_hoodie_partition_path",
+ "_hoodie_file_name");
+ 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 = fileSchema();
+ 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 : 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 testPredicateOnStaleOrdinalKeepsMatchingRows()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // The shadowed column never leaves 0..9, so a domain of "> 900"
applied to it prunes every row group
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s", PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @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
+ 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);
+
+ // A column the file does not carry has to be dropped from the
pushed-down predicate. Pushed positionally
+ // it would land on "c5", which has no nulls at all, and every row
group would be pruned.
Review Comment:
**Cleanliness, nit.** This comment overclaims what the test proves. All five
tests in this file still pass under a variant that maps an absent column to the
`fileSchema.getFieldCount()` sentinel instead of dropping it, because
`getParquetTupleDomain` -> `getBaseColumnParquetType` reports any index at or
beyond the field count as absent and skips it anyway
(`ParquetPageSourceFactory.java:438-442`).
What this test actually pins is the second sentence: the column must not be
left on a stale *in-range* ordinal. The drop-vs-sentinel choice is pinned by
the unit test in `TestHudiPageSourceProviderTest`, not here.
```suggestion
// 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.
```
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -482,41 +486,135 @@ public static List<HiveColumnHandle>
remapColumnIndicesToPhysical(
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);
- }
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, 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(lookupKey(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)
+ {
+ if (predicate.isAll() || predicate.isNone()) {
+ return predicate;
+ }
+
+ Map<String, Integer> physicalIndexMap =
buildPhysicalIndexMap(fileSchema, caseSensitive);
+ Set<Integer> pushedPhysicalIndices = new HashSet<>();
+ Map<HiveColumnHandle, Domain> remappedDomains = new LinkedHashMap<>();
+ for (Map.Entry<HiveColumnHandle, Domain> entry :
predicate.getDomains().orElseThrow().entrySet()) {
+ Integer physicalIndex =
physicalIndexMap.get(lookupKey(entry.getKey().getBaseColumnName(),
caseSensitive));
+ if (physicalIndex == null) {
+ continue;
+ }
+ // Deduplicate on the physical index rather than on the rebuilt
handle: 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 getParquetTupleDomain the same ColumnDescriptor
twice, which it rejects by failing the
+ // split. That needs a metastore holding two such columns, which
Hive's name normalisation rules out,
+ // but keeping only the first domain is a cheap guarantee that the
read can never be made worse than
+ // pushing nothing down.
+ if (pushedPhysicalIndices.add(physicalIndex)) {
Review Comment:
**Correctness, latent.** The dedup keys on the base physical index alone,
which is broader than the collision it is guarding against.
`getParquetTupleDomain` does not skip dereference handles.
`HiveColumnHandle.getHiveType()` returns the *projection's* type, not the base
column's:
```java
// HiveColumnHandle.java:179-182
public HiveType getHiveType() {
return
hiveColumnProjectionInfo.map(HiveColumnProjectionInfo::getHiveType).orElse(baseHiveType);
}
```
so a dereference handle on a primitive subfield passes the
`getHiveType().getCategory() != PRIMITIVE` guard at
`ParquetPageSourceFactory.java:432` and gets a descriptor built for the
subfield path at `:451-460`. Two such handles on the same base column -- `c0.f
> 5` and `c0.g < 3` -- therefore produce two *distinct* `ColumnDescriptor`s, so
`buildOrThrow` would be perfectly happy, yet this dedup maps both to physical
index 5 and silently discards one.
Unreachable today (`HudiMetadata` has no `applyProjection`, so no
dereference handle exists) and the cost would be lost pruning, never a wrong
row -- so this is not a bug in this PR. But since the diff already carries a
dereference-handle test, it is worth making the guard degrade gracefully rather
than leaving a trap for whoever implements `applyProjection`:
```java
// line 542
Set<Map.Entry<Integer, Optional<HiveColumnProjectionInfo>>> pushed = new
HashSet<>();
// line 555
if (pushed.add(Map.entry(physicalIndex,
entry.getKey().getHiveColumnProjectionInfo()))) {
```
Optional -- fine to leave as is if you would rather not widen the diff, but
then please add a line to the comment above saying the guard is
base-column-only.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -397,7 +401,7 @@ static ConnectorPageSource createPageSource(
TupleDomain<ColumnDescriptor> parquetTupleDomain =
options.isIgnoreStatistics() || !enablePredicatePushDown
? TupleDomain.all()
- : getParquetTupleDomain(descriptorsByPath,
getCombinedPredicate(hudiSplit, dynamicFilter), fileSchema, useColumnNames);
+ : getParquetTupleDomain(descriptorsByPath,
getPushdownPredicate(hudiSplit, dynamicFilter, fileSchema, useColumnNames),
fileSchema, useColumnNames);
Review Comment:
**Coverage.** The remap only ever runs where `enablePredicatePushDown` is
true, and today that is only the base-file-only path -- I verified `:213`
passes `true` while the merge path `:253` and the RFC-103 log page source
`:270` both pass `false`.
That invariant is load-bearing and nothing pins it: flip `:253` to `true`
and base-file row groups whose log updates *would* make them match get pruned
before the merge ever sees them, so a MOR query silently loses rows. The new
tests call `createPageSource` directly with `enablePredicatePushDown=true`, so
they cannot catch that, and `TestHudiMorMergeModeSemantics:83-104` only filters
on the record key, which log records never change.
Please add one assertion that a predicate on a *log-updated* value still
returns the merged rows on a MOR split (or, cheaper, a direct assertion that
the merge path calls `createBaseFilePageSource` with `false`).
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java:
##########
@@ -193,6 +203,231 @@ public void testRemapColumnNotFound()
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 testRemapPredicateDropsColumnAbsentFromFile()
+ {
+ // 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
it
+ HiveColumnHandle absent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ Domain presentDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(present, presentDomain,
absent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // The absent column is dropped rather than mapped to the projection
remap's out-of-range sentinel
+ 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 testRemapPredicateWithSeveralAbsentColumnsDoesNotCollide()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle firstAbsent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle secondAbsent = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(
+ firstAbsent, Domain.singleValue(INTEGER, 1L),
+ secondAbsent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // Both would share the sentinel index, which
TupleDomain.transformKeys rejects as a duplicate key.
+ // Dropping them instead leaves nothing to push down, and the engine
still applies the filter itself.
+ assertThat(remapped.isAll()).isTrue();
+ }
Review Comment:
**Cleanliness.** `testRemapPredicateWithSeveralAbsentColumnsDoesNotCollide`
pins the same production decision as
`testRemapPredicateDropsColumnAbsentFromFile` above it -- "an absent column is
dropped, not mapped to the sentinel". Under a sentinel-instead-of-drop variant
of the helper, both fail, so the second test adds no discrimination:
```
testRemapPredicateDropsColumnAbsentFromFile:251
testRemapPredicateWithSeveralAbsentColumnsDoesNotCollide:274
```
The multiple-absent case is worth keeping, but as a widening of the existing
test rather than a second one. In
`testRemapPredicateDropsColumnAbsentFromFile`, add
```java
HiveColumnHandle secondAbsent = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
```
and include it in the `Map.of(...)` at `:246`; the existing `hasSize(1)`
assertion then covers the collision case too. Verified green on the fix and
still failing under the sentinel variant. Then delete this test:
```suggestion
```
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java:
##########
@@ -193,6 +203,231 @@ public void testRemapColumnNotFound()
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 testRemapPredicateDropsColumnAbsentFromFile()
+ {
+ // 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
it
+ HiveColumnHandle absent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ Domain presentDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(present, presentDomain,
absent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // The absent column is dropped rather than mapped to the projection
remap's out-of-range sentinel
+ 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 testRemapPredicateWithSeveralAbsentColumnsDoesNotCollide()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle firstAbsent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle secondAbsent = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(
+ firstAbsent, Domain.singleValue(INTEGER, 1L),
+ secondAbsent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // Both would share the sentinel index, which
TupleDomain.transformKeys rejects as a duplicate key.
+ // Dropping them instead leaves nothing to push down, and the engine
still applies the filter itself.
+ assertThat(remapped.isAll()).isTrue();
+ }
+
+ @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 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 = new HiveColumnHandle(
+ "c0",
+ 0,
+ HiveType.valueOf("struct<f:int>"),
+ baseType,
+ Optional.of(new HiveColumnProjectionInfo(List.of(0),
List.of("f"), HiveType.HIVE_INT, INTEGER)),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+
+ 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);
+ }
+
+ /**
+ * 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 : List.of("_hoodie_commit_time",
"_hoodie_commit_seqno", "_hoodie_record_key", "_hoodie_partition_path",
"_hoodie_file_name")) {
Review Comment:
**Cleanliness, nit.** The five meta-column names are hardcoded here and
again at `TestHudiPredicatePushdownColumnOrdinals.java:89`.
`HoodieRecord.HOODIE_META_COLUMNS`
(`hudi-common/.../HoodieRecord.java:102-104`) is exactly this list in exactly
this order, `hudi-common` is a compile-scope dependency of `hudi-trino`
(`hudi-trino/pom.xml:257`), and
`hudi-trino/src/test/java/io/trino/plugin/hudi/testing/HudiTestUtils.java:19`
already static-imports it.
```suggestion
for (String metaColumn : HOODIE_META_COLUMNS) {
```
with `import static
org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS;`. Feel free to
ignore if you would rather not add a hudi-common import to this test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
