cshuo commented on code in PR #13078:
URL: https://github.com/apache/hudi/pull/13078#discussion_r2052010867
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactOperator.java:
##########
@@ -138,17 +148,41 @@ private void doCompaction(String instantTime,
compactionMetrics.startCompaction();
HoodieFlinkMergeOnReadTableCompactor<?> compactor = new
HoodieFlinkMergeOnReadTableCompactor<>();
HoodieTableMetaClient metaClient =
writeClient.getHoodieTable().getMetaClient();
- String maxInstantTime = compactor.getMaxInstantTime(metaClient);
- List<WriteStatus> writeStatuses = compactor.compact(
- new HoodieFlinkCopyOnWriteTable<>(
- writeConfig,
- writeClient.getEngineContext(),
- metaClient),
- metaClient,
- writeClient.getConfig(),
- compactionOperation,
- instantTime, maxInstantTime,
- writeClient.getHoodieTable().getTaskContextSupplier());
+
+ boolean useFileGroupReaderBasedCompaction = !metaClient.isMetadataTable()
+ &&
writeConfig.getBooleanOrDefault(HoodieReaderConfig.FILE_GROUP_READER_ENABLED)
+ && compactionOperation.getBootstrapFilePath().isEmpty()
+ && writeConfig.populateMetaFields()
// Virtual key support by fg reader is not ready
+ && !(metaClient.getTableConfig().isCDCEnabled() &&
writeConfig.isYieldingPureLogForMor()); // do not support produce cdc log
during fg reader
+
+ List<WriteStatus> writeStatuses;
+ if (useFileGroupReaderBasedCompaction) {
Review Comment:
the initialization of `internalSchemaManager` is costly, and maybe should
not be initialized per compact operation? For example, in the source side,
`internalSchemaManager` is initialized during table source compilation, and
serialized to every source subtasks then.
--
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]
