github-actions[bot] commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4447546565
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2472,6 +2472,15 @@ public class Config extends ConfigBase {
@ConfField(mutable = false, masterOnly = false)
public static int max_external_cache_loader_thread_pool_size = 64;
+ @ConfField(description = "Workers for fair split production")
+ public static int external_scan_split_threads = 64;
+
Review Comment:
[P2] Validate these global scheduler values as positive at configuration
load. A non-positive `external_scan_split_threads` or
`external_scan_metadata_threads` is accepted by ConfigBase and then makes
`ExternalMetaCacheMgr` construction throw in `FairTaskExecutor`;
`external_scan_task_queue_capacity <= 0` reaches `newGroup()` and fails the
first external scan. Add startup range validation (and a regression for invalid
`fe.conf` values) so a typo is rejected at the configuration boundary.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -227,73 +230,150 @@ 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) {
Review Comment:
[P2] Keep the estimate per partition. `getNextBatch()` advances through
`fileCaches`/partitions whenever a range iterator is exhausted, so one returned
batch can contain ranges from many partitions; this line records that combined
batch size as `numSplitsPerPartition`. With one partition containing N ranges
and `splitBatchSize=1`, `numApproximateSplits()` reports only the initial
10-split floor instead of N when N is larger; with several small partitions it
can over-report. The estimate becomes `TSplitSource.numSplits` and drives BE
scanner-count/file-meta-cache heuristics. Track per-partition counts or
maintain a separate batch-size metric.
--
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]
