924060929 commented on code in PR #69133:
URL: https://github.com/apache/doris/pull/69133#discussion_r4282142009
##########
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:
A related implementation simplification: keep partition-batch eligibility in
the engine's existing `shouldUseBatchMode()` instead of also calculating it in
Hive's `streamingSplitEstimate()`. The two currently use different
representations of the partition selection (`selectedPartitions` versus the
handle's pruned partitions). When composing partition batching with streaming,
let the engine select disjoint request scopes and let Hive enumerate each
requested scope lazily. Remove the duplicated Hive partition-threshold decision
as part of that routing change, rather than deleting the condition in isolation
and accidentally restoring eager batch materialization. This is part of the fix
discussed above, not a separate regression finding.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -2239,36 +2266,80 @@ private void startStreamingSplit(QueryFinishCleanup
queryFinishCleanup) {
final ConnectorTableHandle handle = currentHandle;
final ConnectorScanPlanProvider scanProvider = resolveScanProvider();
queryFinishCleanup.bind(scanProvider);
+ long[] partitionCounts = displayPartitionCounts(selectedPartitions);
+ if (partitionCounts != null) {
+ selectedPartitionNum = partitionCounts[0];
+ totalPartitionNum = partitionCounts[1];
+ }
+ List<String> requiredPartitions =
resolveRequiredPartitions(selectedPartitions,
+ onPluginClassLoader(scanProvider,
scanProvider::ignorePartitionPruneShortCircuit));
+ if (requiredPartitions != null && requiredPartitions.isEmpty()) {
+ splitAssignment.finishSchedule();
+ return;
+ }
+ final ConnectorScanRequest request =
ConnectorScanRequest.builder(handle, columns)
+ .filter(remainingFilter)
+ .requiredPartitions(requiredPartitions)
+ .build();
+ SummaryProfile summaryProfile =
SummaryProfile.getSummaryProfile(ConnectContext.get());
+ RuntimeProfile generationProfile = new RuntimeProfile("Split
Generation " + getId());
+ generationProfile.addInfoString("BatchSize",
String.valueOf(sessionVariable.getSplitBatchSize()));
+ generationProfile.addInfoString("QueueCapacity",
String.valueOf(sessionVariable.getSplitQueueCapacity()));
+ Counter generatedSplits =
generationProfile.addCounter("GeneratedSplits", TUnit.UNIT,
+ RuntimeProfile.ROOT_COUNTER);
+ generatedSplits.setLevel(1);
+ if (summaryProfile != null) {
+ summaryProfile.getExecutionSummary().addChild(generationProfile,
true);
+ }
Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
- CompletableFuture.runAsync(() -> {
- ConnectorSplitSource source = null;
+ new SplitProducer(splitAssignment, scheduleExecutor, new
SplitProducer.BatchSource() {
Review Comment:
Refactoring suggestion: now that both async paths use `SplitProducer`, share
their common preparation and completion flow instead of maintaining two
complete pipelines in `startSplit()` and `startStreamingSplit()`. Both prepare
projection/filter state, pin the snapshot and rewrite scope, build a request,
start producers, collect profiles, and finish scheduling. The meaningful
differences are the source factory and producer count. The existing
`SubmittedTaskFinalizer` can cover one producer as well as multiple producers;
no additional lifecycle interface is needed. Preserve the current
initialization order and prune-to-zero/capability semantics. Some duplication
predates this PR; this is an opportunity to consolidate the paths while
changing their common execution mechanism, not a new correctness finding.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -248,6 +247,187 @@ 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);
+ long fileCount = cachedFileCount(hiveHandle);
+ // 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);
+ }
+ List<HmsPartitionInfo> partitions = hiveHandle.getPrunedPartitions();
+ int partitionCount = partitions == null ? 1 : partitions.size();
+ int partitionThreshold = sessionInt(session,
"num_partitions_in_batch_mode", DEFAULT_BATCH_THRESHOLD);
+ boolean partitionBatch = partitionThreshold > 0 && partitionCount >=
partitionThreshold;
+ return fileCount >= fileThreshold || partitionBatch ? fileCount : -1;
+ }
+
+ private long cachedFileCount(HiveTableHandle handle) {
+ List<String> partitionKeys = handle.getPartitionKeyNames();
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return fileListingCache.cachedFileCount(handle.getDbName(),
handle.getTableName(),
+ handle.getLocation(), Collections.emptyList());
+ }
+ List<HmsPartitionInfo> partitions = handle.getPrunedPartitions();
+ if (partitions == null) {
+ return -1;
+ }
+ long fileCount = 0;
+ for (HmsPartitionInfo partition : partitions) {
+ long count = fileListingCache.cachedFileCount(handle.getDbName(),
handle.getTableName(),
+ partition.getLocation(), partition.getValues());
+ if (count < 0) {
+ return -1;
+ }
+ fileCount += count;
+ }
+ return fileCount;
+ }
+
+ @Override
+ public ConnectorSplitSource streamSplits(ConnectorSession session,
ConnectorTableHandle handle,
+ List<ConnectorColumnHandle> columns, Optional<ConnectorExpression>
filter, long limit) {
+ return streamSplits(session, ConnectorScanRequest.builder(handle,
columns).filter(filter).limit(limit).build());
+ }
+
+ @Override
+ public ConnectorSplitSource streamSplits(ConnectorSession session,
ConnectorScanRequest request) {
+ HiveTableHandle hiveHandle = (HiveTableHandle)
request.getTableHandle();
+ if (hiveHandle.isTransactional()) {
+ throw new DorisConnectorException("Transactional Hive scans
require snapshot-aware planning");
+ }
+ return new HiveSplitSource(session, hiveHandle,
request.getRequiredPartitions());
+ }
+
+ private static int sessionInt(ConnectorSession session, String name, int
defaultValue) {
+ Integer value = session.getProperty(name, Integer.class);
+ return value == null ? defaultValue : value;
+ }
+
+ private final class HiveSplitSource implements ConnectorSplitSource {
+ private final HiveTableHandle handle;
+ private final FileSystem fs;
+ private final HiveFileFormat tableFormat;
+ private final boolean isLzo;
+ private final boolean splittable;
+ private final long targetSplitSize;
+ private final int partitionBatchSize;
+ private Iterator<PartitionScanInfo> partitions =
Collections.emptyIterator();
+ private Iterator<HiveFileStatus> files = Collections.emptyIterator();
+ private Iterator<ConnectorScanRange> ranges =
Collections.emptyIterator();
+ private List<String> partitionNames;
+ private PartitionScanInfo partition;
+ private int nextPartition;
+ private boolean unpartitionedInitialized;
+ private volatile boolean closed;
+
+ HiveSplitSource(ConnectorSession session, HiveTableHandle handle,
List<String> requiredPartitions) {
+ this.handle = handle;
+ this.partitionNames = requiredPartitions.isEmpty() ? null :
requiredPartitions;
+ this.fs = storage().getFileSystem(session);
+ this.tableFormat = HiveFileFormat.detect(handle.getInputFormat(),
handle.getSerializationLib(),
Review Comment:
Refactoring suggestion: reuse this partition-scoped lazy enumeration in
Hive's synchronous `planScanForPartitionBatch()` as well, by consuming the
source into a list. `doPlanScanForPartitionBatch()` currently maintains another
copy of partition resolution, format/LZO setup, file listing, and split
traversal. One enumeration implementation with streaming and collecting
consumers would keep those rules consistent, without another SPI method. The
existing `splitFile()` adapter over `splitFileRanges()` already follows this
pattern. Keep the initial consolidation limited to non-ACID partition batches:
whole-scan `planScan()` also tracks `PlanCompleteness` to avoid memoizing
incomplete results, so that signal must be preserved before sharing its entire
path. Verify eager/lazy equivalence for partition subsets, LZO, empty files and
tolerated directory-listing failures.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitAssignment.java:
##########
@@ -70,12 +76,29 @@ public SplitAssignment(
Map<String, String> locationProperties,
List<String> pathPartitionKeys,
boolean fileCacheAdmission) {
+ this(backendPolicy, splitGenerator, splitToScanRange,
locationProperties, pathPartitionKeys,
+ fileCacheAdmission, 128, MAX_QUEUED_BATCHES);
Review Comment:
Cleanup suggestion: remove the six-argument constructor and its
`MAX_QUEUED_BATCHES` fallback, and pass explicit limits in its remaining test
caller. The only production construction site, `FileQueryScanNode`, now
supplies the session values; this overload is used only by
`SplitAssignmentTest`. Keeping it introduces two implicit defaults: 128/10000
here versus the production session defaults of 128/32. There is no remaining
production caller to preserve, and explicit test limits avoid retaining a
misleading compatibility entry point. This is not a claim that the current
production queue ignores its configured capacity.
--
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]
