danny0405 commented on code in PR #13078:
URL: https://github.com/apache/hudi/pull/13078#discussion_r2049873325
##########
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:
can we move the whole logic into the compactor, and maybe the
`internalSchemaManager` can be initialized only necessary.
--
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]
