924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3821028715
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -251,6 +251,11 @@ protected void doInitialize() throws UserException {
// and `the file column name`.
// Split planning and FE-BE schema transport must describe the same
pinned Hudi instant.
ExternalUtil.initSchemaInfo(params, -1L,
table.getFullSchema(relationSnapshot));
+ fsViewLease = Env.getCurrentEnv()
Review Comment:
Fixed on a8597663fb4. The fs-view lease is no longer acquired in scan-node
init; acquisition is delayed until split planning. Batch mode registers a
statement-lifetime owner before submitting work, and statement cleanup stops
the assignment, waits for the producer plus every accepted worker to become
terminal, then releases the exact-generation lease. Focused lifecycle tests and
the full FE build pass.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/HudiExternalMetaCache.java:
##########
@@ -88,8 +88,9 @@ public HudiExternalMetaCache(ExecutorService refreshExecutor)
{
TablePartitionValues.class,
this::loadPartitionValuesCacheValue, defaultEntryCacheSpec(),
MetaCacheEntryInvalidation.forNameMapping(HudiPartitionCacheKey::getNameMapping)));
fsViewEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_FS_VIEW,
HudiFsViewCacheKey.class,
- HoodieTableFileSystemView.class, this::createFsView,
defaultEntryCacheSpec(),
-
MetaCacheEntryInvalidation.forNameMapping(HudiFsViewCacheKey::getNameMapping)));
+ HudiFsViewCacheValue.class, this::createFsView,
defaultEntryCacheSpec(),
+ false,
MetaCacheEntryInvalidation.forNameMapping(HudiFsViewCacheKey::getNameMapping),
Review Comment:
Fixed on a8597663fb4. Detached refresh remains disabled so an unborrowed
replacement generation cannot leak, but every foreground lease now calls
HoodieTableFileSystemView.sync() before handoff. Hudi 1.0.2 serializes sync
with its internal write lock, so active readers are protected.
HudiFsViewCacheValueTest verifies repeated hot-key acquisitions trigger sync
and the full FE build passes.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -494,22 +505,30 @@ private List<HudiSplit> planPartitionSplits(HivePartition
partition) throws IOEx
private void getPartitionsSplits(List<HivePartition> partitions,
List<Split> splits) {
Executor executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
- CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
+ Phaser tasks = new Phaser(1);
AtomicReference<Throwable> throwable = new AtomicReference<>();
long startTime = System.currentTimeMillis();
- partitions.forEach(partition -> executor.execute(() -> {
- try {
- getPartitionSplits(partition, splits);
- } catch (Throwable t) {
- throwable.set(t);
- } finally {
- countDownLatch.countDown();
- }
- }));
try {
- countDownLatch.await();
- } catch (InterruptedException e) {
- throw new RuntimeException(e.getMessage(), e);
+ for (HivePartition partition : partitions) {
+ tasks.register();
Review Comment:
Fixed on a8597663fb4. The non-batch join no longer uses Phaser; it records
each accepted CompletableFuture and waits on allOf().join(), so there is no
65,535-party ceiling and accepted work is terminal before lease release even
after submission rejection. Focused concurrency tests and the full FE build
pass.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
