voonhous commented on code in PR #19869: URL: https://github.com/apache/hudi/pull/19869#discussion_r3966813139
########## 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"); + commitReplace(Collections.singletonList(Pair.of(file2, rows(2, "bob", 4, "carol"))), + Collections.singletonMap(partitionPath, Collections.singletonList(file1.fileId()))); + + tableMetadata = tableMetadata(writeConfig); + assertTrue(readRecordIndex(tableMetadata, recordIndexPartition, file1.keys(3)).isEmpty()); + assertRecordIndex(tableMetadata, recordIndexPartition, file2, 2); + if (!partitionedRecordIndex) { + assertEquals(mapOf("bob", setOf(file2.key(0)), "carol", setOf(file2.key(1))), + readSecondaryIndex(tableMetadata, secondaryIndexPartition(), Arrays.asList("alice", "bob", "carol"))); + } + } + + @ParameterizedTest + @MethodSource("partitionsAndPrefixes") + public void testIndexesAreBuiltFromRegisteredFilesWhenEnabledLater(String partitionPath, Option<String> prefix) throws Exception { + initExternalTable(); + // the first file is registered while both indexes are off + writeClient = getHoodieWriteClient(writeConfig(false, false)); + writeClient.setOperationType(WriteOperationType.UNKNOWN); + ExternalFile file1 = new ExternalFile(partitionPath, prefix, "file_1.parquet"); + commitReplace(Collections.singletonList(Pair.of(file1, rows(1, "alice", 2, "bob", 3, "alice"))), Collections.emptyMap()); + + // the second file is registered with both indexes on, which first builds them from the registered file + HoodieWriteConfig indexedWriteConfig = writeConfig(true, false); + writeClient = getHoodieWriteClient(indexedWriteConfig); + writeClient.setOperationType(WriteOperationType.UNKNOWN); + ExternalFile file2 = new ExternalFile(partitionPath, prefix, "file_2.parquet"); + commitReplace(Collections.singletonList(Pair.of(file2, rows(4, "carol"))), Collections.emptyMap()); + + HoodieBackedTableMetadata tableMetadata = tableMetadata(indexedWriteConfig); + assertRecordIndex(tableMetadata, Option.empty(), file1, 3); + assertRecordIndex(tableMetadata, Option.empty(), file2, 1); + String secondaryIndexPartition = secondaryIndexPartition(); + assertEquals(mapOf("alice", setOf(file1.key(0), file1.key(2)), "bob", setOf(file1.key(1)), "carol", setOf(file2.key(0))), + readSecondaryIndex(tableMetadata, secondaryIndexPartition, Arrays.asList("alice", "bob", "carol"))); + + // the third commit only drops the first file + commitReplace(Collections.emptyList(), Collections.singletonMap(partitionPath, Collections.singletonList(file1.fileId()))); + + tableMetadata = tableMetadata(indexedWriteConfig); + assertTrue(readRecordIndex(tableMetadata, Option.empty(), file1.keys(3)).isEmpty()); + assertRecordIndex(tableMetadata, Option.empty(), file2, 1); + assertEquals(mapOf("carol", setOf(file2.key(0))), + readSecondaryIndex(tableMetadata, secondaryIndexPartition, Arrays.asList("alice", "bob", "carol"))); + } + + /** Tables registered from other formats carry neither meta fields nor record key fields. */ + private void initExternalTable() throws IOException { + Properties tableProperties = new Properties(); + tableProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + tableProperties.setProperty(HoodieTableConfig.RECORDKEY_FIELDS.key(), ""); + initMetaClient(tableProperties); + } + + private HoodieWriteConfig writeConfig(boolean withIndexes, boolean partitionedRecordIndex) { + HoodieMetadataConfig.Builder metadataConfig = HoodieMetadataConfig.newBuilder() + .enable(true) + .withMetadataIndexColumnStats(false); + if (withIndexes && partitionedRecordIndex) { + // the secondary index builds on the global record index only + metadataConfig.withEnableRecordLevelIndex(true); + } else if (withIndexes) { + metadataConfig.withEnableGlobalRecordLevelIndex(true) + .withSecondaryIndexEnabled(true) + .withSecondaryIndexForColumn(NAME_FIELD); + } + return HoodieWriteConfig.newBuilder() + .withPath(metaClient.getBasePath()) + .withSchema(SCHEMA.toString()) + .withPopulateMetaFields(false) + // file ids of external files are file names, not UUIDs, so the record index stores them as raw strings + .withWritesFileIdEncoding(1) + .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(INMEMORY).build()) + .withEmbeddedTimelineServerEnabled(false) + .withMetadataConfig(metadataConfig.build()) + .build(); + } + + /** + * Registers the given files in one replace commit, the way an external writer does: the files are written to + * their location below the table, and the commit records them with the external file marker. + */ + private void commitReplace(List<Pair<ExternalFile, List<Pair<Integer, String>>>> newFiles, + Map<String, List<String>> partitionToReplacedFileIds) throws IOException { + String instantTime = writeClient.startCommit(HoodieTimeline.REPLACE_COMMIT_ACTION, metaClient); + List<WriteStatus> writeStatuses = new ArrayList<>(); + for (Pair<ExternalFile, List<Pair<Integer, String>>> fileAndRows : newFiles) { + ExternalFile file = fileAndRows.getLeft(); + long fileSize = writeParquetFile(new Path(metaClient.getBasePath().toString(), file.relativePath()), fileAndRows.getRight()); + WriteStatus writeStatus = new WriteStatus(); + writeStatus.setFileId(file.fileId()); + writeStatus.setPartitionPath(file.partitionPath); + HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat(); + writeStat.setFileId(file.fileId()); + writeStat.setPath(file.markedPath(instantTime)); + writeStat.setPartitionPath(file.partitionPath); + writeStat.setNumWrites(fileAndRows.getRight().size()); + writeStat.setNumInserts(fileAndRows.getRight().size()); + writeStat.setTotalWriteBytes(fileSize); + writeStat.setFileSizeInBytes(fileSize); + writeStatus.setStat(writeStat); + writeStatuses.add(writeStatus); + } + metaClient.getActiveTimeline().transitionReplaceRequestedToInflight( + INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime), Option.empty()); + writeClient.commit(instantTime, jsc.parallelize(writeStatuses, 1), Option.empty(), + HoodieTimeline.REPLACE_COMMIT_ACTION, partitionToReplacedFileIds); + metaClient = HoodieTableMetaClient.reload(metaClient); + } + + private long writeParquetFile(Path path, List<Pair<Integer, String>> rows) throws IOException { + Configuration conf = metaClient.getStorageConf().unwrapAs(Configuration.class); + try (ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(path) + .withSchema(SCHEMA.toAvroSchema()).withConf(conf).build()) { + for (Pair<Integer, String> row : rows) { + GenericRecord record = new GenericData.Record(SCHEMA.toAvroSchema()); + record.put(ID_FIELD, row.getLeft()); + record.put(NAME_FIELD, row.getRight()); + writer.write(record); + } + } + return path.getFileSystem(conf).getFileStatus(path).getLen(); + } + + private HoodieBackedTableMetadata tableMetadata(HoodieWriteConfig writeConfig) { + return new HoodieBackedTableMetadata(context, metaClient.getStorage(), writeConfig.getMetadataConfig(), writeConfig.getBasePath(), true); + } + + private String secondaryIndexPartition() { + return metaClient.getIndexMetadata().get().getIndexDefinitions().keySet().stream() + .filter(indexName -> indexName.startsWith(HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX)) + .findFirst().get(); + } + + /** Asserts that every row of the file is in the record index and points at the file. */ + private void assertRecordIndex(HoodieBackedTableMetadata tableMetadata, Option<String> recordIndexPartition, ExternalFile file, int rowCount) { + Map<String, HoodieRecordGlobalLocation> locations = readRecordIndex(tableMetadata, recordIndexPartition, file.keys(rowCount)); + assertEquals(rowCount, locations.size()); + locations.values().forEach(location -> { + assertEquals(file.partitionPath, location.getPartitionPath()); + assertEquals(file.fileId(), location.getFileId()); + }); + } + + private Map<String, HoodieRecordGlobalLocation> readRecordIndex(HoodieBackedTableMetadata tableMetadata, Option<String> recordIndexPartition, + List<String> recordKeys) { + return tableMetadata.readRecordIndexLocationsWithKeys(HoodieJavaRDD.of(jsc.parallelize(recordKeys, 1)), recordIndexPartition).collectAsList().stream() Review Comment: **blocker:** As built, the secondary index cannot be consumed. Spark data skipping re-derives the fileId from the on-disk name (`SecondaryIndexSupport.scala:92` via `FSUtils.getFileIdFromFilePath`, i.e. `name.split("_", 2)[0]`), so `file_1.parquet` yields `file` while the RLI location holds `file_1.parquet` (line 265). The candidate set is empty, `HoodieFileIndex.scala:303` drops every slice, and `WHERE <si-col> = v` returns zero rows under the default `hoodie.enable.data.skipping=true`. Flink matches on `fileSlice.getFileGroupId()` (`BaseRecordLevelIndex.java:117`). Could the Spark pruning carry the slice's own fileId instead of re-parsing the path, with one `spark.read ... where(<si-col> = v)` case here asserting the rows and the pruned files? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java: ########## @@ -478,11 +500,51 @@ private HoodieData<HoodieRecord> getRecordIndexAdditionalUpserts( } else if (operationType == WriteOperationType.DELETE_PARTITION) { // all records from the target partition(s) to be deleted from RLI return getRecordIndexReplacedRecords((HoodieReplaceCommitMetadata) commitMetadata, fsView); + } else if (commitMetadata instanceof HoodieReplaceCommitMetadata && WriteOperationType.isUnknown(operationType)) { Review Comment: **major:** This gate, and the two SI gates, are false during async index catch-up: both catch-up tasks (`WriteStatBasedIndexingCatchupTask:56`, `RecordBasedIndexingCatchupTask:56`) call `readCommitMetadata(instant)` for every action, and the SerDe only builds a `HoodieReplaceCommitMetadata` when asked for that class, so a replayed registration commit arrives without `partitionToReplaceFileIds` and the replaced file groups' rows stay in both indexes for good. `CREATE INDEX` requires OCC, so a registration landing in the catch-up window is the normal case; on master the INSERT_OVERWRITE arm at least fails loudly there. Could both tasks use `TimelineUtils.getCommitMetadata(instant, timeline)` (line 350, already action-dispatching), with one test replaying a registration commit through the catch-up task? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java: ########## @@ -478,11 +500,51 @@ private HoodieData<HoodieRecord> getRecordIndexAdditionalUpserts( } else if (operationType == WriteOperationType.DELETE_PARTITION) { // all records from the target partition(s) to be deleted from RLI return getRecordIndexReplacedRecords((HoodieReplaceCommitMetadata) commitMetadata, fsView); + } else if (commitMetadata instanceof HoodieReplaceCommitMetadata && WriteOperationType.isUnknown(operationType)) { + // a replace commit without a known operation type registers files written outside Hudi. The replaced file groups + // are dropped without their records being rewritten under the same key, so the records of the replaced base files + // are deleted from RLI unless this commit wrote the same key again. + HoodiePairData<HoodieKey, HoodieRecord> replacedRecordsByKey = getRecordIndexReplacedFileGroupRecords((HoodieReplaceCommitMetadata) commitMetadata, fsView) + .mapToPair(record -> Pair.of(record.getKey(), record)); + HoodiePairData<HoodieKey, HoodieRecord> writtenRecordsByKey = updatesFromWriteStatuses + .mapToPair(record -> Pair.of(record.getKey(), record)); + return replacedRecordsByKey.leftOuterJoin(writtenRecordsByKey) + .values() + .filter(replacedRecordAndRewrite -> !replacedRecordAndRewrite.getRight().isPresent()) + .map(Pair::getLeft); } else { return engineContext.emptyHoodieData(); } } + /** + * Reads the record keys of the latest base file of every file group replaced by the given commit and + * returns a delete record for each of them. + */ + private HoodieData<HoodieRecord> getRecordIndexReplacedFileGroupRecords(HoodieReplaceCommitMetadata replaceCommitMetadata, Lazy<HoodieTableFileSystemView> fsView) { + List<Pair<String, HoodieBaseFile>> replacedBaseFiles = replaceCommitMetadata.getPartitionToReplaceFileIds().entrySet().stream() + .flatMap(partitionAndFileIds -> partitionAndFileIds.getValue().stream() + .map(fileId -> fsView.get().getLatestBaseFile(partitionAndFileIds.getKey(), fileId)) + .filter(Option::isPresent) Review Comment: **minor:** Not blocking. A replaced file group with no base file in the view is dropped here by `.filter(Option::isPresent)` with no log line, and the SI twin at `SecondaryIndexRecordGenerationUtils:258` returns an empty iterator the same way, so a stale index has no diagnostic trail. Could both `log.warn` the partition and file id when a replaced file group has no readable base file? ########## 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(); final FileSlice fileSlice = partitionAndBaseFile.getFileSlice(); - Option<StoragePath> dataFilePath = Option.ofNullable(fileSlice.getBaseFile().map(baseFile -> FSUtils.getAbsoluteFilePath(basePath, partition, baseFile.getFileName())).orElseGet(null)); + // the storage path keeps the directory prefix of a file written outside Hudi, which its file name alone loses + Option<StoragePath> dataFilePath = fileSlice.getBaseFile().map(HoodieBaseFile::getStoragePath); HoodieSchema readerSchema; if (dataFilePath.isPresent()) { readerSchema = HoodieIOFactory.getIOFactory(metaClient.getStorage()) Review Comment: **minor:** Not blocking. `tableSchema` is resolved eagerly at line 311 but only consumed at line 329, the no-base-file branch the keyless guard forbids; here the reader schema comes from the base file. `TableSchemaResolver.getTableSchema()` can still throw on that path, and its data-file fallback resolves the marked `_hudiext` path, which does not exist on storage. Could the resolution be made lazy, so initializing the index on a keyless table does not depend on a schema it never reads? ########## hudi-common/src/main/java/org/apache/hudi/common/util/ExternalFilePathUtil.java: ########## @@ -93,6 +93,32 @@ public static boolean isExternallyCreatedFile(String fileName) { return fileName.endsWith(EXTERNAL_FILE_SUFFIX); } + /** + * Returns the path of a base file relative to its partition, as the file exists on storage. + * For an external file name, the commit time and the external file marker are stripped and the file group + * prefix, if any, is restored. For example, "data.parquet_123_fg%3Dbucket-0_hudiext" returns "bucket-0/data.parquet". + * A file name that was not created externally is returned as is. + * + * @param fileName The file name as recorded in the commit metadata + * @return The path of the file relative to its partition + */ + public static String getFilePathInPartition(String fileName) { Review Comment: **nit:** Feel free to ignore, pre-existing code that this call site newly reaches from the RLI update path. `getOriginalFileName` (line 157) does `lastIndexOf('_', markerEnd - 1)` and feeds -1 into `substring`, so a name ending in `_hudiext` with no earlier `_` throws a raw `StringIndexOutOfBoundsException`; and `getFullPathOfPartition` (line 170) slices `prefix.length() + 1` characters off the parent without checking it ends with the prefix, so an empty `fg%3D` prefix eats one character. Could the first return `Option.empty()` or a `HoodieException` naming the file, and the second `checkState(parent.endsWith(prefix))`? ########## 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: One thing to watch when adding that case: a same-name re-registration resolves the previous and current slice to the same physical file (`HoodieBaseFile:86` strips the marker; `SecondaryIndexRecordGenerationUtils:166` builds the current slice from the stat path), so the SI sees identical before/after maps and the RLI's replaced-file read returns the new file's keys, leaving a shorter re-registration's tail keys dangling. Would it be worth stating in the `getRecordIndexReplacedFileGroupRecords` javadoc that replaced file ids must be distinct from written ones, or adding an RLI-side `checkState` (the SI side already filters `writtenFileIds`)? -- 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]
