cshuo commented on code in PR #18372: URL: https://github.com/apache/hudi/pull/18372#discussion_r3542056396
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/record/BaseRecordIndexer.java: ########## @@ -0,0 +1,484 @@ +/* + * 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.metadata.index.record; + +import org.apache.hudi.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.config.HoodieConfig; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieReaderContext; +import org.apache.hudi.common.engine.ReaderContextFactory; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieDeltaWriteStat; +import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieReplaceCommitMetadata; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.CollectionUtils; +import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.common.util.VisibleForTesting; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.io.storage.HoodieAvroFileReader; +import org.apache.hudi.io.storage.HoodieIOFactory; +import org.apache.hudi.metadata.BaseFileRecordParsingUtils; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.index.model.DataPartitionAndRecords; +import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords; +import org.apache.hudi.metadata.model.FileSliceAndPartition; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaCache; +import org.apache.hudi.common.schema.HoodieSchemaUtils; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.read.HoodieFileGroupReader; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.collection.CloseableMappingIterator; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.internal.schema.InternalSchema; +import org.apache.hudi.internal.schema.utils.SerDeHelper; +import org.apache.hudi.metadata.HoodieMetadataPayload; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.metadata.index.BaseIndexer; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.HoodieStorageUtils; +import org.apache.hudi.storage.StorageConfiguration; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.util.Lazy; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.schema.HoodieSchemaUtils.getRecordKeySchema; +import static org.apache.hudi.common.util.ValidationUtils.checkState; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.readRecordKeysFromBaseFiles; +import static org.apache.hudi.metadata.MetadataPartitionType.RECORD_INDEX; + +/** + * Base implementation for record-index. + */ +@Slf4j +public abstract class BaseRecordIndexer extends BaseIndexer { + + private static final int RECORD_INDEX_AVERAGE_RECORD_SIZE = 48; + + protected BaseRecordIndexer(HoodieEngineContext engineContext, HoodieWriteConfig dataTableWriteConfig, HoodieTableMetaClient dataTableMetaClient) { + super(engineContext, dataTableWriteConfig, dataTableMetaClient); + } + + protected DataPartitionAndRecords initializeRecordIndexPartition( + List<FileSliceAndPartition> latestMergedPartitionFileSliceList, + int recordIndexMaxParallelism) { + return initializeRecordIndexPartition(null, latestMergedPartitionFileSliceList, recordIndexMaxParallelism); + } + + protected DataPartitionAndRecords initializeRecordIndexPartition( + String dataPartition, + List<FileSliceAndPartition> latestMergedPartitionFileSliceList, + int recordIndexMaxParallelism) { + log.info("Initializing record index from {} file slices", latestMergedPartitionFileSliceList.size()); + HoodieData<HoodieRecord> records = readRecordKeysFromFileSliceSnapshot( + engineContext, + latestMergedPartitionFileSliceList, + recordIndexMaxParallelism, + this.getClass().getSimpleName(), + dataTableMetaClient, + dataTableWriteConfig); + + // Initialize the file groups + final int fileGroupCount = estimateFileGroupCount(records); + log.info("Initializing record index with {} file groups.", fileGroupCount); + return new DataPartitionAndRecords(fileGroupCount, Option.ofNullable(dataPartition), records); + } + + @Override + public void postInitialization(HoodieTableMetaClient metadataMetaClient, HoodieData<HoodieRecord> records, int fileGroupCount, String relativePartitionPath) { + super.postInitialization(metadataMetaClient, records, fileGroupCount, relativePartitionPath); + // Validate record index after commit if validation is enabled + if (dataTableWriteConfig.getMetadataConfig().isRecordIndexInitializationValidationEnabled()) { + validateRecordIndex(records, fileGroupCount, metadataMetaClient); + } + records.unpersist(); + } + + @Override + public List<IndexPartitionAndRecords> buildClean(String instantTime, HoodieCleanMetadata cleanMetadata) { + return Collections.emptyList(); + } + + @Override + public List<IndexPartitionAndRecords> buildUpdate( + String instantTime, + HoodieBackedTableMetadata tableMetadata, + Lazy<HoodieTableFileSystemView> lazyFileSystemView, + HoodieCommitMetadata commitMetadata) { + HoodieData<HoodieRecord> updatesFromWriteStatuses = convertMetadataToRecordIndexRecords(engineContext, commitMetadata, + dataTableWriteConfig.getMetadataConfig(), dataTableMetaClient, dataTableWriteConfig.getWritesFileIdEncoding(), instantTime); + HoodieData<HoodieRecord> additionalUpdates = getRecordIndexAdditionalUpserts(updatesFromWriteStatuses, commitMetadata, lazyFileSystemView); + return Collections.singletonList(IndexPartitionAndRecords.of(RECORD_INDEX.getPartitionPath(), updatesFromWriteStatuses.union(additionalUpdates))); + } + + /** + * Validates the record index after bootstrap by comparing the expected record count with the actual + * record count stored in the metadata table. The validation is performed in a distributed manner + * using the engine context to count records from HFiles in parallel. + * + * @param recordIndexRecords the HoodieData containing the expected records + * @param fileGroupCount the expected number of file groups + * @param metadataMetaClient meta client for the metadata table + */ + protected void validateRecordIndex(HoodieData<HoodieRecord> recordIndexRecords, int fileGroupCount, HoodieTableMetaClient metadataMetaClient) { + String partitionName = MetadataPartitionType.RECORD_INDEX.getPartitionPath(); + HoodieTableFileSystemView fsView = HoodieTableMetadataUtil.getFileSystemViewForMetadataTable(metadataMetaClient); + try { + // Use merged file slices to handle cases with pending compactions + List<FileSlice> fileSlices = HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, fsView, partitionName); + + // Filter to only file slices with base files and extract their storage paths + List<StoragePath> baseFilePaths = fileSlices.stream() + .filter(fs -> fs.getBaseFile().isPresent()) + .map(fs -> fs.getBaseFile().get().getStoragePath()) + .collect(Collectors.toList()); + + // Count records in a distributed manner using the engine context + long totalRecords = countRecordsInHFiles(baseFilePaths, metadataMetaClient); + long expectedRecordCount = recordIndexRecords.count(); + + ValidationUtils.checkArgument(totalRecords == expectedRecordCount, "Record Count Validation failed with " + + totalRecords + " present in record index vs the expected " + expectedRecordCount); + log.info(String.format("Record index initialized on %d shards (expected = %d) with %d records (expected = %d)", + fileSlices.size(), fileGroupCount, totalRecords, expectedRecordCount)); + } finally { + fsView.close(); + } + } + + /** + * Counts the total number of records in HFiles in a distributed manner. + * + * @param baseFilePaths list of storage paths to HFiles + * @param metadataMetaClient meta client for the metadata table + * @return total number of records across all HFiles + */ + private long countRecordsInHFiles(List<StoragePath> baseFilePaths, HoodieTableMetaClient metadataMetaClient) { + if (baseFilePaths.isEmpty()) { + return 0L; + } + + int parallelism = Math.min(baseFilePaths.size(), dataTableWriteConfig.getMetadataConfig().getRecordIndexMaxParallelism()); + StorageConfiguration<?> storageConfBroadcast = metadataMetaClient.getStorageConf(); + HoodieFileFormat baseFileFormat = metadataMetaClient.getTableConfig().getBaseFileFormat(); + + return engineContext.parallelize(baseFilePaths, parallelism) + .mapPartitions(pathIterator -> { + long count = 0L; + while (pathIterator.hasNext()) { + StoragePath path = pathIterator.next(); + try { + HoodieStorage storage = HoodieStorageUtils.getStorage(path, storageConfBroadcast); + HoodieConfig readerConfig = new HoodieConfig(); + HoodieAvroFileReader reader = (HoodieAvroFileReader) HoodieIOFactory.getIOFactory(storage) + .getReaderFactory(HoodieRecord.HoodieRecordType.AVRO) + .getFileReader(readerConfig, path, baseFileFormat, Option.empty()); + try { + count += reader.getTotalRecords(); + } finally { + reader.close(); + } + } catch (IOException e) { + throw new HoodieIOException("Error reading total records from file " + path, e); + } + } + return Collections.singletonList(count).iterator(); + }, true) + .collectAsList() + .stream() + .mapToLong(Long::longValue) + .sum(); + } + + /** + * Fetch record locations from FileSlice snapshot. + * + * @param engineContext context ot use. + * @param partitionFileSlicePairs list of pairs of partition and file slice. + * @param recordIndexMaxParallelism parallelism to use. + * @param activeModule active module of interest. + * @param metaClient metaclient instance to use. + * @param dataWriteConfig write config to use. + * @return metadata records for initializing record index entries. + */ + protected <T> HoodieData<HoodieRecord> readRecordKeysFromFileSliceSnapshot( + HoodieEngineContext engineContext, + List<FileSliceAndPartition> partitionFileSlicePairs, + int recordIndexMaxParallelism, + String activeModule, + HoodieTableMetaClient metaClient, + HoodieWriteConfig dataWriteConfig) { + if (partitionFileSlicePairs.isEmpty()) { + return engineContext.emptyHoodieData(); + } + + Option<String> instantTime = metaClient.getActiveTimeline().getCommitsTimeline() + .filterCompletedInstants() + .lastInstant() + .map(HoodieInstant::requestedTime); + if (!instantTime.isPresent()) { + return engineContext.emptyHoodieData(); + } + + engineContext.setJobStatus(activeModule, "Record Index: reading record keys from " + partitionFileSlicePairs.size() + " file slices"); + final int parallelism = Math.min(partitionFileSlicePairs.size(), recordIndexMaxParallelism); + ReaderContextFactory<T> readerContextFactory = engineContext.getReaderContextFactory(metaClient); + return engineContext.parallelize(partitionFileSlicePairs, parallelism).flatMap(partitionAndFileSlice -> { + final String partition = partitionAndFileSlice.partitionPath(); + final FileSlice fileSlice = partitionAndFileSlice.fileSlice(); + final String fileId = fileSlice.getFileId(); + HoodieReaderContext<T> readerContext = readerContextFactory.getContext(); + HoodieSchema dataSchema = HoodieSchemaCache.intern(HoodieSchemaUtils.addMetadataFields(HoodieSchema.parse(dataWriteConfig.getWriteSchema()), dataWriteConfig.allowOperationMetadataField())); + HoodieSchema requestedSchema = metaClient.getTableConfig().populateMetaFields() ? getRecordKeySchema() + : HoodieSchemaUtils.projectSchema(dataSchema, Arrays.asList(metaClient.getTableConfig().getRecordKeyFields().orElse(new String[0]))); + Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(dataWriteConfig.getInternalSchema()); + HoodieFileGroupReader<T> fileGroupReader = HoodieFileGroupReader.<T>newBuilder() + .withReaderContext(readerContext) + .withHoodieTableMetaClient(metaClient) + .withFileSlice(fileSlice) + .withLatestCommitTime(instantTime.get()) + .withDataSchema(dataSchema) + .withRequestedSchema(requestedSchema) + .withInternalSchema(internalSchemaOption) + .withShouldUseRecordPosition(false) + .withProps(metaClient.getTableConfig().getProps()) + .build(); + String baseFileInstantTime = fileSlice.getBaseInstantTime(); + return new CloseableMappingIterator<>(fileGroupReader.getClosableIterator(), record -> { + String recordKey = readerContext.getRecordContext().getRecordKey(record, requestedSchema); + return HoodieMetadataPayload.createRecordIndexUpdate(recordKey, partition, fileId, + baseFileInstantTime, 0); + }); + }); + } + + protected int estimateFileGroupCount(HoodieData<HoodieRecord> records) { + int minFileGroupCount; + int maxFileGroupCount; + if (dataTableWriteConfig.isRecordLevelIndexEnabled()) { + minFileGroupCount = dataTableWriteConfig.getRecordLevelIndexMinFileGroupCount(); + maxFileGroupCount = dataTableWriteConfig.getRecordLevelIndexMaxFileGroupCount(); + } else { + minFileGroupCount = dataTableWriteConfig.getGlobalRecordLevelIndexMinFileGroupCount(); + maxFileGroupCount = dataTableWriteConfig.getGlobalRecordLevelIndexMaxFileGroupCount(); + } + Supplier<Long> recordCountSupplier = () -> { + records.persist("MEMORY_AND_DISK_SER"); + long count = records.count(); + log.info("Initializing record index with {} mappings", count); + return count; + }; + return HoodieTableMetadataUtil.estimateFileGroupCount( + MetadataPartitionType.RECORD_INDEX, + recordCountSupplier, + RECORD_INDEX_AVERAGE_RECORD_SIZE, + minFileGroupCount, + maxFileGroupCount, + dataTableWriteConfig.getRecordIndexGrowthFactor(), + dataTableWriteConfig.getRecordIndexMaxFileGroupSizeBytes() + ); + } + + @VisibleForTesting + public static <T> HoodieData<HoodieRecord> convertMetadataToRecordIndexRecords(HoodieEngineContext engineContext, + HoodieCommitMetadata commitMetadata, + HoodieMetadataConfig metadataConfig, + HoodieTableMetaClient dataTableMetaClient, + int writesFileIdEncoding, + String instantTime) { + List<HoodieWriteStat> allWriteStats = commitMetadata.getPartitionToWriteStats().values().stream() + .flatMap(Collection::stream).collect(Collectors.toList()); + // Return early if there are no write stats, or if the operation is a compaction. + if (allWriteStats.isEmpty() || commitMetadata.getOperationType() == WriteOperationType.COMPACT) { + return engineContext.emptyHoodieData(); + } + // RLI cannot support logs having inserts with current offering. So, lets validate that. + if (allWriteStats.stream().anyMatch(writeStat -> { + String fileName = FSUtils.getFileName(writeStat.getPath(), writeStat.getPartitionPath()); + return FSUtils.isLogFile(fileName) && writeStat.getNumInserts() > 0; + })) { + throw new HoodieIOException("RLI cannot support logs having inserts with current offering. Would recommend disabling Record Level Index"); + } + + try { + Map<String, List<HoodieWriteStat>> writeStatsByFileId = allWriteStats.stream().collect(Collectors.groupingBy(HoodieWriteStat::getFileId)); + int parallelism = Math.max(Math.min(writeStatsByFileId.size(), metadataConfig.getRecordIndexMaxParallelism()), 1); + String basePath = dataTableMetaClient.getBasePath().toString(); + HoodieFileFormat baseFileFormat = dataTableMetaClient.getTableConfig().getBaseFileFormat(); Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/secondary/SecondaryIndexer.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.metadata.index.secondary; + +import org.apache.hudi.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.metadata.model.FileInfo; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.exception.HoodieIndexException; +import org.apache.hudi.exception.HoodieMetadataException; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords; +import org.apache.hudi.metadata.model.FileSliceAndPartition; +import org.apache.hudi.common.model.HoodieIndexDefinition; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.metadata.index.model.IndexPartitionInitialization; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.metadata.index.BaseIndexer; +import org.apache.hudi.util.Lazy; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_SECONDARY_INDEX_PREFIX; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getSecondaryIndexPartitionsToInit; +import static org.apache.hudi.metadata.MetadataPartitionType.RECORD_INDEX; +import static org.apache.hudi.metadata.MetadataPartitionType.SECONDARY_INDEX; +import static org.apache.hudi.metadata.SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords; +import static org.apache.hudi.metadata.SecondaryIndexRecordGenerationUtils.readSecondaryKeysFromFileSlices; + +/** + * Implementation of {@link MetadataPartitionType#SECONDARY_INDEX} index + */ +@Slf4j +public class SecondaryIndexer extends BaseIndexer { + + private static final int RECORD_INDEX_AVERAGE_RECORD_SIZE = 48; + Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/bloomfilters/TestBloomFiltersIndexer.java: ########## @@ -0,0 +1,173 @@ +/* + * 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 + */ + +package org.apache.hudi.metadata.index.bloomfilters; + +import org.apache.hudi.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieMetadataPayload; +import org.apache.hudi.metadata.HoodieTableMetadataUtil; +import org.apache.hudi.metadata.MetadataPartitionType; +import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords; +import org.apache.hudi.metadata.index.model.IndexPartitionInitialization; +import org.apache.hudi.metadata.model.FileInfo; +import org.apache.hudi.metadata.model.FileSliceAndPartition; +import org.apache.hudi.util.Lazy; + +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +class TestBloomFiltersIndexer { + + @SuppressWarnings("unchecked") + @Test + void testInitializeDataWithRealEngineContextAndIndexDataContent() throws IOException { + HoodieEngineContext engineContext = new HoodieLocalEngineContext(getDefaultStorageConf()); + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + + when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig); + when(writeConfig.getBloomIndexParallelism()).thenReturn(8); + when(writeConfig.getBloomFilterType()).thenReturn("DYNAMIC_V0"); + when(metadataConfig.getBloomFilterIndexFileGroupCount()).thenReturn(3); + + HoodieData<HoodieRecord> records = (HoodieData<HoodieRecord>) (HoodieData<?>) engineContext.parallelize( + Collections.singletonList(HoodieMetadataPayload.createPartitionFilesRecord("p_bloom", + Collections.singletonMap("f1.parquet", 11L), Collections.emptyList())), + 1); + + try (MockedStatic<HoodieTableMetadataUtil> mockedUtil = mockStatic(HoodieTableMetadataUtil.class)) { + mockedUtil.when(() -> HoodieTableMetadataUtil.convertFilesToBloomFilterRecords(any(), any(), any(), any(), any(), anyInt(), any())) + .thenReturn(records); + + ExposedBloomFiltersIndexer indexer = new ExposedBloomFiltersIndexer(engineContext, writeConfig, metaClient); + List<IndexPartitionInitialization> initializationList = indexer.callGetData("001", "002", Collections.emptyMap(), Lazy.lazily(Collections::emptyList)); + assertEquals(1, initializationList.size()); + + assertEquals(MetadataPartitionType.BLOOM_FILTERS.getPartitionPath(), initializationList.get(0).indexPartitionName()); + assertEquals(3, initializationList.get(0).totalFileGroups()); + List<HoodieRecord> collected = initializationList.get(0).dataPartitionAndRecords().get(0).indexRecords().collectAsList(); + assertEquals(1, collected.size()); + assertEquals("p_bloom", collected.get(0).getRecordKey()); + } + } + + @Test + @SuppressWarnings("unchecked") + void testBuildUpdateWithNonEmptyCommitMetadataAndWriteStat() throws Exception { + HoodieEngineContext engineContext = mock(HoodieEngineContext.class); + HoodieEngineContext realEngineContext = new HoodieLocalEngineContext(getDefaultStorageConf()); + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class); + when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig); + when(writeConfig.getBloomIndexParallelism()).thenReturn(4); + when(writeConfig.getBloomFilterType()).thenReturn("DYNAMIC_V0"); + + HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata(); + HoodieWriteStat writeStat = new HoodieWriteStat(); + writeStat.setPartitionPath("p1"); + writeStat.setPath("p1/f1.parquet"); + commitMetadata.addWriteStat("p1", writeStat); + + HoodieData<HoodieWriteStat> writeStatsData = (HoodieData<HoodieWriteStat>) mock(HoodieData.class); + String baseFileName = FSUtils.makeBaseFileName("004", "1-0-1", "fileid-1", ".parquet"); + HoodieData<HoodieRecord> bloomIndexData = (HoodieData<HoodieRecord>) (HoodieData<?>) realEngineContext.parallelize( + Collections.singletonList(HoodieMetadataPayload.createBloomFilterMetadataRecord( + "p1", baseFileName, "004", "DYNAMIC_V0", + ByteBuffer.wrap("bf".getBytes(StandardCharsets.UTF_8)), false)), + 1); + when(engineContext.parallelize(any(List.class), anyInt())).thenReturn((HoodieData) writeStatsData); + when(writeStatsData.flatMap(any())).thenReturn((HoodieData) bloomIndexData); + + ExposedBloomFiltersIndexer indexer = new ExposedBloomFiltersIndexer(engineContext, writeConfig, metaClient); + List<IndexPartitionAndRecords> result = indexer.buildUpdate( + "004", + mock(HoodieBackedTableMetadata.class), + Lazy.lazily(() -> mock(HoodieTableFileSystemView.class)), + commitMetadata); + + assertEquals(1, result.size()); + assertEquals(MetadataPartitionType.BLOOM_FILTERS.getPartitionPath(), result.get(0).indexPartitionName()); + List<HoodieRecord> indexRecords = result.get(0).indexRecords().collectAsList(); + assertEquals(1, indexRecords.size()); + HoodieMetadataPayload payload = (HoodieMetadataPayload) indexRecords.get(0).getData(); + assertTrue(payload.getBloomFilterMetadata().isPresent()); + assertEquals("DYNAMIC_V0", payload.getBloomFilterMetadata().get().getType()); + assertEquals("004", payload.getBloomFilterMetadata().get().getTimestamp()); + assertFalse(payload.getBloomFilterMetadata().get().getIsDeleted()); + assertEquals("bf", new String(payload.getBloomFilterMetadata().get().getBloomFilter().array(), StandardCharsets.UTF_8)); + } + + @Test + void testBuildCleanProducesDeleteRecordsForBaseFilesOnly() { + HoodieEngineContext engineContext = new HoodieLocalEngineContext(getDefaultStorageConf()); + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class); + when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig); + when(writeConfig.getBloomIndexParallelism()).thenReturn(4); + + HoodieCleanMetadata cleanMetadata = mock(HoodieCleanMetadata.class); + org.apache.hudi.avro.model.HoodieCleanPartitionMetadata partitionMetadata = mock(org.apache.hudi.avro.model.HoodieCleanPartitionMetadata.class); + when(cleanMetadata.getPartitionMetadata()).thenReturn(Collections.singletonMap("p1", partitionMetadata)); + when(partitionMetadata.getDeletePathPatterns()).thenReturn(Collections.singletonList("f1_1-0-1_000.parquet")); + + ExposedBloomFiltersIndexer indexer = new ExposedBloomFiltersIndexer(engineContext, writeConfig, metaClient); + List<IndexPartitionAndRecords> result = indexer.buildClean("005", cleanMetadata); + + assertEquals(1, result.size()); + assertEquals(MetadataPartitionType.BLOOM_FILTERS.getPartitionPath(), result.get(0).indexPartitionName()); + assertEquals(1, result.get(0).indexRecords().collectAsList().size()); + } + + private static class ExposedBloomFiltersIndexer extends BloomFiltersIndexer { + ExposedBloomFiltersIndexer(HoodieEngineContext engineContext, HoodieWriteConfig dataTableWriteConfig, HoodieTableMetaClient dataTableMetaClient) { Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexRecordGenerationUtils.java: ########## @@ -235,7 +236,7 @@ public static <T> Map<String, String> getRecordKeyToSecondaryKey(HoodieTableMeta } public static <T> HoodieData<HoodieRecord> readSecondaryKeysFromFileSlices(HoodieEngineContext engineContext, Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/ExpressionIndexRecordGenerator.java: ########## @@ -0,0 +1,84 @@ +/* + * 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.metadata.index; + +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.EngineType; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieIndexDefinition; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.metadata.model.FileInfoAndPartition; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.storage.StorageConfiguration; + +import java.util.List; + +/** + * Engine-specific generator for expression index records used during metadata index bootstrap. + */ +public interface ExpressionIndexRecordGenerator { + EngineType getEngineType(); + + /** + * Generates expression index records. + * + * @param filesToIndex triplet of partition, (file path, file size) Review Comment: Fixed. -- 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]
