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. This can use the existing `streamSplits(session, request)` rather than
adding another SPI method: make it the canonical streaming entry point, and
explicitly extend the existing partition-batching capability contract so that
opted-in connectors must honor each request's partition subset in both eager
and streaming paths. Hive already reads that subset. Bounded producers can call
it with `request.withRequiredPartitions(batch)` for disjoint batches, sharing
assignment/backpressure and all-producers completion. Migrate the old streaming
overload and its implementations/tests so the default bridge no longer silently
drops request fields. Keep streaming-only connectors such as Iceberg on their
existing whole-scan path; do not slice them into Hive-style partition batches.
Connectors without streaming support can retain `planScanForPartitionBatch`.
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]
