wombatu-kun commented on code in PR #19298: URL: https://github.com/apache/hudi/pull/19298#discussion_r3690268822
########## 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: `isLessThanOrEqualTo` holds even when nothing is pruned, so this does not show that p2 was skipped. Assert the exact split count, the way the partition-pruning checks in `TestHudiSmokeTest` do. ########## hudi-trino/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java: ########## @@ -67,6 +84,22 @@ public HudiSnapshotDirectoryLister( IndexSupportFactory.createIndexSupport(tableHandle, lazyMetaClient, lazyTableMetadata, tableHandle.getRegularPredicates(), session) : Optional.empty(); } + /** + * Builds a file system view that lists files directly from storage, bypassing the metadata table. + * Used as the fallback when the metadata-table-backed view cannot be loaded (e.g. an MDT read + * failure); it lists lazily per partition, so no {@code loadAllPartitions()} here. + */ + private static HoodieTableFileSystemView createDirectListingFileSystemView(HoodieTableMetaClient metaClient) Review Comment: `HoodieTableFileSystemView.fileListingBasedFileSystemView` already builds a view over a `FileSystemBackedTableMetadata`, so the `HoodieMetadataConfig.enable(false)` plus `NativeTableMetadataFactory` detour is redundant here and its `reuse=true` argument is dead on that branch. Call it with the same completed-commits timeline `HudiUtil.getFileSystemView` uses. ########## 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: Both new guards - the catch in `prunePartitionsSafely` and the direct-listing fallback in `HudiSnapshotDirectoryLister` - have no test; every case here goes through the path where the MDT now reads cleanly. Could the initializer corrupt the MDT log files in the temp dir before `copyDir` so the fallbacks are pinned too? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/query/HudiSnapshotDirectoryLister.java: ########## @@ -54,9 +58,22 @@ public HudiSnapshotDirectoryLister( this.lazyFileSystemView = Lazy.lazily(() -> { HoodieTimer timer = HoodieTimer.start(); HoodieTableMetaClient metaClient = tableHandle.getMetaClient(); - HoodieTableFileSystemView fileSystemView = getFileSystemView(lazyTableMetadata.get(), metaClient); - if (enableMetadataTable) { - fileSystemView.loadAllPartitions(); + HoodieTableFileSystemView fileSystemView = null; + try { + fileSystemView = getFileSystemView(lazyTableMetadata.get(), metaClient); + if (enableMetadataTable) { + fileSystemView.loadAllPartitions(); + } + } + catch (Exception e) { + // A failure here is a metadata-table read failure (the metastore/table itself is + // fine), so fall back to direct file listing instead of failing the query. + if (fileSystemView != null && !fileSystemView.isClosed()) { + fileSystemView.close(); Review Comment: `fileSystemView.close()` reaches `AbstractTableFileSystemView.closeResources`, which closes the `HoodieTableMetadata` the view was constructed with - here that is the shared `lazyTableMetadata` that `HudiBackgroundSplitLoader` and the index supports still read through the same `Lazy`, and `Lazy` never rebuilds it. Drop the close, since nothing else in the connector closes that instance today. ########## 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: These five `_hoodie_*` columns are already declared as `HUDI_META_COLUMNS` in `AbstractMergerHudiTablesInitializer`. Widen that constant's visibility and reuse it here - follow-up, not a blocker. -- 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]
