wombatu-kun commented on code in PR #19456: URL: https://github.com/apache/hudi/pull/19456#discussion_r3698924423
########## 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: Done 40261aca65ce ########## 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: Done 40261aca65ce - the surviving test is now testRemapPredicateDropsColumnsAbsentFromFile, plural, since it drops two. -- 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]
