924060929 commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4449031333
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -1737,74 +1740,79 @@ public void startSplit(int numBackends) throws
UserException {
public void doStartSplit() throws UserException {
TableScan scan = createTableScan();
- ExecutorService executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
Closeable generationLease =
IcebergUtils.retainTableGenerationForAsyncPlanning(
source.getTargetTable(), frozenGenerationSource);
- AsyncPlanningTask planningTask = new AsyncPlanningTask(executor,
splitAssignment, generationLease, () -> {
- AtomicReference<CloseableIterable<FileScanTask>> taskRef = new
AtomicReference<>();
- try {
- preExecutionAuthenticator.execute(
- () -> {
- long startTime = System.currentTimeMillis();
- try {
- CloseableIterable<FileScanTask> fileScanTasks
= planFileScanTask(scan);
- taskRef.set(fileScanTasks);
- CloseableIterator<FileScanTask> iterator =
fileScanTasks.iterator();
- while (splitAssignment.needMoreSplit() &&
iterator.hasNext()) {
- try {
- splitAssignment.addToQueue(
-
Lists.newArrayList(createIcebergSplit(iterator.next())));
- } catch (UserException e) {
- throw new RuntimeException(e);
- }
- }
- } finally {
- if (getSummaryProfile() != null) {
-
getSummaryProfile().addExternalTableGetFileScanTasksTime(
- System.currentTimeMillis() -
startTime);
- }
- }
- }
- );
- splitAssignment.finishSchedule();
- recordManifestCacheProfile();
- } catch (Exception e) {
- Optional<NotSupportedException> opt =
checkNotSupportedException(e);
- if (opt.isPresent()) {
- splitAssignment.setException(new
UserException(opt.get().getMessage(), opt.get()));
- } else {
- splitAssignment.setException(new
UserException(e.getMessage(), e));
Review Comment:
Fixed in 7e8a562c31a413d44a51decc8588e98ea1935fba. The executor/group lookup
now happens before retaining the independent generation lease. Cancellation
during that lookup therefore cannot strand an unregistered lease; after
acquisition, SplitProducer still owns cleanup on executor rejection,
cancellation and normal completion. A production-entry regression reproduces
the distinct statement-reference release followed by admission failure (one
leaked reference before the fix, zero after). All 218 related FE tests and full
FE packaging/Checkstyle pass locally.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -227,73 +230,161 @@ public List<Split> getSplits(int numBackends) throws
UserException {
return allFiles;
} catch (Throwable t) {
LOG.warn("get file split failed for table: {}",
hmsTable.getName(), t);
- throw new UserException(
- "get file split failed for table: " + hmsTable.getName() + ",
err: " + Util.getRootCauseMessage(t),
- t);
+ throw createSplitPlanningException(t);
}
}
+ private UserException createSplitPlanningException(Throwable cause) {
+ return new UserException("get file split failed for table: " +
hmsTable.getName()
+ + ", err: " + Util.getRootCauseMessage(cause), cause);
+ }
+
+ @Override
+ protected SplitAssignment createSplitAssignment(Map<String, String>
properties,
+ List<String> partitionKeys, boolean admission) {
+ return new SplitAssignment(backendPolicy, this,
this::splitToScanRange, properties,
+ partitionKeys, admission, sessionVariable.getSplitBatchSize(),
+ sessionVariable.getSplitQueueCapacity());
+ }
+
@Override
public void startSplit(int numBackends) {
+ ConnectContext context = ConnectContext.get();
+ StatementContext statement = context == null ? null :
context.getStatementContext();
+ if (statement != null) {
+ StmtExecutor executor = context.getExecutor();
+ statement.getOrRegisterStatementResource(this, () -> (Closeable)
() -> {
+ // A deferred Flight query still fetches splits after
statement handling ends.
+ // Its coordinator owns cleanup; abandoned planning and
EXPLAIN stop here.
+ if (executor == null || !executor.isDeferredForArrowFlight()) {
+ stop();
+ }
+ });
+ }
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
return;
}
HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()
.hive(hmsTable.getCatalog().getId());
- Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
- String bindBrokerName = hmsTable.getCatalog().bindBrokerName();
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
- long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- CompletableFuture.runAsync(() -> {
+ Executor executor = getSplitTaskExecutor();
+ String broker = hmsTable.getCatalog().bindBrokerName();
+ // Without the file cache, the eager path listed partitions
concurrently. Keep that overlap.
+ boolean parallelListing = !fileCountBatch ||
Config.max_external_file_cache_num <= 0;
+ int producers = parallelListing
+ ? Math.min(prunedPartitions.size(),
Config.max_external_cache_loader_thread_pool_size) : 1;
+ AtomicInteger nextPartition = new AtomicInteger();
+ AtomicInteger remaining = new AtomicInteger(producers);
+ long started = System.currentTimeMillis();
+ for (int i = 0; i < producers; i++) {
+ new SplitProducer(splitAssignment, executor,
+ createSplitBatchSource(cache, nextPartition, broker,
numBackends), failure -> {
try {
Review Comment:
Fixed in 7e8a562c31a413d44a51decc8588e98ea1935fba. Both Hive and Iceberg now
use one batch allocator that limits only the initial ArrayList capacity to 128,
retaining the configured batch maximum and growing with actual work. An
isolated 64MB JDK probe reproduced Integer.MAX_VALUE failing before any split.
Actual-source tests now cover that maximum with 257 Hive ranges (no artificial
128-range cap) and empty Iceberg planning with generation cleanup. All 218
related FE tests and full FE packaging/Checkstyle pass locally.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -227,73 +230,161 @@ public List<Split> getSplits(int numBackends) throws
UserException {
return allFiles;
} catch (Throwable t) {
LOG.warn("get file split failed for table: {}",
hmsTable.getName(), t);
- throw new UserException(
- "get file split failed for table: " + hmsTable.getName() + ",
err: " + Util.getRootCauseMessage(t),
- t);
+ throw createSplitPlanningException(t);
}
}
+ private UserException createSplitPlanningException(Throwable cause) {
+ return new UserException("get file split failed for table: " +
hmsTable.getName()
+ + ", err: " + Util.getRootCauseMessage(cause), cause);
+ }
+
+ @Override
+ protected SplitAssignment createSplitAssignment(Map<String, String>
properties,
+ List<String> partitionKeys, boolean admission) {
+ return new SplitAssignment(backendPolicy, this,
this::splitToScanRange, properties,
+ partitionKeys, admission, sessionVariable.getSplitBatchSize(),
+ sessionVariable.getSplitQueueCapacity());
+ }
+
@Override
public void startSplit(int numBackends) {
+ ConnectContext context = ConnectContext.get();
+ StatementContext statement = context == null ? null :
context.getStatementContext();
+ if (statement != null) {
+ StmtExecutor executor = context.getExecutor();
+ statement.getOrRegisterStatementResource(this, () -> (Closeable)
() -> {
+ // A deferred Flight query still fetches splits after
statement handling ends.
+ // Its coordinator owns cleanup; abandoned planning and
EXPLAIN stop here.
+ if (executor == null || !executor.isDeferredForArrowFlight()) {
+ stop();
+ }
+ });
+ }
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
return;
}
HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()
.hive(hmsTable.getCatalog().getId());
- Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
- String bindBrokerName = hmsTable.getCatalog().bindBrokerName();
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
- long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- CompletableFuture.runAsync(() -> {
+ Executor executor = getSplitTaskExecutor();
+ String broker = hmsTable.getCatalog().bindBrokerName();
+ // Without the file cache, the eager path listed partitions
concurrently. Keep that overlap.
+ boolean parallelListing = !fileCountBatch ||
Config.max_external_file_cache_num <= 0;
+ int producers = parallelListing
+ ? Math.min(prunedPartitions.size(),
Config.max_external_cache_loader_thread_pool_size) : 1;
+ AtomicInteger nextPartition = new AtomicInteger();
+ AtomicInteger remaining = new AtomicInteger(producers);
+ long started = System.currentTimeMillis();
+ for (int i = 0; i < producers; i++) {
+ new SplitProducer(splitAssignment, executor,
+ createSplitBatchSource(cache, nextPartition, broker,
numBackends), failure -> {
try {
- List<Split> allFiles = Lists.newArrayList();
- getFileSplitByPartitions(
- cache,
Collections.singletonList(partition), allFiles, bindBrokerName,
- numBackends, true);
- if (allFiles.size() > numSplitsPerPartition.get())
{
- numSplitsPerPartition.set(allFiles.size());
+ if (failure != null) {
+
splitAssignment.setException(createSplitPlanningException(failure));
}
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(allFiles);
- }
- } catch (Exception e) {
- batchException.set(new
UserException(e.getMessage(), e));
} finally {
- splittersOnFlight.release();
- if (batchException.get() != null) {
-
splitAssignment.setException(batchException.get());
- }
- if (numFinishedPartitions.incrementAndGet() ==
prunedPartitions.size()) {
+ if (remaining.decrementAndGet() == 0) {
if (getSummaryProfile() != null) {
getSummaryProfile().addExternalTableGetPartitionFilesTime(
- System.currentTimeMillis() -
startTime);
+ System.currentTimeMillis() -
started);
}
splitAssignment.finishSchedule();
}
}
- }, scheduleExecutor);
- } catch (Exception e) {
- // When submitting a task, an exception will be thrown if
the task pool(scheduleExecutor) is full
- batchException.set(new UserException(e.getMessage(), e));
- break;
+ }).start();
+ }
+ }
+
+ SplitProducer.BatchSource createSplitBatchSource(HiveExternalMetaCache
cache, AtomicInteger nextPartition,
+ String broker, int numBackends) {
+ return new SplitProducer.BatchSource() {
+ private Iterator<FileCacheValue> fileCaches =
Collections.emptyIterator();
+ private Iterator<HiveExternalMetaCache.HiveFileStatus> files =
Collections.emptyIterator();
+ private Iterator<Split> ranges = Collections.emptyIterator();
+ private FileCacheValue currentCache;
+ private long targetSize;
+ private boolean needSplit;
Review Comment:
Fixed in 7e8a562c31a413d44a51decc8588e98ea1935fba. COUNT(*) now makes one
scan-wide decision using bounded metadata lookahead on the existing concurrent,
cancellable producers. They retain nonempty file metadata until the
parallelism/backend threshold is reached, allowing only already in-flight
partitions to complete, then share that decision and reuse the metadata before
lazily listing further partitions. Singleton listing and original
interruption/fair scheduling are preserved; no split-executor task waits for
child tasks. The 100-file regression produced 800 ranges before repair and 100
after; a two-file scan still splits to 16 ranges. Tests cover both scanner
versions, Parquet/ORC, disabled file caching without relisting, bounded
lookahead, metadata failure and cancellation. All 218 related FE tests and full
FE packaging/Checkstyle pass locally.
--
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]
