cshuo commented on code in PR #18372: URL: https://github.com/apache/hudi/pull/18372#discussion_r3542058438
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/Indexer.java: ########## @@ -0,0 +1,126 @@ +/* + * 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.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.index.model.IndexPartitionInitialization; +import org.apache.hudi.metadata.model.FileInfo; +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.util.Lazy; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * Interface to building records for initializing and updating a type of metadata or index + * in the metadata table. + * <p> + * When a new type of index is added to MetadataPartitionType, an + * implementation of the {@link Indexer} interface is required, and it + * must be added to {@link IndexerFactory}. + */ +public interface Indexer { + /** + * Generates records for initializing the index. + * + * @param dataTableInstantTime instant time of the data table that the metadata table is initialized on + * @param instantTimeForPartition instant time used for initializing a specific metadata partition + * @param partitionToAllFilesMap map of partition to files + * @param lazyLatestMergedPartitionFileSliceList lazily-evaluated list of file slices for the indexer that needs it + * @return zero or more {@link IndexPartitionInitialization} entries to be initialized. + * Returning an empty list means no metadata partition needs initialization in this invocation. + * @throws IOException upon IO error + */ + List<IndexPartitionInitialization> buildInitialization( + String dataTableInstantTime, + String instantTimeForPartition, + Map<String, List<FileInfo>> partitionToAllFilesMap, + Lazy<List<FileSliceAndPartition>> lazyLatestMergedPartitionFileSliceList) throws IOException; Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/bloomfilters/BloomFiltersIndexer.java: ########## @@ -0,0 +1,213 @@ +/* + * 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.bloomfilters; + +import org.apache.hudi.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.bloom.BloomFilter; +import org.apache.hudi.common.config.HoodieConfig; +import org.apache.hudi.common.data.HoodieData; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieDeltaWriteStat; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.StringUtils; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.io.storage.HoodieFileReader; +import org.apache.hudi.io.storage.HoodieIOFactory; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieMetadataPayload; +import org.apache.hudi.metadata.index.model.IndexPartitionAndRecords; +import org.apache.hudi.metadata.model.FileInfo; +import org.apache.hudi.metadata.model.FileSliceAndPartition; +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.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.storage.StoragePath; +import org.apache.hudi.util.Lazy; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes; +import static org.apache.hudi.metadata.MetadataPartitionType.BLOOM_FILTERS; + +/** + * Implementation of {@link MetadataPartitionType#BLOOM_FILTERS} index + */ +@Slf4j +public class BloomFiltersIndexer extends BaseIndexer { + + public BloomFiltersIndexer(HoodieEngineContext engineContext, + HoodieWriteConfig dataTableWriteConfig, + HoodieTableMetaClient dataTableMetaClient) { + super(engineContext, dataTableWriteConfig, dataTableMetaClient); + } + + @Override + public List<IndexPartitionInitialization> buildInitialization( + String dataTableInstantTime, + String instantTimeForPartition, + Map<String, List<FileInfo>> partitionToAllFilesMap, + Lazy<List<FileSliceAndPartition>> lazyLatestMergedPartitionFileSliceList) throws IOException { + HoodieData<HoodieRecord> records = HoodieTableMetadataUtil.convertFilesToBloomFilterRecords( + engineContext, Collections.emptyMap(), partitionToAllFilesMap, dataTableInstantTime, dataTableMetaClient, + dataTableWriteConfig.getBloomIndexParallelism(), dataTableWriteConfig.getBloomFilterType()); + final int fileGroupCount = dataTableWriteConfig.getMetadataConfig().getBloomFilterIndexFileGroupCount(); + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, BLOOM_FILTERS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildUpdate( + String instantTime, + HoodieBackedTableMetadata tableMetadata, + Lazy<HoodieTableFileSystemView> lazyFileSystemView, + HoodieCommitMetadata commitMetadata) { + final HoodieData<HoodieRecord> records = convertMetadataToBloomFilterRecords( + engineContext, dataTableWriteConfig, commitMetadata, instantTime, dataTableMetaClient, + dataTableWriteConfig.getBloomFilterType(), dataTableWriteConfig.getBloomIndexParallelism()); + return Collections.singletonList(IndexPartitionAndRecords.of(BLOOM_FILTERS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildClean(String instantTime, HoodieCleanMetadata cleanMetadata) { + final HoodieData<HoodieRecord> records = + convertMetadataToBloomFilterRecords(cleanMetadata, engineContext, instantTime, dataTableWriteConfig.getBloomIndexParallelism()); + return Collections.singletonList(IndexPartitionAndRecords.of(BLOOM_FILTERS.getPartitionPath(), records)); + } + + /** + * Convert commit action metadata to bloom filter records. + * + * @param context - Engine context to use + * @param hoodieConfig - Hudi configs + * @param commitMetadata - Commit action metadata + * @param instantTime - Action instant time + * @param dataMetaClient - HoodieTableMetaClient for data + * @param bloomFilterType - Type of generated bloom filter records + * @param bloomIndexParallelism - Parallelism for bloom filter record generation + * @return HoodieData of metadata table records + */ + private static HoodieData<HoodieRecord> convertMetadataToBloomFilterRecords( + HoodieEngineContext context, + HoodieConfig hoodieConfig, + HoodieCommitMetadata commitMetadata, + String instantTime, + HoodieTableMetaClient dataMetaClient, + String bloomFilterType, + int bloomIndexParallelism) { + final List<HoodieWriteStat> allWriteStats = commitMetadata.getPartitionToWriteStats().values().stream() + .flatMap(Collection::stream).collect(Collectors.toList()); + if (allWriteStats.isEmpty()) { + return context.emptyHoodieData(); + } + + final int parallelism = Math.max(Math.min(allWriteStats.size(), bloomIndexParallelism), 1); + HoodieData<HoodieWriteStat> allWriteStatsRDD = context.parallelize(allWriteStats, parallelism); + return allWriteStatsRDD.flatMap(hoodieWriteStat -> { + final String partition = hoodieWriteStat.getPartitionPath(); + + // For bloom filter index, delta writes do not change the base file bloom filter entries + if (hoodieWriteStat instanceof HoodieDeltaWriteStat) { + return Collections.emptyListIterator(); + } + + String pathWithPartition = hoodieWriteStat.getPath(); + if (pathWithPartition == null) { + // Empty partition + log.error("Failed to find path in write stat to update metadata table {}", hoodieWriteStat); + return Collections.emptyListIterator(); + } + + String fileName = FSUtils.getFileName(pathWithPartition, partition); + if (!FSUtils.isBaseFile(new StoragePath(fileName))) { + return Collections.emptyListIterator(); + } + + final StoragePath writeFilePath = new StoragePath(dataMetaClient.getBasePath(), pathWithPartition); + try (HoodieFileReader fileReader = HoodieIOFactory.getIOFactory(dataMetaClient.getStorage()) + .getReaderFactory(HoodieRecord.HoodieRecordType.AVRO).getFileReader(hoodieConfig, writeFilePath)) { + try { + final BloomFilter fileBloomFilter = fileReader.readBloomFilter(); + if (fileBloomFilter == null) { + log.error("Failed to read bloom filter for {}", writeFilePath); + return Collections.emptyListIterator(); + } + ByteBuffer bloomByteBuffer = ByteBuffer.wrap(getUTF8Bytes(fileBloomFilter.serializeToString())); + HoodieRecord record = HoodieMetadataPayload.createBloomFilterMetadataRecord( + partition, fileName, instantTime, bloomFilterType, bloomByteBuffer, false); + return Collections.singletonList(record).iterator(); + } catch (Exception e) { + log.error("Failed to read bloom filter for {}", writeFilePath); + return Collections.emptyListIterator(); + } + } catch (IOException e) { + log.error("Failed to get bloom filter for file: {}, write stat: {}", writeFilePath, hoodieWriteStat); + } + return Collections.emptyListIterator(); + }); + } + + /** + * Convert clean metadata to bloom filter index records. + * + * @param cleanMetadata - Clean action metadata + * @param engineContext - Engine context + * @param instantTime - Clean action instant time + * @param bloomIndexParallelism - Parallelism for bloom filter record generation + * @return List of bloom filter index records for the clean metadata + */ + public static HoodieData<HoodieRecord> convertMetadataToBloomFilterRecords( + HoodieCleanMetadata cleanMetadata, + HoodieEngineContext engineContext, + String instantTime, + int bloomIndexParallelism) { + List<Pair<String, String>> deleteFileList = new ArrayList<>(); + cleanMetadata.getPartitionMetadata().forEach((partition, partitionMetadata) -> { + // Files deleted from a partition + List<String> deletedFiles = partitionMetadata.getDeletePathPatterns(); + deletedFiles.forEach(entry -> { Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/UnsupportedExpressionIndexRecordGenerator.java: ########## @@ -0,0 +1,82 @@ +/* + * 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.common.table.HoodieTableVersion; +import org.apache.hudi.exception.HoodieNotSupportedException; +import org.apache.hudi.metadata.HoodieTableMetadata; +import org.apache.hudi.metadata.model.FileInfoAndPartition; +import org.apache.hudi.storage.StorageConfiguration; + +import java.util.List; + +/** + * Fallback {@link ExpressionIndexRecordGenerator} that throws a not-supported exception + * when expression index bootstrap is requested for unsupported engines. + */ +public class UnsupportedExpressionIndexRecordGenerator implements ExpressionIndexRecordGenerator { + + private final EngineType engineType; + + public UnsupportedExpressionIndexRecordGenerator(EngineType engineType) { + this.engineType = engineType; + } + + @Override + public EngineType getEngineType() { + return engineType; + } + + @Override + public HoodieData<HoodieRecord> buildInitialization( + List<FileInfoAndPartition> filesToIndex, + HoodieIndexDefinition indexDefinition, + HoodieTableMetaClient metaClient, + int parallelism, + HoodieSchema tableSchema, + HoodieSchema readerSchema, + StorageConfiguration<?> storageConf, + String instantTime) { + if (metaClient.getTableConfig().getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) { + throw new HoodieNotSupportedException("Table version 6 and below does not support expression index"); + } + throw new HoodieNotSupportedException(engineType + " engine does not support building expression index yet"); + } + + @Override + public HoodieData<HoodieRecord> buildUpdate( Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/columnstats/ColumnStatsIndexer.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.columnstats; + +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.metadata.model.FileInfo; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieIndexVersion; +import org.apache.hudi.metadata.HoodieMetadataPayload; +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.Option; +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.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.util.Lazy; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.apache.hudi.common.fs.FSUtils.getFileNameFromPath; +import static org.apache.hudi.index.HoodieIndexUtils.register; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.existingIndexVersionOrDefault; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getColumnStatsRecords; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getColumnsToIndex; +import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS; + +/** + * Implementation of {@link MetadataPartitionType#COLUMN_STATS} metadata + */ +@Slf4j +public class ColumnStatsIndexer extends BaseIndexer { + private Lazy<List<String>> columnsToIndex; + + public ColumnStatsIndexer(HoodieEngineContext engineContext, + HoodieWriteConfig dataTableWriteConfig, + HoodieTableMetaClient dataTableMetaClient) { + super(engineContext, dataTableWriteConfig, dataTableMetaClient); + + this.columnsToIndex = Lazy.lazily(() -> + new ArrayList<>(getColumnsToIndex( + dataTableMetaClient.getTableConfig(), + dataTableWriteConfig.getMetadataConfig(), + Lazy.lazily(() -> HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient)), + true, + Option.of(dataTableWriteConfig.getRecordMerger().getRecordType()), + existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, dataTableMetaClient)).keySet())); + } + + @Override + public List<IndexPartitionInitialization> buildInitialization( + String dataTableInstantTime, + String instantTimeForPartition, + Map<String, List<FileInfo>> partitionToAllFilesMap, + Lazy<List<FileSliceAndPartition>> lazyLatestMergedPartitionFileSliceList) throws IOException { + final int fileGroupCount = dataTableWriteConfig.getMetadataConfig().getColumnStatsIndexFileGroupCount(); + if (partitionToAllFilesMap.isEmpty()) { + this.columnsToIndex = Lazy.lazily(Collections::emptyList); + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), engineContext.emptyHoodieData())); + } + + if (columnsToIndex.get().isEmpty()) { + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), engineContext.emptyHoodieData())); + } + + log.info("Indexing {} columns for column stats index", columnsToIndex.get().size()); + // during initialization, we need stats for base and log files. + HoodieData<HoodieRecord> records = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords( + engineContext, Collections.emptyMap(), partitionToAllFilesMap, + dataTableMetaClient, dataTableWriteConfig.getColumnStatsIndexParallelism(), + dataTableWriteConfig.getMetadataConfig().getMaxReaderBufferSize(), + columnsToIndex.get()); + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildUpdate( + String instantTime, + HoodieBackedTableMetadata tableMetadata, + Lazy<HoodieTableFileSystemView> lazyFileSystemView, + HoodieCommitMetadata commitMetadata) { + final HoodieData<HoodieRecord> records = convertMetadataToColumnStatsRecords(commitMetadata, engineContext, + dataTableMetaClient, dataTableWriteConfig.getMetadataConfig(), Option.of(dataTableWriteConfig.getRecordMerger().getRecordType())); + return Collections.singletonList(IndexPartitionAndRecords.of(COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildClean(String instantTime, HoodieCleanMetadata cleanMetadata) { + final HoodieData<HoodieRecord> records = + convertMetadataToColumnStatsRecords(cleanMetadata, engineContext, dataTableMetaClient, + dataTableWriteConfig.getMetadataConfig(), Option.of(dataTableWriteConfig.getRecordMerger().getRecordType())); + return Collections.singletonList(IndexPartitionAndRecords.of(COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildRestore(String instantTime, List<String> deletedPartitions, Map<String, List<FileInfo>> filesAdded, Map<String, List<String>> filesDeleted) { + if (filesDeleted.isEmpty() && filesAdded.isEmpty()) { + return Collections.emptyList(); + } + Lazy<Option<HoodieSchema>> tableSchema = + Lazy.lazily(() -> HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient)); + final List<String> columnsToIndex = new ArrayList<>(HoodieTableMetadataUtil.getColumnsToIndex( Review Comment: Fixed. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/index/columnstats/ColumnStatsIndexer.java: ########## @@ -0,0 +1,267 @@ +/* + * 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.columnstats; + +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.metadata.model.FileInfo; +import org.apache.hudi.common.model.HoodieCommitMetadata; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.metadata.HoodieBackedTableMetadata; +import org.apache.hudi.metadata.HoodieIndexVersion; +import org.apache.hudi.metadata.HoodieMetadataPayload; +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.Option; +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.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.util.Lazy; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.apache.hudi.common.fs.FSUtils.getFileNameFromPath; +import static org.apache.hudi.index.HoodieIndexUtils.register; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.existingIndexVersionOrDefault; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getColumnStatsRecords; +import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getColumnsToIndex; +import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS; + +/** + * Implementation of {@link MetadataPartitionType#COLUMN_STATS} metadata + */ +@Slf4j +public class ColumnStatsIndexer extends BaseIndexer { + private Lazy<List<String>> columnsToIndex; + + public ColumnStatsIndexer(HoodieEngineContext engineContext, + HoodieWriteConfig dataTableWriteConfig, + HoodieTableMetaClient dataTableMetaClient) { + super(engineContext, dataTableWriteConfig, dataTableMetaClient); + + this.columnsToIndex = Lazy.lazily(() -> + new ArrayList<>(getColumnsToIndex( + dataTableMetaClient.getTableConfig(), + dataTableWriteConfig.getMetadataConfig(), + Lazy.lazily(() -> HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient)), + true, + Option.of(dataTableWriteConfig.getRecordMerger().getRecordType()), + existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, dataTableMetaClient)).keySet())); + } + + @Override + public List<IndexPartitionInitialization> buildInitialization( + String dataTableInstantTime, + String instantTimeForPartition, + Map<String, List<FileInfo>> partitionToAllFilesMap, + Lazy<List<FileSliceAndPartition>> lazyLatestMergedPartitionFileSliceList) throws IOException { + final int fileGroupCount = dataTableWriteConfig.getMetadataConfig().getColumnStatsIndexFileGroupCount(); + if (partitionToAllFilesMap.isEmpty()) { + this.columnsToIndex = Lazy.lazily(Collections::emptyList); + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), engineContext.emptyHoodieData())); + } + + if (columnsToIndex.get().isEmpty()) { + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), engineContext.emptyHoodieData())); + } + + log.info("Indexing {} columns for column stats index", columnsToIndex.get().size()); + // during initialization, we need stats for base and log files. + HoodieData<HoodieRecord> records = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords( + engineContext, Collections.emptyMap(), partitionToAllFilesMap, + dataTableMetaClient, dataTableWriteConfig.getColumnStatsIndexParallelism(), + dataTableWriteConfig.getMetadataConfig().getMaxReaderBufferSize(), + columnsToIndex.get()); + return Collections.singletonList(IndexPartitionInitialization.of(fileGroupCount, COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildUpdate( + String instantTime, + HoodieBackedTableMetadata tableMetadata, + Lazy<HoodieTableFileSystemView> lazyFileSystemView, + HoodieCommitMetadata commitMetadata) { + final HoodieData<HoodieRecord> records = convertMetadataToColumnStatsRecords(commitMetadata, engineContext, + dataTableMetaClient, dataTableWriteConfig.getMetadataConfig(), Option.of(dataTableWriteConfig.getRecordMerger().getRecordType())); + return Collections.singletonList(IndexPartitionAndRecords.of(COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildClean(String instantTime, HoodieCleanMetadata cleanMetadata) { + final HoodieData<HoodieRecord> records = + convertMetadataToColumnStatsRecords(cleanMetadata, engineContext, dataTableMetaClient, + dataTableWriteConfig.getMetadataConfig(), Option.of(dataTableWriteConfig.getRecordMerger().getRecordType())); + return Collections.singletonList(IndexPartitionAndRecords.of(COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public List<IndexPartitionAndRecords> buildRestore(String instantTime, List<String> deletedPartitions, Map<String, List<FileInfo>> filesAdded, Map<String, List<String>> filesDeleted) { + if (filesDeleted.isEmpty() && filesAdded.isEmpty()) { + return Collections.emptyList(); + } + Lazy<Option<HoodieSchema>> tableSchema = + Lazy.lazily(() -> HoodieTableMetadataUtil.tryResolveSchemaForTable(dataTableMetaClient)); + final List<String> columnsToIndex = new ArrayList<>(HoodieTableMetadataUtil.getColumnsToIndex( + dataTableMetaClient.getTableConfig(), + dataTableWriteConfig.getMetadataConfig(), + tableSchema, + false, + Option.of(dataTableWriteConfig.getRecordMerger().getRecordType()), + HoodieTableMetadataUtil.existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, dataTableMetaClient)).keySet()); + if (columnsToIndex.isEmpty()) { + log.info("Since there are no columns to index, stop to generate ColumnStats records."); + return Collections.emptyList(); + } + HoodieData<HoodieRecord> records = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords( + engineContext, filesDeleted, + filesAdded, dataTableMetaClient, + dataTableWriteConfig.getMetadataConfig().getColumnStatsIndexParallelism(), + dataTableWriteConfig.getMetadataConfig().getMaxReaderBufferSize(), + columnsToIndex); + return Collections.singletonList(IndexPartitionAndRecords.of(COLUMN_STATS.getPartitionPath(), records)); + } + + @Override + public void postInitialization(HoodieTableMetaClient metadataMetaClient, HoodieData<HoodieRecord> records, int fileGroupCount, String relativePartitionPath) { + HoodieIndexDefinition indexDefinition = HoodieIndexDefinition.newBuilder() + .withIndexName(PARTITION_NAME_COLUMN_STATS) + .withIndexType(PARTITION_NAME_COLUMN_STATS) + .withIndexFunction(PARTITION_NAME_COLUMN_STATS) + .withSourceFields(columnsToIndex.get()) + // Use the existing version if exists, otherwise fall back to the default version. + .withVersion(existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS, dataTableMetaClient)) + .withIndexOptions(Collections.EMPTY_MAP) 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]
