voonhous commented on code in PR #19298: URL: https://github.com/apache/hudi/pull/19298#discussion_r3690573000
########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiUncompactedMetadataTable.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.Session; +import io.trino.plugin.hudi.testing.UncompactedMetadataHudiTablesInitializer; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.MaterializedResult; +import io.trino.testing.QueryRunner; +import org.junit.jupiter.api.Test; + +import static io.trino.plugin.hudi.testing.UncompactedMetadataHudiTablesInitializer.TABLE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Regression tests for apache/hudi#19279: queries on tables whose metadata table (MDT) has + * UNCOMPACTED delta commits. Those deltas are native HFILE log files, which the connector previously + * rejected ("Native HFILE log files are not supported..."), failing every query through the unguarded + * partition-stats pruning path. The table written by {@link UncompactedMetadataHudiTablesInitializer} + * keeps its MDT deliberately uncompacted (the zip fixtures always compact after every commit), so the + * queries below only succeed if the connector reads HFILE log deltas in the MDT's + * {@code files}/{@code column_stats}/{@code partition_stats} partitions. + */ +public class TestHudiUncompactedMetadataTable Review Comment: Added. The initializer now writes a corrupted twin table (hudi_corrupted_mdt_pt_cow) whose MDT log files are damaged before copyDir, and two new tests pin the direct-listing fallback and the unpruned-pruning fallback on it. One subtlety: the damage targets the HFile trailer fields inside the block content, not the log block framing -- framing damage is classified as a HoodieCorruptBlock and skipped silently (records vanish, nothing throws), and bytes near EOF are only trailer padding. The helper has a comment on the offsets. ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiUncompactedMetadataTable.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.Session; +import io.trino.plugin.hudi.testing.UncompactedMetadataHudiTablesInitializer; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.MaterializedResult; +import io.trino.testing.QueryRunner; +import org.junit.jupiter.api.Test; + +import static io.trino.plugin.hudi.testing.UncompactedMetadataHudiTablesInitializer.TABLE_NAME; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Regression tests for apache/hudi#19279: queries on tables whose metadata table (MDT) has + * UNCOMPACTED delta commits. Those deltas are native HFILE log files, which the connector previously + * rejected ("Native HFILE log files are not supported..."), failing every query through the unguarded + * partition-stats pruning path. The table written by {@link UncompactedMetadataHudiTablesInitializer} + * keeps its MDT deliberately uncompacted (the zip fixtures always compact after every commit), so the + * queries below only succeed if the connector reads HFILE log deltas in the MDT's + * {@code files}/{@code column_stats}/{@code partition_stats} partitions. + */ +public class TestHudiUncompactedMetadataTable + extends AbstractTestQueryFramework +{ + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .setDataLoader(new UncompactedMetadataHudiTablesInitializer()) + .build(); + } + + @Test + public void testSnapshotReadWithUncompactedMetadataTable() + { + // MDT-backed file listing must read the files partition's HFILE log deltas + assertQuery( + mdtEnabled(), + "SELECT id, name, price FROM " + TABLE_NAME + " ORDER BY id", + "VALUES ('k1', 'k1_c3', CAST(15 AS BIGINT)), ('k2', 'k2_c1', 1000), ('k3', 'k3_c2', 20), ('k4', 'k4_c2', 2000)"); + assertThat(computeScalar(mdtEnabled(), "SELECT count(*) FROM " + TABLE_NAME)).isEqualTo(4L); + } + + @Test + public void testResultsMatchWithMetadataTableDisabled() + { + String query = "SELECT id, name, price, part_col FROM " + TABLE_NAME + " ORDER BY id"; + MaterializedResult withMdt = getQueryRunner().execute(mdtEnabled(), query); + MaterializedResult withoutMdt = getQueryRunner().execute(mdtDisabled(), query); + assertThat(withMdt.getMaterializedRows()).isEqualTo(withoutMdt.getMaterializedRows()); + } + + @Test + public void testPartitionStatsIndexPruningOverUncompactedStats() + { + // The exact crash from the issue: partition-stats pruning reads the partition_stats MDT + // partition, whose deltas are uncompacted HFILE log files. Partition p2 holds prices + // [1000, 2000], so `price < 100` lets the index prune it entirely. + Session session = SessionBuilder.from(getSession()) + .withMdtEnabled(true) + .withColStatsIndexEnabled(false) + .withRecordLevelIndexEnabled(false) + .withSecondaryIndexEnabled(false) + .withPartitionStatsIndexEnabled(true) + .build(); + MaterializedResult pruned = getQueryRunner().execute(session, + "SELECT id, price FROM " + TABLE_NAME + " WHERE price < 100"); + assertThat(pruned.getMaterializedRows()).hasSize(2); + + MaterializedResult unpruned = getQueryRunner().execute(mdtEnabled(), + "SELECT id, price FROM " + TABLE_NAME); + // Pruning must not scan more than the full table does; correctness is asserted above + assertThat(pruned.getStatementStats().get().getTotalSplits()) + .isLessThanOrEqualTo(unpruned.getStatementStats().get().getTotalSplits()); Review Comment: Tightened, though not to a literal: the pruned scan must now have exactly the split count of a metastore-pruned part_col = 'p1' scan, and strictly fewer than the full scan. Unlike the zip fixtures TestHudiSmokeTest asserts on, this table is written at test runtime, so file groups per partition depend on the write client's small-file packing; a hardcoded number would couple the test to that. ########## hudi-trino/src/test/java/io/trino/plugin/hudi/testing/UncompactedMetadataHudiTablesInitializer.java: ########## @@ -0,0 +1,283 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi.testing; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.trino.filesystem.Location; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.TrinoFileSystemFactory; +import io.trino.metastore.Column; +import io.trino.metastore.HiveMetastore; +import io.trino.metastore.HiveMetastoreFactory; +import io.trino.metastore.Partition; +import io.trino.metastore.PartitionStatistics; +import io.trino.metastore.PartitionWithStatistics; +import io.trino.metastore.PrincipalPrivileges; +import io.trino.metastore.StorageFormat; +import io.trino.metastore.Table; +import io.trino.plugin.hudi.HudiConnector; +import io.trino.spi.security.ConnectorIdentity; +import io.trino.testing.QueryRunner; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.hudi.client.HoodieJavaWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.client.common.HoodieJavaEngineContext; +import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.model.HoodieAvroPayload; +import org.apache.hudi.common.model.HoodieAvroRecord; +import org.apache.hudi.common.model.HoodieKey; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.marker.MarkerType; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieCompactionConfig; +import org.apache.hudi.config.HoodieIndexConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.index.HoodieIndex; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static com.google.common.io.MoreFiles.deleteRecursively; +import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE; +import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT; +import static io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS; +import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS; +import static io.trino.metastore.HiveType.HIVE_LONG; +import static io.trino.metastore.HiveType.HIVE_STRING; +import static io.trino.plugin.hive.HivePartitionManager.extractPartitionValues; +import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE; +import static java.nio.file.Files.createTempDirectory; + +/** + * Creates a partitioned COW table at test runtime with an ENABLED, UNCOMPACTED metadata table (MDT): + * {@code hoodie.metadata.compact.max.delta.commits} is set high (the zip fixtures use {@code =1}, so + * their MDTs are always freshly compacted) and several commits are written, leaving the MDT's + * {@code files}/{@code column_stats}/{@code partition_stats} partitions with native HFILE LOG deltas + * that the connector must read at query time (issue apache/hudi#19279). + * <p> + * Note: MDT writing here is fully native -- HFILE base files and log blocks are written via hudi-io's + * pure-Java {@code HFileWriterImpl}, so no hbase dependency is involved (the "requires hbase" note in + * older initializers is stale). + * <p> + * Data layout (partitions {@code part_col=p1} / {@code part_col=p2}, hive-style paths so the MDT + * partition listing and the metastore agree on names): + * <pre> + * commit 1 (insert): k1(p1, price 10, ts 100), k2(p2, price 1000, ts 100) + * commit 2 (insert): k3(p1, price 20, ts 200), k4(p2, price 2000, ts 200) + * commit 3 (upsert): k1(p1, price 15, ts 300) + * </pre> + * Final rows: k1=15, k2=1000, k3=20, k4=2000. Partition p1 holds prices [15, 20] and p2 holds + * [1000, 2000], so a predicate like {@code price < 100} lets the partition-stats index prune p2. + */ +public class UncompactedMetadataHudiTablesInitializer + implements HudiTablesInitializer +{ + public static final String TABLE_NAME = "hudi_uncompacted_mdt_pt_cow"; + + private static final String RECORD_KEY_FIELD = "id"; + private static final String PARTITION_FIELD = "part_col"; + private static final String ORDERING_FIELD = "ts"; + private static final List<String> PARTITION_PATHS = ImmutableList.of(PARTITION_FIELD + "=p1", PARTITION_FIELD + "=p2"); + + private static final List<Column> DATA_COLUMNS = ImmutableList.of( + new Column("_hoodie_commit_time", HIVE_STRING, Optional.empty(), Map.of()), Review Comment: Did it here since it was small: HUDI_META_COLUMNS in AbstractMergerHudiTablesInitializer is package-private now and the new initializer builds its DATA_COLUMNS on top of it. The Tpch and Resource initializers still carry identical private copies; left those for a separate sweep. -- 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]
