924060929 commented on code in PR #69133:
URL: https://github.com/apache/doris/pull/69133#discussion_r4282018943
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -248,6 +247,169 @@ public boolean supportsBatchScan(ConnectorSession
session, ConnectorTableHandle
return partKeyNames != null && !partKeyNames.isEmpty() &&
!hiveHandle.isTransactional();
}
+ @Override
+ public long streamingSplitEstimate(ConnectorSession session,
ConnectorTableHandle handle,
+ Optional<ConnectorExpression> filter, boolean countPushdown) {
+ HiveTableHandle hiveHandle = (HiveTableHandle) handle;
+ if (hiveHandle.isTransactional()
+ ||
Boolean.FALSE.equals(session.getProperty("enable_external_table_batch_mode",
Boolean.class))) {
+ return -1;
+ }
+ int fileThreshold = sessionInt(session, "num_files_in_batch_mode",
DEFAULT_BATCH_THRESHOLD);
+ List<HmsPartitionInfo> partitions = hiveHandle.getPrunedPartitions();
+ long fileCount = 0;
+ int partitionCount = partitions == null ? 1 : partitions.size();
+ if (hiveHandle.getPartitionKeyNames() == null ||
hiveHandle.getPartitionKeyNames().isEmpty()) {
+ fileCount =
fileListingCache.cachedFileCount(hiveHandle.getDbName(),
hiveHandle.getTableName(),
+ hiveHandle.getLocation(), Collections.emptyList());
+ } else if (partitions == null) {
+ fileCount = -1;
+ } else {
+ for (HmsPartitionInfo partition : partitions) {
+ long count =
fileListingCache.cachedFileCount(hiveHandle.getDbName(),
hiveHandle.getTableName(),
+ partition.getLocation(), partition.getValues());
+ if (count < 0) {
+ fileCount = -1;
+ break;
+ }
+ fileCount += count;
+ }
+ }
+ // Unknown file counts must not trigger a full listing on the planning
thread. Use a
+ // nonzero concurrency hint and let the lazy source perform the
listing asynchronously.
+ if (fileCount < 0) {
+ return Math.max(DEFAULT_BATCH_THRESHOLD, fileThreshold);
+ }
+ int partitionThreshold = sessionInt(session,
"num_partitions_in_batch_mode", DEFAULT_BATCH_THRESHOLD);
+ boolean partitionBatch = partitionThreshold > 0 && partitionCount >=
partitionThreshold;
+ return fileCount >= fileThreshold || partitionBatch ? fileCount : -1;
Review Comment:
[P2] Preserve partition-listing concurrency when enabling streaming
A non-negative estimate diverts this scan through `computeBatchMode()` into
`startStreamingSplit()`, which creates only one producer/source for the entire
scan. `HiveSplitSource.hasNext()` then lists partition directories
synchronously, so the bounded parallelism in `createPartitionBatchSources()` is
bypassed for Hive.
For example, with batch mode enabled, 8192 selected non-transactional Hive
partitions, output slots, no TABLESAMPLE, and
`num_partitions_in_batch_mode=1024`, the base can enumerate eight partition
batches concurrently when workers are available. With cold file-list caches,
this path instead enumerates all directories through one source. That removes
existing remote-listing overlap; the single-partition measurements do not cover
this case. This is a source-confirmed concurrency regression, not a measured
claim about an end-to-end slowdown factor.
Please preserve bounded partition concurrency while streaming within each
batch. One scoped approach is to add an explicit partition-batch split-source
hook: default it to a deferred adapter over `planScanForPartitionBatch`, and
override it in Hive with a lazy source restricted to that batch. Existing
bounded producers can claim disjoint batches and share assignment/backpressure
and all-producers completion. Do not infer partition scoping from the two
existing capability flags alone: the default `streamSplits(session, request)`
overload drops `requiredPartitions`.
Please add coverage through the actual Hive mode-selection/dispatch path,
using latch-controlled directory listing to verify concurrent progress, the
concurrency bound, and complete unique split identities. The current
partition-source helper tests bypass the streaming branch that causes this
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]
