github-actions[bot] commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4448650933
##########
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:
[P1] Release the retained Iceberg generation when admission races
cancellation. `doStartSplit` retains `generationLease` before evaluating
`getSplitTaskExecutor()` here; if statement cleanup closes `scanTaskLease` in
that window, `Lease.group(false)` throws before `SplitProducer.start()`
registers the source and before `BatchSource.close()` can release
`generationLease`. The statement's original lease is a separate reference, so
repeated cancelled scans retain retired table generations/FileIO. Acquire the
executor first or close `generationLease` in a failure path around producer
construction.
##########
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:
[P1] Bound the producer batch allocation independently of the user setting.
`split_batch_size` accepts any positive int, and this new
`ArrayList<>(splitAssignment.getSplitBatchSize())` allocates that many
references before backpressure or file availability is known. A query hint such
as `split_batch_size=2147483647` can trigger an FE OOME (the Iceberg batch
source has the same allocation). Cap the setting or start with a small bounded
capacity and grow as batches are filled.
##########
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:
[P2] Keep the COUNT(*) split decision scan-wide in streaming mode.
`createSplitBatchSource` now calls `needSplitFiles(loaded, numBackends)` after
loading each singleton partition, so a table with 100 one-file partitions and a
parallelism threshold of 32 makes every 1-GB Parquet/ORC file splittable even
though the aggregate file count would disable splitting in the eager path. Each
file is then split into many ranges and its metadata is reread, defeating the
count-pushdown optimization. Carry the aggregate selected-file count (or an
equivalent shared decision) while retaining lazy range materialization, and add
a multi-partition COUNT(*) regression.
--
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]
