danny0405 commented on code in PR #13535:
URL: https://github.com/apache/hudi/pull/13535#discussion_r2194937434
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactOperator.java:
##########
@@ -168,28 +187,28 @@ private void doCompaction(String instantTime,
compactionOperation,
instantTime,
flinkTable.getTaskContextSupplier(),
- readerContextOpt,
+ createReaderContext(writeClient, needReloadMetaClient),
flinkTable);
compactionMetrics.endCompaction();
collector.collect(new CompactionCommitEvent(instantTime,
compactionOperation.getFileId(), writeStatuses, taskID));
}
- private Option<HoodieReaderContext<?>>
initReaderContext(HoodieFlinkWriteClient<?> writeClient) {
- HoodieTableMetaClient metaClient = flinkTable.getMetaClient();
- boolean useFileGroupReaderBasedCompaction = !metaClient.isMetadataTable()
- &&
writeClient.getConfig().getBooleanOrDefault(HoodieReaderConfig.FILE_GROUP_READER_ENABLED)
- && writeClient.getConfig().populateMetaFields()
// Virtual key support by fg reader is not
ready
- && !(metaClient.getTableConfig().isCDCEnabled() &&
writeClient.getConfig().isYieldingPureLogForMor()); // do not support produce
cdc log during fg reader
+ private Option<HoodieReaderContext<?>>
createReaderContext(HoodieFlinkWriteClient<?> writeClient, boolean
needReloadMetaClient) {
if (useFileGroupReaderBasedCompaction) {
- // CAUTION: reuse the meta client so that the timeline is updated
- Supplier<InternalSchemaManager> internalSchemaManager = () ->
InternalSchemaManager.get(conf, metaClient);
+ HoodieTableMetaClient metaClient = flinkTable.getMetaClient();
// initialize storage conf lazily.
StorageConfiguration<?> readerConf =
writeClient.getEngineContext().getStorageConf();
- return Option.of(new FlinkRowDataReaderContext(readerConf,
internalSchemaManager, Collections.emptyList(), metaClient.getTableConfig(),
Option.empty()));
- } else {
- // always using avro record merger for legacy compaction since log
scanner do not support rowdata reading yet.
-
writeClient.getConfig().setRecordMergerClass(HoodieAvroRecordMerger.class.getName());
+ return Option.of(new FlinkRowDataReaderContext(
+ readerConf, () -> {
+ // CAUTION: InternalSchemaManager will scan timeline, reusing the
meta client so that the timeline is updated.
+ if (internalSchemaManager == null || needReloadMetaClient) {
+ internalSchemaManager = InternalSchemaManager.get(conf,
metaClient);
+ }
Review Comment:
Supplier usually does not have side-effect.
--
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]
