wombatu-kun commented on code in PR #19467: URL: https://github.com/apache/hudi/pull/19467#discussion_r3859252892
########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiEvolvedColumnPredicates.java: ########## @@ -0,0 +1,388 @@ +/* + * 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.airlift.slice.Slices; +import io.trino.filesystem.local.LocalInputFile; +import io.trino.metastore.HiveType; +import io.trino.parquet.ParquetReaderOptions; +import io.trino.plugin.base.metrics.FileFormatDataSourceStats; +import io.trino.plugin.hive.HiveColumnHandle; +import io.trino.plugin.hive.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 org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +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.plugin.hive.HiveColumnHandle.createBaseColumn; +import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static io.trino.testing.MaterializedResult.materializeSourceDataStream; +import static org.apache.hudi.common.model.HoodieRecord.HOODIE_META_COLUMNS; +import static org.apache.parquet.schema.Type.Repetition.OPTIONAL; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Covers what a pushed-down predicate does to a base file written before the column it constrains was evolved. + * <p> + * Hudi lets a column's type widen and hive sync then reports the NEW type, while every base file written before the + * evolution keeps storing the old one. The parquet reader copes with that on its own -- {@code ColumnReaderFactory} + * decodes {@code FLOAT} into {@code DOUBLE} and {@code INT32} into {@code BIGINT}, and + * {@code ParquetTypeTranslator.createCoercer} handles the rest -- but the statistics do not: {@code + * TupleDomainParquetPredicate.getDomain} reads them as whatever the DOMAIN's type says, so a {@code DOUBLE} domain + * over a {@code FLOAT} column casts a {@code Float} to a {@code Double} and fails the whole split with {@code + * HUDI_BAD_DATA}. See apache/hudi#19457. + * <p> + * The fixture writes every data column as the type it had BEFORE the evolution and every handle carries the type the + * metastore reports AFTER it, which is exactly the state an unrewritten base file is in. Values grow with the row + * index so pruning stays observable: a predicate that survives pushdown reads fewer rows than the file holds, and one + * that was dropped reads all of them. Do not "simplify" that to asserting the matching rows alone -- both a working + * pushdown and no pushdown at all produce the same matching rows, since the connector's pushdown is an optimization + * and the engine re-applies the predicate above the scan. + */ +class TestHudiEvolvedColumnPredicates +{ + private static final String STABLE_COLUMN = "stable_int"; + /** Written as parquet FLOAT, reported by the metastore as double. */ + private static final String FLOAT_TO_DOUBLE_COLUMN = "evolved_double"; + /** Written as parquet INT32, reported by the metastore as bigint. */ + private static final String INT_TO_BIGINT_COLUMN = "evolved_bigint"; + /** Written as parquet INT32, reported by the metastore as string. */ + private static final String INT_TO_VARCHAR_COLUMN = "evolved_varchar"; + + 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 = preEvolutionFileSchema(); + baseFile = tempDir.resolve("evolved_base_file.parquet"); + SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema); + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(baseFile)) + .withType(schema) + .withConf(new PlainParquetConfiguration()) + .withRowGroupSize(1024L) + .withPageSize(512) + .build()) { + for (int row = 0; row < ROW_COUNT; row++) { + Group group = groupFactory.newGroup(); + for (String metaColumn : HOODIE_META_COLUMNS) { + group.append(metaColumn, metaColumn + "_" + row); + } + group.append(STABLE_COLUMN, row); + group.append(FLOAT_TO_DOUBLE_COLUMN, (float) row); + group.append(INT_TO_BIGINT_COLUMN, row); + group.append(INT_TO_VARCHAR_COLUMN, row); + writer.write(group); + } + } + // With a single row group there would be nothing to prune and every "still prunes" assertion below would + // hold without proving anything, so assert the outcome rather than the writer knobs that produce it. + assertThat(rowGroupCount(baseFile)).as("row groups written").isGreaterThan(1); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnFloatColumnEvolvedToDouble(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle evolved = column(FLOAT_TO_DOUBLE_COLUMN, HiveType.HIVE_DOUBLE, DOUBLE); + List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER), evolved); + + MaterializedResult result = read(projection, greaterThanThreshold(evolved, DOUBLE, (double) THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + // The domain cannot be matched against FLOAT statistics, so it is dropped and nothing is pruned. Before the + // fix this threw HUDI_BAD_DATA ("Corrupted statistics for column") instead of reading anything at all. + assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT); + // The read itself promotes, so the rows the engine will filter carry the widened values + assertThat(result.getMaterializedRows().get(7).getField(1)).as("promoted value of row 7").isEqualTo(7.0d); + assertThat(valuesOver(result, 1)).as("rows matching %s > %s", FLOAT_TO_DOUBLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToVarchar(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle evolved = column(INT_TO_VARCHAR_COLUMN, HiveType.HIVE_STRING, VARCHAR); + List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER), evolved); + + MaterializedResult result = read(projection, + greaterThanThreshold(evolved, VARCHAR, Slices.utf8Slice("900")), + useParquetColumnNames, DynamicFilter.EMPTY); + + // A varchar domain over an INT32 column would cast an Integer to a Slice + assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT); + assertThat(result.getMaterializedRows().get(7).getField(1)).as("promoted value of row 7").isEqualTo("7"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnIntColumnEvolvedToBigintStillPrunes(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle evolved = column(INT_TO_BIGINT_COLUMN, HiveType.HIVE_LONG, BIGINT); + List<HiveColumnHandle> projection = List.of(column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER), evolved); + + MaterializedResult result = read(projection, greaterThanThreshold(evolved, BIGINT, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + // asLong takes an Integer as happily as a Long, so this promotion is one the statistics CAN answer and the + // guard must leave it alone. This is what catches a check that drops more than it should. + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 1)).as("rows matching %s > %s after pruning", INT_TO_BIGINT_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testPredicateOnUnevolvedColumnStillPrunes(boolean useParquetColumnNames) + throws Exception + { + HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER); + List<HiveColumnHandle> projection = List.of(stable); + + MaterializedResult result = read(projection, greaterThanThreshold(stable, INTEGER, THRESHOLD), + useParquetColumnNames, DynamicFilter.EMPTY); + + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 0)).as("rows matching %s > %s after pruning", STABLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @Test + public void testOnlyTheEvolvedColumnsDomainIsDropped() + throws Exception + { + HiveColumnHandle stable = column(STABLE_COLUMN, HiveType.HIVE_INT, INTEGER); + HiveColumnHandle evolved = column(FLOAT_TO_DOUBLE_COLUMN, HiveType.HIVE_DOUBLE, DOUBLE); + List<HiveColumnHandle> projection = List.of(stable, evolved); + + MaterializedResult result = read(projection, + greaterThanThreshold(stable, INTEGER, THRESHOLD) + .intersect(greaterThanThreshold(evolved, DOUBLE, (double) THRESHOLD)), + false, DynamicFilter.EMPTY); + + // One unusable domain must not cost the whole predicate its pushdown: the stable column's domain still + // prunes, which is only visible because reading everything and reading nothing are both wrong here. + assertThat(result.getRowCount()).as("rows read out of %s", ROW_COUNT).isLessThan(ROW_COUNT); + assertThat(valuesOver(result, 0)).as("rows matching %s > %s after pruning", STABLE_COLUMN, THRESHOLD).isEqualTo(MATCHING_ROW_COUNT); + } + + @Test + public void testEvolvedColumnArrivingThroughADynamicFilter() Review Comment: Done 44c2344b18e4, as a sibling on `stable_int` asserting fewer rows read. ########## hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestParquetStatisticsDomains.java: ########## @@ -0,0 +1,294 @@ +/* + * 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.util; + +import io.trino.parquet.ParquetDataSourceId; +import io.trino.parquet.predicate.TupleDomainParquetPredicate; +import io.trino.spi.predicate.Domain; +import io.trino.spi.predicate.TupleDomain; +import io.trino.spi.type.DecimalType; +import io.trino.spi.type.Type; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.statistics.Statistics; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Types; +import org.joda.time.DateTimeZone; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.util.ParquetStatisticsDomains.dropIncomparableDomains; +import static io.trino.plugin.hudi.util.ParquetStatisticsDomains.hasComparableStatistics; +import static io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.ALL; +import static io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.NARROW; +import static io.trino.plugin.hudi.util.TestParquetStatisticsDomains.LibraryOutcome.THROWS; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DateType.DATE; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.TimestampType.TIMESTAMP_MILLIS; +import static io.trino.spi.type.TinyintType.TINYINT; +import static io.trino.spi.type.VarbinaryType.VARBINARY; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static java.nio.ByteOrder.LITTLE_ENDIAN; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64; +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96; +import static org.apache.parquet.schema.Type.Repetition.OPTIONAL; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Pins {@link ParquetStatisticsDomains#hasComparableStatistics} against the method it exists to protect. Every case + * below states BOTH what the guard decides and what {@code TupleDomainParquetPredicate.getDomain} actually does with + * the same pair, and the check is run against the real {@code getDomain}, not a description of it. A Trino upgrade + * that moves a branch therefore fails here, where the mismatch is a line of test output, instead of in a query. + * <p> + * The three library outcomes are worth telling apart, because the guard exists for two different reasons: + * <ul> + * <li>{@link LibraryOutcome#THROWS} - the cast fails and the whole split dies with {@code HUDI_BAD_DATA}. This + * is apache/hudi#19457 as reported.</li> + * <li>{@link LibraryOutcome#NARROW} on a pair the guard drops - far worse: a domain IS produced, from bytes that + * mean something else entirely, and row groups get pruned on a comparison that is simply false. Nothing fails, + * rows just go missing.</li> + * <li>{@link LibraryOutcome#ALL} - the library declines the pair itself, so dropping it changes nothing.</li> + * </ul> + * The invariant that ties them together is asserted for every case: whatever the guard keeps must be a pair the + * library reads a real range out of. + */ +class TestParquetStatisticsDomains +{ + private static final ParquetDataSourceId DATA_SOURCE_ID = new ParquetDataSourceId("test"); + private static final long VALUE_COUNT = 10; + + enum LibraryOutcome + { + /** getDomain read the statistics and returned a range narrower than "any value". */ + NARROW, + /** getDomain declined to use the statistics and returned a domain covering every value. */ + ALL, + /** getDomain failed, which the connector reports as a corrupt-statistics error over the whole split. */ + THROWS, + } + + private record TypePair(String description, Type domainType, PrimitiveType fileType, boolean comparable, LibraryOutcome outcome) + { + @Override + public String toString() + { + return description; + } + } + + private static List<TypePair> typePairs() + { + return List.of( + // A column that never evolved: the domain's type is the one the file was written with + new TypePair("boolean over BOOLEAN", BOOLEAN, plain(PrimitiveTypeName.BOOLEAN), true, NARROW), + new TypePair("integer over INT32", INTEGER, plain(INT32), true, NARROW), + new TypePair("bigint over INT64", BIGINT, plain(INT64), true, NARROW), + new TypePair("tinyint over INT32", TINYINT, plain(INT32), true, NARROW), + new TypePair("date over INT32 date", DATE, annotated(INT32, LogicalTypeAnnotation.dateType()), true, NARROW), + new TypePair("real over FLOAT", REAL, plain(FLOAT), true, NARROW), + new TypePair("double over DOUBLE", DOUBLE, plain(PrimitiveTypeName.DOUBLE), true, NARROW), + new TypePair("varchar over BINARY string", VARCHAR, annotated(BINARY, LogicalTypeAnnotation.stringType()), true, NARROW), + new TypePair("decimal(9,2) over INT32 decimal(9,2)", DecimalType.createDecimalType(9, 2), decimal(INT32, 9, 2), true, NARROW), + new TypePair("timestamp over INT64 timestamp", TIMESTAMP_MILLIS, annotated(INT64, LogicalTypeAnnotation.timestampType(false, TimeUnit.MILLIS)), true, NARROW), + new TypePair("timestamp over INT96", TIMESTAMP_MILLIS, plain(INT96), true, NARROW), + + // Promotions the statistics can answer, so pushdown must survive them + new TypePair("int -> long", BIGINT, plain(INT32), true, NARROW), + new TypePair("decimal(9,2) -> decimal(9,4)", DecimalType.createDecimalType(9, 4), decimal(INT32, 9, 2), true, NARROW), + new TypePair("decimal(20,2) -> decimal(38,4)", DecimalType.createDecimalType(38, 4), decimal(FIXED_LEN_BYTE_ARRAY, 20, 2), true, NARROW), + new TypePair("integer over a zero-scale INT32 decimal", INTEGER, decimal(INT32, 9, 0), true, NARROW), + + // Promotions that fail the split today: apache/hudi#19457 and its neighbours + new TypePair("float -> double", DOUBLE, plain(FLOAT), false, THROWS), + new TypePair("int -> double", DOUBLE, plain(INT32), false, THROWS), + new TypePair("long -> double", DOUBLE, plain(INT64), false, THROWS), + new TypePair("int -> float", REAL, plain(INT32), false, THROWS), + new TypePair("int -> string", VARCHAR, plain(INT32), false, THROWS), + new TypePair("long -> string", VARCHAR, plain(INT64), false, THROWS), Review Comment: Done 44c2344b18e4. -- 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]
