cshuo commented on code in PR #18372:
URL: https://github.com/apache/hudi/pull/18372#discussion_r3541971960
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/index/record/TestPartitionedRecordIndexer.java:
##########
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.
+ */
+
+package org.apache.hudi.metadata.index.record;
+
+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.model.FileSlice;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Option;
+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.BaseFileRecordParsingUtils;
+import org.apache.hudi.metadata.index.model.DataPartitionAndRecords;
+import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords;
+import org.apache.hudi.metadata.index.model.IndexPartitionInitialization;
+import org.apache.hudi.metadata.model.FileSliceAndPartition;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.util.Lazy;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.util.Arrays;
+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.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+class TestPartitionedRecordIndexer {
+
+ @SuppressWarnings("unchecked")
+ @Test
+ void testInitializeWithRealEngineContextAndIndexDataContent() throws
IOException {
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+
+ when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metadataConfig.getRecordIndexMaxParallelism()).thenReturn(8);
+ when(dataMetaClient.getTableConfig()).thenReturn(tableConfig);
+
+ HoodieData<HoodieRecord> p1Data = (HoodieData<HoodieRecord>)
(HoodieData<?>) engineContext.parallelize(
+
Collections.singletonList(HoodieMetadataPayload.createPartitionFilesRecord("p1_data",
+ Collections.singletonMap("f1.parquet", 1L),
Collections.emptyList())),
+ 1);
+ HoodieData<HoodieRecord> p2Data = (HoodieData<HoodieRecord>)
(HoodieData<?>) engineContext.parallelize(
+
Collections.singletonList(HoodieMetadataPayload.createPartitionFilesRecord("p2_data",
+ Collections.singletonMap("f2.parquet", 2L),
Collections.emptyList())),
+ 1);
+
+ DataPartitionAndRecords p1Init = new DataPartitionAndRecords(1,
Option.of("p1"), p1Data);
+ DataPartitionAndRecords p2Init = new DataPartitionAndRecords(2,
Option.of("p2"), p2Data);
+
+ FileSliceAndPartition fs1 = FileSliceAndPartition.of("p1", new
FileSlice("p1", "001", "f1"));
+ FileSliceAndPartition fs2 = FileSliceAndPartition.of("p2", new
FileSlice("p2", "001", "f2"));
+ List<FileSliceAndPartition> input = Arrays.asList(fs1, fs2);
+
+ ExposedPartitionedRecordIndexer indexer = new
ExposedPartitionedRecordIndexer(engineContext, writeConfig, dataMetaClient,
p1Init, p2Init);
+
+ try (MockedStatic<HoodieTableMetadataUtil> mockedUtil =
mockStatic(HoodieTableMetadataUtil.class)) {
+ List<IndexPartitionInitialization> initializationList =
indexer.buildInitialization("001", "002", Map.of(), Lazy.lazily(() -> input));
+
+ mockedUtil.verify(() ->
HoodieTableMetadataUtil.createRecordIndexDefinition(any(), any()));
+ assertEquals(1, initializationList.size());
+ assertEquals(2,
initializationList.get(0).dataPartitionAndRecords().size());
+ assertEquals(2, indexer.initializePartitionCalls);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ void testBuildUpdateWithEmptyCommitMetadataProducesEmptyRecords() {
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+
+ when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metadataConfig.getRecordIndexMaxParallelism()).thenReturn(8);
+ when(dataMetaClient.getTableConfig()).thenReturn(tableConfig);
+
+ HoodieData<HoodieRecord> empty = (HoodieData<HoodieRecord>)
(HoodieData<?>) engineContext.emptyHoodieData();
+ ExposedPartitionedRecordIndexer indexer = new
ExposedPartitionedRecordIndexer(
+ engineContext, writeConfig, dataMetaClient,
+ new DataPartitionAndRecords(1, Option.of("p1"), empty),
+ new DataPartitionAndRecords(1, Option.of("p2"), empty));
+
+ List<IndexPartitionAndRecords> result = indexer.buildUpdate(
+ "018",
+ mock(HoodieBackedTableMetadata.class),
+ Lazy.lazily(() -> mock(HoodieTableFileSystemView.class)),
+ new HoodieCommitMetadata());
+
+ assertEquals(1, result.size());
+ assertEquals(MetadataPartitionType.RECORD_INDEX.getPartitionPath(),
result.get(0).indexPartitionName());
+ assertEquals(0, result.get(0).indexRecords().collectAsList().size());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testBuildUpdateWithNonEmptyCommitMetadataProducesPartitionEntry() {
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(getDefaultStorageConf());
+ HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+
+ when(writeConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metadataConfig.getRecordIndexMaxParallelism()).thenReturn(8);
+ when(metadataConfig.isRecordLevelIndexEnabled()).thenReturn(true);
+ when(dataMetaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.getBaseFileFormat()).thenReturn(HoodieFileFormat.PARQUET);
+ when(dataMetaClient.getBasePath()).thenReturn(new
StoragePath("/tmp/hudi-partitioned-record-index-test"));
+
when(dataMetaClient.getStorageConf()).thenReturn((org.apache.hudi.storage.StorageConfiguration)
getDefaultStorageConf());
Review Comment:
Fixed.
##########
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();
+ StorageConfiguration storageConfiguration =
dataTableMetaClient.getStorageConf();
+ Option<HoodieSchema> writerSchemaOpt =
HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient);
+ Option<HoodieSchema> finalWriterSchemaOpt = writerSchemaOpt;
+ ReaderContextFactory<T> readerContextFactory =
engineContext.getReaderContextFactory(dataTableMetaClient);
+ HoodieData<HoodieRecord> recordIndexRecords =
engineContext.parallelize(new ArrayList<>(writeStatsByFileId.entrySet()),
parallelism)
+ .flatMap(writeStatsByFileIdEntry -> {
+ String fileId = writeStatsByFileIdEntry.getKey();
+ List<HoodieWriteStat> writeStats =
writeStatsByFileIdEntry.getValue();
+ // Partition the write stats into base file and log file write
stats
+ List<HoodieWriteStat> baseFileWriteStats = writeStats.stream()
+ .filter(writeStat ->
writeStat.getPath().endsWith(baseFileFormat.getFileExtension()))
+ .collect(Collectors.toList());
+ List<HoodieWriteStat> logFileWriteStats = writeStats.stream()
+ .filter(writeStat -> FSUtils.isLogFile(new
StoragePath(writeStats.get(0).getPath())))
+ .collect(Collectors.toList());
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]
