hudi-agent commented on code in PR #19456:
URL: https://github.com/apache/hudi/pull/19456#discussion_r3698449564
##########
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);
+ }
Review Comment:
🤖 nit: `lookupKey` describes the output format but not what the method
actually does — could you rename it to something like `normalizeColumnName`?
That makes the `caseSensitive` parameter and the caller sites at lines 499,
548, and 573 read more naturally at a glance.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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()
+ {
+ List<org.apache.parquet.schema.Type> fields = new ArrayList<>();
+ for (String metaColumn : META_COLUMNS) {
+ fields.add(Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY,
OPTIONAL).as(LogicalTypeAnnotation.stringType()).named(metaColumn));
+ }
+ for (int column = 0; column < DATA_COLUMN_COUNT; column++) {
+ fields.add(Types.primitive(PrimitiveType.PrimitiveTypeName.INT32,
OPTIONAL).named("c" + column));
+ }
+ 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 {@link #META_COLUMNS} short of its
physical position.
+ */
+ private static HiveColumnHandle dataColumn(String columnName)
+ {
+ return createBaseColumn(columnName, parseInt(columnName.substring(1)),
HIVE_INT, INTEGER, REGULAR, Optional.empty());
+ }
+
+ 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
Review Comment:
🤖 nit: `parseInt(columnName.substring(1))` silently assumes the column name
is always `"c" + digit(s)` — it would throw an unhelpful
`NumberFormatException` if a caller ever passes a meta column name like
`"_hoodie_record_key"`. Could you add a brief inline comment (or rename the
helper to `dataColumnByIndex`) to make it clear this is only meant for `c0..cN`
style names?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
