voonhous commented on code in PR #19869: URL: https://github.com/apache/hudi/pull/19869#discussion_r3966510302
########## hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/functional/TestExternalFileRecordAndSecondaryIndex.java: ########## @@ -0,0 +1,342 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hudi.client.functional; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.model.HoodieDeltaWriteStat; +import org.apache.hudi.common.model.HoodieRecordGlobalLocation; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.util.ExternalFilePathUtil; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.config.HoodieIndexConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.data.HoodieJavaRDD; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.testutils.HoodieClientTestBase; + +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.parquet.avro.AvroParquetWriter; +import org.apache.parquet.hadoop.ParquetWriter; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR; +import static org.apache.hudi.index.HoodieIndex.IndexType.INMEMORY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Asserts that the record index and the secondary index are maintained for parquet files written outside Hudi + * and registered in the table through replace commits, the way tables converted from other formats are. + * Such a table has no record key, so every row is keyed by the file path relative to the table and the row position. + * Some writers, e.g. Paimon, place their files in a directory below the partition; that prefix is part of the file id. + */ +public class TestExternalFileRecordAndSecondaryIndex extends HoodieClientTestBase { + + private static final String ID_FIELD = "id"; + private static final String NAME_FIELD = "name"; + private static final String PARTITION = "americas/brazil"; + private static final String PREFIX = "bucket-0"; + private static final HoodieSchema SCHEMA = HoodieSchema.createRecord("external", null, null, false, Arrays.asList( + HoodieSchemaField.of(ID_FIELD, HoodieSchema.create(HoodieSchemaType.INT)), + HoodieSchemaField.of(NAME_FIELD, HoodieSchema.create(HoodieSchemaType.STRING)))); + + /** Partitioned and unpartitioned tables, with the files placed directly in the partition or below a prefix. */ + static Stream<Arguments> partitionsAndPrefixes() { + return Stream.of( + Arguments.of(PARTITION, Option.empty()), + Arguments.of("", Option.empty()), + Arguments.of(PARTITION, Option.of(PREFIX)), + Arguments.of("", Option.of(PREFIX))); + } + + /** The shapes above with a global record index, plus a partitioned record index, which the secondary index does not build on. */ + static Stream<Arguments> partitionsPrefixesAndRecordIndexKinds() { + return Stream.concat( + partitionsAndPrefixes().map(arguments -> Arguments.of(arguments.get()[0], arguments.get()[1], false)), + Stream.of(Arguments.of(PARTITION, Option.empty(), true))); + } + + @ParameterizedTest + @MethodSource("partitionsPrefixesAndRecordIndexKinds") + public void testRecordAndSecondaryIndexForExternalFiles(String partitionPath, Option<String> prefix, boolean partitionedRecordIndex) throws Exception { + initExternalTable(); + HoodieWriteConfig writeConfig = writeConfig(true, partitionedRecordIndex); + writeClient = getHoodieWriteClient(writeConfig); + writeClient.setOperationType(WriteOperationType.UNKNOWN); + Option<String> recordIndexPartition = partitionedRecordIndex ? Option.of(partitionPath) : Option.empty(); + + // first commit registers one file with three rows + ExternalFile file1 = new ExternalFile(partitionPath, prefix, "file_1.parquet"); + commitReplace(Collections.singletonList(Pair.of(file1, rows(1, "alice", 2, "bob", 3, "alice"))), Collections.emptyMap()); + + HoodieBackedTableMetadata tableMetadata = tableMetadata(writeConfig); + assertRecordIndex(tableMetadata, recordIndexPartition, file1, 3); + if (!partitionedRecordIndex) { + assertEquals(mapOf("alice", setOf(file1.key(0), file1.key(2)), "bob", setOf(file1.key(1))), + readSecondaryIndex(tableMetadata, secondaryIndexPartition(), Arrays.asList("alice", "bob", "carol"))); + } + + // second commit replaces the file with one that keeps bob and adds carol + ExternalFile file2 = new ExternalFile(partitionPath, prefix, "file_2.parquet"); Review Comment: **major:** The RLI anti-join (`BaseRecordIndexer:513`) and the SI `writtenFileIds` filter (`SecondaryIndexRecordGenerationUtils:246`) are never discriminating here: only `file_1.parquet` and `file_2.parquet` are registered and no name is re-registered, so keys never collide and deleting either filter leaves every parameterization green. Re-registering an overwritten file under the same name is the case the filters exist for. Could a third commit re-register `file_1.parquet`, replacing its own file id, and assert its keys survive in both indexes? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/secondary/SecondaryIndexer.java: ########## @@ -115,7 +116,7 @@ public List<IndexPartitionAndRecords> buildUpdate(IndexUpdateContext context) { // If write operation type based on commit metadata is COMPACT or CLUSTER then no need to update, // because these operations do not change the secondary key - record key mapping. WriteOperationType operationType = context.commitMetadata().getOperationType(); - if (operationType.isInsertOverwriteOrDeletePartition()) { + if (operationType != null && operationType.isInsertOverwriteOrDeletePartition()) { throw new HoodieIndexException(String.format("Can not perform operation %s on secondary index", operationType)); } else if (operationType == WriteOperationType.COMPACT || operationType == WriteOperationType.CLUSTER) { Review Comment: **major:** This early return, and the missing CLUSTER branch in `BaseRecordIndexer.getRecordIndexAdditionalUpserts`, assume the record key survives a rewrite. A positional `<path>_<pos>` key does not, so clustering a keyless table leaves RLI pointing at the replaced file group and never inserts the new file's keys. HUDI-6443 (#9055) excluded CLUSTER from replace handling only because keys are preserved, and `HoodieDatasetBulkInsertHelper.scala:93` needs no key generator without meta fields, so the path is reachable. Could both indexers fail fast when `!hasRecordKey()` and the operation is CLUSTER, with a functional case that clusters a registered file? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java: ########## @@ -258,11 +260,29 @@ protected static <T> HoodieData<HoodieRecord> readRecordKeysFromFileSliceSnapsho engineContext.setJobStatus(activeModule, "Record Index: reading record keys from " + fileSlices.size() + " file slices"); final int parallelism = Math.min(fileSlices.size(), recordIndexMaxParallelism); + // a table without record keys, e.g. one that registers files written outside Hudi, keys every row by the path of + // its base file relative to the table and the row position, the same way the record index update path does + final boolean generateRecordKeys = !metaClient.getTableConfig().hasRecordKey(); + final int fileIdEncoding = dataWriteConfig.getWritesFileIdEncoding(); ReaderContextFactory<T> readerContextFactory = engineContext.getReaderContextFactory(metaClient); return engineContext.parallelize(fileSlices, parallelism).flatMap(partitionAndFileSlice -> { final String partition = partitionAndFileSlice.getPartitionPath(); final FileSlice fileSlice = partitionAndFileSlice.getFileSlice(); final String fileId = fileSlice.getFileId(); + long baseFileInstantTimeMillis = HoodieMetadataPayload.parseRecordIndexInstantTime(fileSlice.getBaseInstantTime()); + if (generateRecordKeys) { + checkState(fileSlice.getBaseFile().isPresent() && !fileSlice.hasLogFiles(), + "File group " + fileId + " in partition " + partition + " needs a base file and no log files to key its rows by " + + "position, because the table has no record key"); + StoragePath dataFilePath = fileSlice.getBaseFile().get().getStoragePath(); + HoodieStorage storage = metaClient.getStorage(); + Set<String> recordKeys = HoodieIOFactory.getIOFactory(storage) + .getFileFormatUtils(metaClient.getTableConfig().getBaseFileFormat()) + .readRowKeys(storage, dataFilePath, metaClient.getBasePath()); Review Comment: **major:** `readRowKeys` returns `Set<String>` (`FileFormatUtils:164`), so this branch materialises every key of a base file on the executor before streaming them out, where the keyed branch just below (line 305) streams through `CloseableMappingIterator`. Positional keys are long strings and initialize is the whole-table scan, so a 10M-row external file holds roughly a gigabyte of `String` per task. Could `FileFormatUtils` gain a `ClosableIterator<String>` counterpart of `filterRowKeys(storage, path, basePath, filter)` so this path streams too? ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/record/TestPartitionedRecordIndexer.java: ########## @@ -198,7 +198,7 @@ void testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry() { mockedMetadataUtil.when(() -> HoodieTableMetadataUtil.reduceByKeys(any(), anyInt(), anyBoolean())) .thenAnswer(invocation -> invocation.getArgument(0)); mockedBaseFileParsingUtils.when(() -> BaseFileRecordParsingUtils - .generateRLIMetadataHoodieRecordsForBaseFile(any(), any(), any(), any(), any(), anyBoolean())) + .generateRLIMetadataHoodieRecordsForBaseFile(any(), any(), any(), any(), any(), anyBoolean(), anyBoolean())) Review Comment: **major:** This test never stubs `tableConfig.hasRecordKey()`, so Mockito returns false and `BaseRecordIndexer:389` computes `generateRecordKeys = true`; the `anyBoolean()` here and in the `verify` at line 220 hide it. The pre-existing keyed partitioned-RLI update path therefore has no unit test any more. `TestRecordIndexer:197` stubs it to true and verifies `eq(false)`. Could this test do the same? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java: ########## @@ -107,7 +115,13 @@ public static <T> HoodieData<HoodieRecord> convertWriteStatsToSecondaryIndexReco HoodieSchema tableSchema; try { - tableSchema = tryResolveSchemaForTable(dataMetaClient).get(); + // a table without any completed commit, e.g. one that registers files written outside Hudi for the first time, + // only has the schema of the current commit. + tableSchema = tryResolveSchemaForTable(dataMetaClient) + .orElseGet(() -> Option.ofNullable(commitMetadata.getMetadata(HoodieCommitMetadata.SCHEMA_KEY)) Review Comment: **minor:** Not blocking. This fallback cannot fire on a Hudi-built commit: `CommitUtils:98` always stores `SCHEMA_KEY`, as `""` when the schema is null, so `HoodieSchema.parse("")` throws the generic `HoodieSchema:1705` message first, and the throw sits inside the `try` whose `catch` rewraps it. The new test pins the absent-key shape with `contains("schema")`, which `SCHEMA_KEY == "schema"` satisfies for any schema failure. Could the fallback treat an empty value as absent and sit outside the `try`, with the test using `addMetadata(SCHEMA_KEY, "")` and asserting the full message? ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/secondary/TestSecondaryIndexer.java: ########## @@ -171,6 +176,51 @@ void testBuildUpdateForCompactReturnsEmpty() { commitMetadata)).isEmpty()); } + @Test + void testBuildUpdateForReplaceCommitFromExternalWriterWithoutWriteStats() { + // files written outside Hudi are registered through replace commits without a known operation type. A fresh + // HoodieCommitMetadata carries UNKNOWN; a writer may also leave the type null. A commit that only drops files has + // no write stats but still removes the records of the replaced file groups from the index. + HoodieEngineContext engineContext = new HoodieLocalEngineContext(getDefaultStorageConf()); + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + HoodieTableConfig tableConfig = mock(HoodieTableConfig.class); + HoodieIndexMetadata indexMetadata = mock(HoodieIndexMetadata.class); + HoodieIndexDefinition indexDefinition = mock(HoodieIndexDefinition.class); + + when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig); + when(metaClient.getIndexMetadata()).thenReturn(Option.of(indexMetadata)); + when(metaClient.getTableConfig()).thenReturn(tableConfig); + when(metaClient.getIndexForMetadataPartition("secondary_index_idx")).thenReturn(Option.of(indexDefinition)); + when(tableConfig.getMetadataPartitions()).thenReturn(Collections.singleton("secondary_index_idx")); + when(indexMetadata.getIndexDefinitions()).thenReturn(Collections.singletonMap("secondary_index_idx", indexDefinition)); + when(indexDefinition.getIndexName()).thenReturn("secondary_index_idx"); + + HoodieReplaceCommitMetadata commitMetadata = new HoodieReplaceCommitMetadata(); + commitMetadata.setOperationType(null); + commitMetadata.addReplaceFileId("p1", "file_1.parquet"); + + HoodieData<HoodieRecord> deletes = engineContext.parallelize(Collections.singletonList( + HoodieMetadataPayload.createSecondaryIndexRecord("p1/file_1.parquet_0", "alice", "secondary_index_idx", true)), 1); + try (MockedStatic<SecondaryIndexRecordGenerationUtils> mockedGenerationUtils = mockStatic(SecondaryIndexRecordGenerationUtils.class)) { + mockedGenerationUtils.when(() -> SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords( + eq(Collections.emptyList()), eq("016"), eq(indexDefinition), eq(metadataConfig), eq(metaClient), eq(engineContext), eq(writeConfig), eq(commitMetadata))) + .thenReturn(deletes); + + SecondaryIndexer indexer = new SecondaryIndexer(engineContext, writeConfig, metaClient); + List<IndexPartitionAndRecords> result = indexer.buildUpdate(IndexUpdateContext.of( + "016", + mock(HoodieBackedTableMetadata.class), + Lazy.lazily(() -> mock(HoodieTableFileSystemView.class)), + commitMetadata)); + + assertEquals(1, result.size()); + assertEquals("secondary_index_idx", result.get(0).indexPartitionName()); + assertEquals(1, result.get(0).indexRecords().collectAsList().size()); + } Review Comment: **minor:** Not blocking. This pins two of the three clauses of `dropsReplacedFileGroups` (`SecondaryIndexer:151-153`): the test always sets a replaced file id, so dropping `!getPartitionToReplaceFileIds().isEmpty()` leaves it green, and `assertEquals(1, ...)` is the size of the stubbed `deletes` list. Could a second case with an empty replace map assert an empty result, so the third clause is pinned too? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java: ########## @@ -258,11 +260,29 @@ protected static <T> HoodieData<HoodieRecord> readRecordKeysFromFileSliceSnapsho engineContext.setJobStatus(activeModule, "Record Index: reading record keys from " + fileSlices.size() + " file slices"); final int parallelism = Math.min(fileSlices.size(), recordIndexMaxParallelism); + // a table without record keys, e.g. one that registers files written outside Hudi, keys every row by the path of + // its base file relative to the table and the row position, the same way the record index update path does + final boolean generateRecordKeys = !metaClient.getTableConfig().hasRecordKey(); + final int fileIdEncoding = dataWriteConfig.getWritesFileIdEncoding(); ReaderContextFactory<T> readerContextFactory = engineContext.getReaderContextFactory(metaClient); return engineContext.parallelize(fileSlices, parallelism).flatMap(partitionAndFileSlice -> { final String partition = partitionAndFileSlice.getPartitionPath(); final FileSlice fileSlice = partitionAndFileSlice.getFileSlice(); final String fileId = fileSlice.getFileId(); + long baseFileInstantTimeMillis = HoodieMetadataPayload.parseRecordIndexInstantTime(fileSlice.getBaseInstantTime()); + if (generateRecordKeys) { Review Comment: **minor:** Not blocking. Neither this guard nor its twin at `SecondaryIndexRecordGenerationUtils:356` has a test: `grep "needs a base file and no log files"` hits only the two production sites, so either could be dropped later without a failure. It also surfaces mid-commit, during metadata initialization, rather than when the index is enabled. Could a unit case with a log file on a `hasRecordKey() == false` slice assert the throw, and would it be worth rejecting RLI/SI enablement up front for such a table instead? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/partitionstats/PartitionStatsIndexer.java: ########## @@ -114,7 +114,7 @@ public List<IndexPartitionAndRecords> buildUpdate(IndexUpdateContext context) { checkState(MetadataPartitionType.COLUMN_STATS.isMetadataPartitionAvailable(dataTableMetaClient), "Column stats partition must be enabled to generate partition stats. Please enable: " + HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key()); // Generate Hoodie Pair data of partition name and list of column range metadata for all the files in that partition - boolean isDeletePartition = context.commitMetadata().getOperationType().equals(WriteOperationType.DELETE_PARTITION); + boolean isDeletePartition = WriteOperationType.DELETE_PARTITION.equals(context.commitMetadata().getOperationType()); Review Comment: **minor:** Not blocking. The null this guard now tolerates is reachable (`BaseHoodieWriteClient:149` declares `operationType` transient, and `CommitUtils.buildMetadata` passes it through unset), but nothing tests it: `TestPartitionStatsIndexer:167` only sets `UPSERT` and the functional test disables column stats. Could `testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry` be parameterized over `UPSERT` and `null`? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java: ########## @@ -258,7 +318,8 @@ public static <T> HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(Hoodi return engineContext.parallelize(fileSlices, parallelism).flatMap(partitionAndBaseFile -> { final String partition = partitionAndBaseFile.getPartitionPath(); Review Comment: **minor:** Not blocking. This local and `basePath` at line 308 are dead now that the path comes from `HoodieBaseFile::getStoragePath`; nothing in the lambda reads either. Could both be dropped? ```suggestion ``` ########## hudi-common/src/main/java/org/apache/hudi/metadata/BaseFileRecordParsingUtils.java: ########## @@ -65,16 +69,35 @@ public static Iterator<HoodieRecord> generateRLIMetadataHoodieRecordsForBaseFile String instantTime, HoodieStorage storage, boolean isPartitionedRLI) { + return generateRLIMetadataHoodieRecordsForBaseFile(basePath, writeStat, writesFileIdEncoding, instantTime, storage, isPartitionedRLI, false); Review Comment: **nit:** Feel free to ignore. This 6-arg overload has no main-source caller: the only one, `BaseRecordIndexer:413`, uses the 7-arg form, and the rest are tests. (`getRecordKeyStatuses` keeps its 6-arg form alive through `getRecordKeysDeletedOrUpdated:134`.) Could it be dropped, with the test call sites passing `false`? ########## hudi-common/src/main/java/org/apache/hudi/common/util/FileFormatUtils.java: ########## @@ -252,6 +266,22 @@ public abstract Map<String, String> readFooter(HoodieStorage storage, boolean re */ public abstract Set<Pair<String, Long>> filterRowKeys(HoodieStorage storage, StoragePath filePath, Set<String> filter); + /** + * Read the rowKey list matching the given filter, from the given data file. + * If the filter is empty, then this will return all the row keys and corresponding positions. + * Formats that support data files written outside Hudi override this method to key every row of such a file, + * which carries no record key, by the file path relative to the table base path and the row position. + * + * @param storage {@link HoodieStorage} instance. + * @param filePath the data file path. + * @param basePath the table base path. + * @param filter record keys filter. + * @return set of pairs of row key and position matching candidateRecordKeys. + */ + public Set<Pair<String, Long>> filterRowKeys(HoodieStorage storage, StoragePath filePath, StoragePath basePath, Set<String> filter) { + return filterRowKeys(storage, filePath, filter); Review Comment: **minor:** Not blocking. This default drops `basePath` and delegates to the record-key read, so a keyless table whose `getBaseFileFormat()` is not parquet gets the wrong keys silently: `BaseRecordIndexer:280` picks the implementation by table format, `OrcUtils:178` then fails with `Couldn't find row keys`, and HFile/Lance/Vortex return meta-field keys. Could the base implementation name the gap instead? ```suggestion throw new UnsupportedOperationException("Positional row keys are only supported for parquet base files, not " + getClass().getSimpleName()); ``` ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java: ########## @@ -205,6 +227,44 @@ public static <T> HoodieData<HoodieRecord> convertWriteStatsToSecondaryIndexReco return HoodieTableMetadataUtil.reduceByKeys(secondaryIndexRecords, parallelism, false); } + /** + * Generates delete records for every record of the file groups that the given replace commit replaces without + * writing to them again, e.g. files written outside Hudi that are superseded by newer files. Records that the same Review Comment: **nit:** Feel free to ignore. This describes a same-commit collision positional keys cannot produce: the `writtenFileIds` filter below removes any file id written again, and a positional key embeds the file path, so a delete and an insert never share a key here. The `reduceByKeys` precedence only matters for a keyed table taking the same unknown-op branch. Could the sentence name that case instead? -- 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]
