yihua commented on code in PR #12983:
URL: https://github.com/apache/hudi/pull/12983#discussion_r2076562209
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -381,137 +380,118 @@ private boolean initializeFromFilesystem(String
initializationTime, List<Metadat
}
}
- // Already initialized partitions can be ignored
- partitionsToInit.removeIf(metadataPartition ->
dataMetaClient.getTableConfig().isMetadataPartitionAvailable((metadataPartition)));
+ indexerMapForPartitionsToInit.keySet().removeIf(
+ metadataPartition ->
dataMetaClient.getTableConfig().isMetadataPartitionAvailable(metadataPartition));
// Get a complete list of files and partitions from the file system or
from already initialized FILES partition of MDT
List<DirectoryInfo> partitionInfoList;
if (filesPartitionAvailable) {
- partitionInfoList = listAllPartitionsFromMDT(initializationTime,
pendingDataInstants);
+ partitionInfoList = listAllPartitionsFromMDT(dataTableInstantTime,
pendingDataInstants);
} else {
// if auto initialization is enabled, then we need to list all
partitions from the file system
if (dataWriteConfig.getMetadataConfig().shouldAutoInitialize()) {
- partitionInfoList =
listAllPartitionsFromFilesystem(initializationTime, pendingDataInstants);
+ partitionInfoList =
listAllPartitionsFromFilesystem(dataTableInstantTime, pendingDataInstants);
} else {
// if auto initialization is disabled, we can return an empty list
partitionInfoList = Collections.emptyList();
}
}
- Map<String, Map<String, Long>> partitionToFilesMap =
partitionInfoList.stream()
+ Map<String, Map<String, Long>> partitionToAllFilesMap =
partitionInfoList.stream()
.map(p -> {
String partitionName =
HoodieTableMetadataUtil.getPartitionIdentifierForFilesPartition(p.getRelativePath());
return Pair.of(partitionName, p.getFileNameToSizeMap());
})
.collect(Collectors.toMap(Pair::getKey, Pair::getValue));
- // validate that each index is eligible to be initialized
- Iterator<MetadataPartitionType> iterator = partitionsToInit.iterator();
- while (iterator.hasNext()) {
- MetadataPartitionType partitionType = iterator.next();
- if (partitionType == PARTITION_STATS &&
!dataMetaClient.getTableConfig().isTablePartitioned()) {
- LOG.warn("Partition stats index cannot be enabled for a
non-partitioned table. Removing from initialization list. Please disable {}",
- HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key());
- iterator.remove();
- this.enabledPartitionTypes.remove(partitionType);
- }
+ // TODO(yihua): check this wrt files partition, i.e., after files
partition is ready in
+ // MDT, could it be leveraged afterward?
+ Lazy<List<Pair<String, FileSlice>>> lazyLatestMergedPartitionFileSliceList
= getLazyPartitionFileSliceList();
+ if (!filesPartitionAvailable) {
+ // FILES partition should always be initialized first if enabled
+ initializeMetadataPartition(
+ FILES, indexerMapForPartitionsToInit.get(FILES),
dataTableInstantTime,
+ partitionToAllFilesMap, lazyLatestMergedPartitionFileSliceList);
}
+ for (Map.Entry<MetadataPartitionType, Indexer> entry :
+ indexerMapForPartitionsToInit.entrySet().stream()
+ .filter(e -> e.getKey() != FILES).collect(Collectors.toList())) {
+ initializeMetadataPartition(
+ entry.getKey(), entry.getValue(), dataTableInstantTime,
+ partitionToAllFilesMap, lazyLatestMergedPartitionFileSliceList);
+ }
+ return true;
+ }
- for (MetadataPartitionType partitionType : partitionsToInit) {
- // Find the commit timestamp to use for this partition. Each
initialization should use its own unique commit time.
- String instantTimeForPartition =
generateUniqueInstantTime(initializationTime);
- String partitionTypeName = partitionType.name();
- LOG.info("Initializing MDT partition {} at instant {}",
partitionTypeName, instantTimeForPartition);
- String partitionName;
- Pair<Integer, HoodieData<HoodieRecord>> fileGroupCountAndRecordsPair;
- List<String> columnsToIndex = new ArrayList<>();
- try {
- switch (partitionType) {
- case FILES:
- fileGroupCountAndRecordsPair =
initializeFilesPartition(partitionInfoList);
- partitionName = FILES.getPartitionPath();
- break;
- case BLOOM_FILTERS:
- fileGroupCountAndRecordsPair =
initializeBloomFiltersPartition(initializationTime, partitionToFilesMap);
- partitionName = BLOOM_FILTERS.getPartitionPath();
- break;
- case COLUMN_STATS:
- Pair<List<String>, Pair<Integer, HoodieData<HoodieRecord>>>
colStatsColumnsAndRecord = initializeColumnStatsPartition(partitionToFilesMap);
- columnsToIndex = colStatsColumnsAndRecord.getKey();
- fileGroupCountAndRecordsPair = colStatsColumnsAndRecord.getValue();
- partitionName = COLUMN_STATS.getPartitionPath();
- break;
- case RECORD_INDEX:
- fileGroupCountAndRecordsPair = initializeRecordIndexPartition();
- partitionName = RECORD_INDEX.getPartitionPath();
- break;
- case EXPRESSION_INDEX:
- Set<String> expressionIndexPartitionsToInit =
getExpressionIndexPartitionsToInit(partitionType,
dataWriteConfig.getMetadataConfig(), dataMetaClient);
- if (expressionIndexPartitionsToInit.size() != 1) {
- if (expressionIndexPartitionsToInit.size() > 1) {
- LOG.warn("Skipping expression index initialization as only one
expression index bootstrap at a time is supported for now. Provided: {}",
expressionIndexPartitionsToInit);
- }
- continue;
- }
- partitionName = expressionIndexPartitionsToInit.iterator().next();
- fileGroupCountAndRecordsPair =
initializeExpressionIndexPartition(partitionName, instantTimeForPartition);
- break;
- case PARTITION_STATS:
- // For PARTITION_STATS, COLUMN_STATS should also be enabled
- if (!dataWriteConfig.isMetadataColumnStatsIndexEnabled()) {
- LOG.warn("Skipping partition stats initialization as column
stats index is not enabled. Please enable {}",
-
HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key());
- continue;
- }
- fileGroupCountAndRecordsPair = initializePartitionStatsIndex();
- partitionName = PARTITION_STATS.getPartitionPath();
- break;
- case SECONDARY_INDEX:
- Set<String> secondaryIndexPartitionsToInit =
getSecondaryIndexPartitionsToInit(partitionType,
dataWriteConfig.getMetadataConfig(), dataMetaClient);
- if (secondaryIndexPartitionsToInit.size() != 1) {
- if (secondaryIndexPartitionsToInit.size() > 1) {
- LOG.warn("Skipping secondary index initialization as only one
secondary index bootstrap at a time is supported for now. Provided: {}",
secondaryIndexPartitionsToInit);
- }
- continue;
- }
- partitionName = secondaryIndexPartitionsToInit.iterator().next();
- fileGroupCountAndRecordsPair =
initializeSecondaryIndexPartition(partitionName);
- break;
- default:
- throw new HoodieMetadataException(String.format("Unsupported MDT
partition type: %s", partitionType));
- }
- } catch (Exception e) {
- String metricKey = partitionType.getPartitionPath() + "_" +
HoodieMetadataMetrics.BOOTSTRAP_ERR_STR;
- metrics.ifPresent(m -> m.setMetric(metricKey, 1));
- String errMsg = String.format("Bootstrap on %s partition failed for
%s",
- partitionType.getPartitionPath(),
metadataMetaClient.getBasePath());
- LOG.error(errMsg, e);
- throw new HoodieMetadataException(errMsg, e);
+ private Lazy<List<Pair<String, FileSlice>>> getLazyPartitionFileSliceList() {
+ return Lazy.lazily(() -> {
+ String latestInstant =
dataMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant()
+ .map(HoodieInstant::requestedTime).orElse(SOLO_COMMIT_TIMESTAMP);
+ // now we need to rely on outside caller to pass fsView and close the
fsView.
+ // also see if we can reuse fsView across indexes.
+ // Collect the list of latest file slices present in each partition
+ try (HoodieTableFileSystemView fsView = getMetadataView()) {
+ List<String> partitions = metadata.getAllPartitionPaths();
+ fsView.loadAllPartitions();
+ List<Pair<String, FileSlice>> partitionFileSlicePairs = new
ArrayList<>();
+ partitions.forEach(partition ->
fsView.getLatestMergedFileSlicesBeforeOrOn(partition, latestInstant)
+ .forEach(fs -> partitionFileSlicePairs.add(Pair.of(partition,
fs))));
+ return partitionFileSlicePairs;
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
}
+ });
+ }
- if (LOG.isInfoEnabled()) {
- LOG.info("Initializing {} index with {} mappings", partitionTypeName,
fileGroupCountAndRecordsPair.getKey());
- }
- HoodieTimer partitionInitTimer = HoodieTimer.start();
-
- // Generate the file groups
- final int fileGroupCount = fileGroupCountAndRecordsPair.getKey();
- ValidationUtils.checkArgument(fileGroupCount > 0, "FileGroup count for
MDT partition " + partitionTypeName + " should be > 0");
- initializeFileGroups(dataMetaClient, partitionType,
instantTimeForPartition, fileGroupCount, partitionName);
-
- // Perform the commit using bulkCommit
- HoodieData<HoodieRecord> records =
fileGroupCountAndRecordsPair.getValue();
- bulkCommit(instantTimeForPartition, partitionName, records,
fileGroupCount);
- if (partitionType == COLUMN_STATS) {
- // initialize Col Stats index definition
- updateColumnsToIndexWithColStats(columnsToIndex);
- }
-
dataMetaClient.getTableConfig().setMetadataPartitionState(dataMetaClient,
partitionName, true);
- // initialize the metadata reader again so the MDT partition can be read
after initialization
- initMetadataReader();
- long totalInitTime = partitionInitTimer.endTimer();
- LOG.info("Initializing {} index in metadata table took {} in ms",
partitionTypeName, totalInitTime);
+ private void initializeMetadataPartition(
+ MetadataPartitionType partitionType,
+ Indexer indexer,
+ String dataTableInstantTime,
+ Map<String, Map<String, Long>> partitionToAllFilesMap,
+ Lazy<List<Pair<String, FileSlice>>>
lazyLatestMergedPartitionFileSliceList)
+ throws IOException {
+ // Find the commit timestamp to use for this partition. Each
initialization should use its own unique commit time.
+ String instantTimeForPartition =
generateUniqueInstantTime(dataTableInstantTime);
+ String partitionTypeName = partitionType.name();
+ LOG.info("Initializing MDT partition {} at instant {}", partitionTypeName,
instantTimeForPartition);
+ List<Indexer.InitialIndexPartitionData> initialIndexPartitionDataList;
+ try {
+ initialIndexPartitionDataList = indexer.initialize(
+ dataTableInstantTime, partitionToAllFilesMap,
lazyLatestMergedPartitionFileSliceList);
+ } catch (Exception e) {
+ String metricKey = partitionType.getPartitionPath() + "_" +
HoodieMetadataMetrics.BOOTSTRAP_ERR_STR;
+ metrics.ifPresent(m -> m.setMetric(metricKey, 1));
+ String errMsg = String.format("Bootstrap on %s partition failed for %s",
+ partitionType.getPartitionPath(), metadataMetaClient.getBasePath());
+ LOG.error(errMsg, e);
+ throw new HoodieMetadataException(errMsg, e);
}
- return true;
+
+ if (initialIndexPartitionDataList.isEmpty()) {
+ LOG.info("Skip building {} index in metadata table", partitionTypeName);
+ return;
+ }
+
+ ValidationUtils.checkArgument(initialIndexPartitionDataList.size() == 1,
Review Comment:
Each indexer implementation need to enforce it. The metadata writer forces
the check so that if the indexer does not follow this convention the
initialization fails.
--
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]
