This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new dcee644c957 [improvement](fe) Default external scans to bounded split
batches (#69683)
dcee644c957 is described below
commit dcee644c957b135d9548f31f59139ba8cf892681
Author: 924060929 <[email protected]>
AuthorDate: Thu Oct 29 10:08:37 2026 +0800
[improvement](fe) Default external scans to bounded split batches (#69683)
### What problem does this PR solve?
Problem Summary: External scans currently enter Batch Mode only above
1024 partitions or files. Even in Batch Mode, a producer batch defaults
to 128 splits and each backend queue is bounded by 32 **batches**. That
couples the memory bound to batch size and delays small LIMIT scans
while large groups are prepared.
This PR enables Batch Mode from one partition or file by default. It
separates the Hive partition-processing batch size
(`split_partition_batch_size`, default 16) from the Batch Mode trigger.
Producers still assign and offer splits in batches, but each backend
queue stores individual splits and enforces a capacity of 1024 splits; a
partly full queue accepts only the available slots and wakes the
producer when the consumer drains it. The producer batch defaults to 8
splits. The consumer still obeys the BE request batch size, while
preserving cancellation, first-fetch limits, backpressure, and error
propagation. The new connector SPI method advances the plugin API
version to 25 and updates its recorded surface.
A slow Hive partition can outlast the one-second split-fetch RPC wait
after another partition supplies the first split. FE now returns
`INCOMPLETE` while production is still running, and BE retries after
checking cancellation. A completed assignment still returns EOF, and
producer failures remain errors. If a local LIMIT stops a scanner while
other query branches remain active, its BE retry loop observes that
scanner stop and exits without waiting for more splits.
In a matched, one-BE Release benchmark on the integrated test branch,
refreshed-metadata LIMIT medians at `split_batch_size=8` versus `32`
were **75.8 vs 102.5 ms** for Hive 10k partitions and **23.5 vs 30.6
ms** for Iceberg 1000 manifests. Eight beat 32 in 11/12 paired blocks in
each LIMIT scenario. Warm full scans had no stable winner among
8/16/32/64. These numbers select the producer batch size; they are not a
fresh end-to-end benchmark of this rebased PR against current master.
### Release note
External table Batch Mode now starts at one partition or file by
default. The default producer batch size is 8 splits, queue capacity is
1024 splits per backend, and `split_partition_batch_size` controls Hive
partition-processing batches separately.
---
be/src/exec/scan/file_scanner.cpp | 4 +-
be/src/exec/scan/file_scanner_v2.cpp | 2 +-
be/src/exec/scan/scanner.h | 2 +-
be/src/exec/scan/split_source_connector.cpp | 51 ++++--
be/src/exec/scan/split_source_connector.h | 11 +-
be/test/exec/scan/file_scanner_v2_test.cpp | 13 ++
be/test/exec/scan/vfile_scanner_exception_test.cpp | 3 +-
be/test/format/wal/wal_manager_test.cpp | 3 +-
fe/fe-connector/docs/scan-task-scheduling.md | 4 +-
.../apache/doris/connector/hive/HiveConnector.java | 16 +-
.../doris/connector/hive/HiveScanPlanProvider.java | 9 +-
.../hive/HiveConnectorPartitionViewCacheTest.java | 23 +++
.../connector/hive/HiveScanBatchModeTest.java | 47 ++++-
.../spi/scan/ConnectorScanPlanProvider.java | 10 ++
.../connector/spi/ConnectorPluginSurfaceTest.java | 5 +-
.../ConnectorScanPlanProviderBatchScanTest.java | 6 +
.../test/resources/connector-plugin-surface.txt | 1 +
fe/fe-connector/pom.xml | 2 +-
.../doris/datasource/scan/FileQueryScanNode.java | 14 +-
.../datasource/scan/PluginDrivenScanNode.java | 9 +-
.../doris/datasource/split/SplitAssignment.java | 189 ++++++++++++++-------
.../doris/datasource/split/SplitProducer.java | 12 +-
.../apache/doris/datasource/split/SplitSource.java | 73 ++++----
.../java/org/apache/doris/qe/SessionVariable.java | 33 +++-
.../apache/doris/service/FrontendServiceImpl.java | 4 +
.../scan/PluginDrivenScanNodeBatchModeTest.java | 1 +
.../PluginDrivenScanNodeHiveStreamingTest.java | 26 ++-
.../PluginDrivenScanNodePartitionBatchTest.java | 1 -
.../datasource/split/SplitAssignmentTest.java | 89 +++++++---
.../doris/datasource/split/SplitProducerTest.java | 49 ++++--
.../doris/datasource/split/SplitSourceTest.java | 161 +++++++++++++++---
.../split/SplitStreamingConcurrencyTest.java | 26 ++-
.../org/apache/doris/qe/SessionVariablesTest.java | 9 +-
.../hive/test_hive_streaming_splits.groovy | 4 +-
.../iceberg/test_iceberg_deletion_vector.groovy | 6 +-
.../iceberg/test_iceberg_position_delete.groovy | 7 +
.../paimon/test_paimon_deletion_vector_oss.groovy | 4 +-
37 files changed, 696 insertions(+), 233 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index ff210680576..f4cac2c5a2f 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -493,7 +493,7 @@ Status FileScanner::_open_impl(RuntimeState* state) {
if (_local_state) {
_condition_cache_digest = _local_state->get_condition_cache_digest();
}
- RETURN_IF_ERROR(_split_source->get_next(&_first_scan_range,
&_current_range));
+ RETURN_IF_ERROR(_split_source->get_next(&_first_scan_range,
&_current_range, &_should_stop));
if (_first_scan_range) {
RETURN_IF_ERROR(_init_expr_ctxes());
if (_state->query_options().enable_runtime_filter_partition_prune &&
@@ -1021,7 +1021,7 @@ Status FileScanner::_get_next_reader() {
_src_block_init = false;
bool has_next = _first_scan_range;
if (!_first_scan_range) {
- RETURN_IF_ERROR(_split_source->get_next(&has_next,
&_current_range));
+ RETURN_IF_ERROR(_split_source->get_next(&has_next,
&_current_range, &_should_stop));
}
_first_scan_range = false;
if (!has_next || _should_stop) {
diff --git a/be/src/exec/scan/file_scanner_v2.cpp
b/be/src/exec/scan/file_scanner_v2.cpp
index d9ab6a377fc..a1a2456a0c9 100644
--- a/be/src/exec/scan/file_scanner_v2.cpp
+++ b/be/src/exec/scan/file_scanner_v2.cpp
@@ -492,7 +492,7 @@ Status FileScannerV2::_open_impl(RuntimeState* state) {
Status FileScannerV2::_get_next_scan_range(bool* has_next) {
SCOPED_TIMER(_get_next_range_timer);
DORIS_CHECK(has_next != nullptr);
- RETURN_IF_ERROR(_split_source->get_next(has_next, &_current_range));
+ RETURN_IF_ERROR(_split_source->get_next(has_next, &_current_range,
&_should_stop));
if (*has_next) {
RETURN_IF_ERROR(_validate_scan_range(*_params, _current_range));
}
diff --git a/be/src/exec/scan/scanner.h b/be/src/exec/scan/scanner.h
index dfed572406e..9dd23f5b84b 100644
--- a/be/src/exec/scan/scanner.h
+++ b/be/src/exec/scan/scanner.h
@@ -318,7 +318,7 @@ protected:
int64_t _per_scanner_open_timer = 0;
int64_t _projection_timer = 0;
- bool _should_stop = false;
+ std::atomic_bool _should_stop = false;
// Cached pointer to ScanOperator's remaining-limit counter. Null when
// this scanner is on the topn path or the query has no LIMIT.
diff --git a/be/src/exec/scan/split_source_connector.cpp
b/be/src/exec/scan/split_source_connector.cpp
index 1e22f8990d1..ebcac31662d 100644
--- a/be/src/exec/scan/split_source_connector.cpp
+++ b/be/src/exec/scan/split_source_connector.cpp
@@ -26,7 +26,8 @@ namespace doris {
using apache::thrift::transport::TTransportException;
-Status LocalSplitSourceConnector::get_next(bool* has_next, TFileRangeDesc*
range) {
+Status LocalSplitSourceConnector::get_next(bool* has_next, TFileRangeDesc*
range,
+ const std::atomic_bool*
/*scanner_stop*/) {
std::lock_guard<std::mutex> l(_range_lock);
*has_next = false;
if (_scan_index < _scan_ranges.size()) {
@@ -43,31 +44,47 @@ Status LocalSplitSourceConnector::get_next(bool* has_next,
TFileRangeDesc* range
return Status::OK();
}
-Status RemoteSplitSourceConnector::get_next(bool* has_next, TFileRangeDesc*
range) {
+Status RemoteSplitSourceConnector::get_next(bool* has_next, TFileRangeDesc*
range,
+ const std::atomic_bool*
scanner_stop) {
std::lock_guard<std::mutex> l(_range_lock);
*has_next = false;
if (_scan_index == _scan_ranges.size() && !_last_batch) {
SCOPED_TIMER(_get_split_timer);
- Status coord_status;
- // No need to set timeout because on FE side, there is a max fetch time
- FrontendServiceConnection
coord(_state->exec_env()->frontend_client_cache(),
- _state->get_query_ctx()->coord_addr,
&coord_status);
- RETURN_IF_ERROR(coord_status);
TFetchSplitBatchRequest request;
request.__set_split_source_id(_split_source_id);
request.__set_max_num_splits(config::remote_split_source_batch_size);
TFetchSplitBatchResult result;
- try {
- coord->fetchSplitBatch(result, request);
- if (result.__isset.status && result.status.status_code !=
TStatusCode::OK) {
- return Status::IOError<false>("Failed to get batch of split
source: {}",
- result.status.error_msgs.empty()
- ? "unknown error"
- :
result.status.error_msgs[0]);
+ do {
+ RETURN_IF_CANCELLED(_state);
+ // A local LIMIT stops this scanner without cancelling the query.
Do not keep a scan
+ // worker in the retry loop while another branch of the same query
remains active.
+ if (scanner_stop->load(std::memory_order_acquire)) {
+ return Status::OK();
}
- } catch (std::exception& e) {
- return Status::IOError<false>("Failed to get batch of split
source: {}", e.what());
- }
+ Status coord_status;
+ // Each FE call is bounded by the split-fetch wait interval.
Release the client between
+ // retries so a slow listing does not reserve a connection for the
whole query.
+ FrontendServiceConnection
coord(_state->exec_env()->frontend_client_cache(),
+
_state->get_query_ctx()->coord_addr, &coord_status);
+ RETURN_IF_ERROR(coord_status);
+ try {
+ coord->fetchSplitBatch(result, request);
+ if (result.__isset.status && result.status.status_code ==
TStatusCode::INCOMPLETE) {
+ // One Hive partition may have supplied the first range
while another is still
+ // listing files. FE's bounded wait is not EOF or a query
failure in that case.
+ continue;
+ }
+ if (result.__isset.status && result.status.status_code !=
TStatusCode::OK) {
+ return Status::IOError<false>("Failed to get batch of
split source: {}",
+
result.status.error_msgs.empty()
+ ? "unknown error"
+ :
result.status.error_msgs[0]);
+ }
+ } catch (std::exception& e) {
+ return Status::IOError<false>("Failed to get batch of split
source: {}", e.what());
+ }
+ break;
+ } while (true);
_last_batch = result.splits.empty();
_merge_ranges<TScanRangeLocations>(_scan_ranges, result.splits);
_scan_index = 0;
diff --git a/be/src/exec/scan/split_source_connector.h
b/be/src/exec/scan/split_source_connector.h
index f13190cf064..7dbcebe7543 100644
--- a/be/src/exec/scan/split_source_connector.h
+++ b/be/src/exec/scan/split_source_connector.h
@@ -18,6 +18,7 @@
#pragma once
#include <algorithm>
+#include <atomic>
#include "common/config.h"
#include "core/custom_allocator.h"
@@ -40,8 +41,10 @@ public:
* Get the next scan range. has_next should be to true to fetch the next
scan range.
* @param has_next whether exists the next scan range
* @param range the obtained next scan range
+ * @param scanner_stop the caller's local stop signal, which can be set
without query cancellation
*/
- virtual Status get_next(bool* has_next, TFileRangeDesc* range) = 0;
+ virtual Status get_next(bool* has_next, TFileRangeDesc* range,
+ const std::atomic_bool* scanner_stop) = 0;
virtual int num_scan_ranges() = 0;
@@ -116,7 +119,8 @@ public:
_merge_ranges<TScanRangeParams>(_scan_ranges, scan_ranges);
}
- Status get_next(bool* has_next, TFileRangeDesc* range) override;
+ Status get_next(bool* has_next, TFileRangeDesc* range,
+ const std::atomic_bool* scanner_stop) override;
int num_scan_ranges() override { return
static_cast<int>(_scan_ranges.size()); }
@@ -174,7 +178,8 @@ public:
_max_scanners = max_scanners;
}
- Status get_next(bool* has_next, TFileRangeDesc* range) override;
+ Status get_next(bool* has_next, TFileRangeDesc* range,
+ const std::atomic_bool* scanner_stop) override;
/*
* Remote split source is fetched in batch mode, and the splits are
generated while scanning,
diff --git a/be/test/exec/scan/file_scanner_v2_test.cpp
b/be/test/exec/scan/file_scanner_v2_test.cpp
index fd3d19a7337..9637a8a51e7 100644
--- a/be/test/exec/scan/file_scanner_v2_test.cpp
+++ b/be/test/exec/scan/file_scanner_v2_test.cpp
@@ -20,6 +20,7 @@
#include <gen_cpp/PlanNodes_types.h>
#include <gtest/gtest.h>
+#include <atomic>
#include <memory>
#include <optional>
#include <string>
@@ -587,6 +588,18 @@ TEST(FileScannerV2Test,
LegacyCountExemptionRequiresMetadataCountOnEveryRange) {
EXPECT_FALSE(invalid.all_ranges_have_table_level_row_count());
}
+TEST(FileScannerV2Test, RemoteSplitSourceStopsLocallyWithoutQueryCancellation)
{
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ RemoteSplitSourceConnector source(&state, nullptr, 1, 1, 1);
+ std::atomic_bool scanner_stop {true};
+ bool has_next = true;
+ TFileRangeDesc range;
+
+ ASSERT_FALSE(state.is_cancelled());
+ ASSERT_TRUE(source.get_next(&has_next, &range, &scanner_stop).ok());
+ EXPECT_FALSE(has_next);
+}
+
TEST(FileScannerV2Test, JniCompatibilityShapesUseV2Scanner) {
TQueryOptions query_options;
query_options.__set_enable_file_scanner_v2(true);
diff --git a/be/test/exec/scan/vfile_scanner_exception_test.cpp
b/be/test/exec/scan/vfile_scanner_exception_test.cpp
index 62bc3d73011..4e03153c54f 100644
--- a/be/test/exec/scan/vfile_scanner_exception_test.cpp
+++ b/be/test/exec/scan/vfile_scanner_exception_test.cpp
@@ -54,7 +54,8 @@ private:
public:
TestSplitSourceConnectorStub(const TFileScanRange& scan_range) :
_scan_range(scan_range) {}
- Status get_next(bool* has_next, TFileRangeDesc* range) override {
+ Status get_next(bool* has_next, TFileRangeDesc* range,
+ const std::atomic_bool* /*scanner_stop*/) override {
std::lock_guard<std::mutex> l(_range_lock);
if (_range_index < _scan_range.ranges.size()) {
*has_next = true;
diff --git a/be/test/format/wal/wal_manager_test.cpp
b/be/test/format/wal/wal_manager_test.cpp
index 7bdafc58cfa..96762d9b442 100644
--- a/be/test/format/wal/wal_manager_test.cpp
+++ b/be/test/format/wal/wal_manager_test.cpp
@@ -47,7 +47,8 @@ private:
public:
TestSplitSourceConnector(const TFileScanRange& scan_range) :
_scan_range(scan_range) {}
- Status get_next(bool* has_next, TFileRangeDesc* range) override {
+ Status get_next(bool* has_next, TFileRangeDesc* range,
+ const std::atomic_bool* /*scanner_stop*/) override {
std::lock_guard<std::mutex> l(_range_lock);
if (_range_index < _scan_range.ranges.size()) {
*has_next = true;
diff --git a/fe/fe-connector/docs/scan-task-scheduling.md
b/fe/fe-connector/docs/scan-task-scheduling.md
index ebfb05f5a39..35e8f9bd84a 100644
--- a/fe/fe-connector/docs/scan-task-scheduling.md
+++ b/fe/fe-connector/docs/scan-task-scheduling.md
@@ -108,11 +108,11 @@ shared cache loader. Normal completion retains the
separate read-lease cleanup.
### Consumer completion notification
-Each backend has a bounded split queue and its own data-ready wait channel. A
consumer registers before checking queue contents and terminal state, then
waits only when neither is available. Successful enqueue, producer completion,
failure and cancellation signal that channel. The empty queue is not
periodically polled to discover EOF: otherwise finishing just after a poll
begins adds 100 ms despite completed production. Producers avoid the channel
monitor when no consumer waits; callbac [...]
+Each backend has a split queue bounded by the number of queued splits and its
own data-ready wait channel. Producers offer batches, but individual splits are
the queue entries. A consumer registers before checking queue contents and
terminal state, then waits only when neither is available. Successful enqueue,
producer completion, failure and cancellation signal that channel. The empty
queue is not periodically polled to discover EOF: otherwise finishing just
after a poll begins adds 100 [...]
### Initial delivery and producer lookahead
-`split_initial_batch_size` (default 64) bounds the first Split fetch per
backend, also capped by the BE request size. Subsequent fetches use the BE
request size. The bound is applied before compatible file ranges are compacted
into RPC envelopes. It is independent of `split_batch_size` (producer batch
size) and `split_queue_capacity` (buffered batches per backend). Producers
continue listing files and preparing splits before BE consumption, yielding
only for task fairness and configured [...]
+`split_initial_batch_size` (default 64) bounds the first Split fetch per
backend, also capped by the BE request size. Subsequent fetches use the BE
request size. The bound is applied before compatible file ranges are compacted
into RPC envelopes. `split_batch_size` (default 8) limits producer batches,
while `split_queue_capacity` (default 1024) limits buffered splits per backend.
A configured queue capacity smaller than the batch size also reduces the
effective producer batch size. Produ [...]
Actual Manifest readers retain a lease on the exact read-table generation
until the callable exits. Query cleanup releases the statement owner, while an
already running shared reader keeps its FileIO alive. Future cancellation is
not evidence that the callable has finished.
diff --git
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java
index 013db6a79f0..4a8016822b1 100644
---
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java
+++
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector.hive;
+import org.apache.doris.connector.cache.CacheSpec;
import org.apache.doris.connector.cache.CatalogMetaCache;
import org.apache.doris.connector.cache.ConnectorMetadataCache;
import org.apache.doris.connector.cache.ScopePath;
@@ -149,14 +150,23 @@ public class HiveConnector implements Connector {
this.context = context;
this.metaCache = CatalogMetaCache.managed(context.getCatalogId(),
"hive", this.properties);
this.fileListingCache = new HiveFileListingCache(metaCache, props);
- // Reads its own
meta.cache.hive.partition_view.(enable|ttl-second|capacity) from the catalog
properties
- // via the framework's CacheSpec (default ON / 24h / 1000).
+ // The derived view must inherit the partition-name TTL when no view
TTL is specified. Otherwise
+ // partition.cache.ttl-second=0 still leaves a 24h cached view, and
batch scans miss new partitions.
this.partitionViewCache = new ConnectorMetadataCache<>(metaCache,
"hive-partition-view",
- "hive", "partition_view", this.properties,
+ "hive", "partition_view",
partitionViewProperties(this.properties),
key -> ScopePath.partitionCollection(key.getDb(),
key.getTable()),
HivePartitionViewSizeEstimator::estimateEntry);
}
+ private static Map<String, String> partitionViewProperties(Map<String,
String> properties) {
+ String partitionNamesTtl = CacheSpec.metaCacheTtlKey("hive",
"partition_names");
+ String partitionViewTtl = CacheSpec.metaCacheTtlKey("hive",
"partition_view");
+ Map<String, String> withLegacyTtl =
CacheSpec.applyCompatibilityMap(properties,
+ Collections.singletonMap("partition.cache.ttl-second",
partitionNamesTtl));
+ return CacheSpec.applyCompatibilityMap(withLegacyTtl,
+ Collections.singletonMap(partitionNamesTtl, partitionViewTtl));
+ }
+
@Override
public ConnectorMetadata getMetadata(ConnectorSession session) {
return newMetadata(getOrCreateClient());
diff --git
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
index b424305cf5d..13f1685d6eb 100644
---
a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
@@ -83,6 +83,7 @@ public class HiveScanPlanProvider implements
ConnectorScanPlanProvider {
/** Maximum number of partitions to list from HMS. */
private static final int MAX_PARTITIONS = 100000;
private static final int DEFAULT_BATCH_THRESHOLD = 1024;
+ private static final int DEFAULT_PARTITION_BATCH_SIZE = 16;
/**
* Session variable gating the OpenX-JSON "read the whole JSON row into
one CSV column" mode (legacy
@@ -213,6 +214,12 @@ public class HiveScanPlanProvider implements
ConnectorScanPlanProvider {
return partKeyNames != null && !partKeyNames.isEmpty();
}
+ @Override
+ public int partitionBatchSize(ConnectorSession session, int
partitionCount, int parallelism) {
+ int configured = sessionInt(session, "split_partition_batch_size",
DEFAULT_PARTITION_BATCH_SIZE);
+ return Math.min(configured, (partitionCount - 1) / parallelism + 1);
+ }
+
@Override
public long streamingSplitEstimate(ConnectorSession session,
ConnectorTableHandle handle,
Optional<ConnectorExpression> filter, boolean countPushdown) {
@@ -303,7 +310,7 @@ public class HiveScanPlanProvider implements
ConnectorScanPlanProvider {
this.splittable = tableFormat.isSplittable() && !isLzo;
this.targetSplitSize = getTargetSplitSize(session);
this.partitionBatchSize = Math.max(1,
- sessionInt(session, "num_partitions_in_batch_mode",
DEFAULT_BATCH_THRESHOLD));
+ sessionInt(session, "split_partition_batch_size",
DEFAULT_PARTITION_BATCH_SIZE));
recordPruningProfile(handle);
}
diff --git
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorPartitionViewCacheTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorPartitionViewCacheTest.java
index 309a5c64153..9c6b92b0fe8 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorPartitionViewCacheTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveConnectorPartitionViewCacheTest.java
@@ -55,6 +55,29 @@ public class HiveConnectorPartitionViewCacheTest {
"a fresh hive connector must always build the partition-view
cache");
}
+ @Test
+ public void disabledPartitionNameCacheAlsoDisablesDerivedView() {
+ for (String ttlKey : new String[] {
+ "partition.cache.ttl-second",
"meta.cache.hive.partition_names.ttl-second"}) {
+ Map<String, String> catalogProperties = props();
+ catalogProperties.put(ttlKey, "0");
+ HiveConnector connector = new HiveConnector(catalogProperties, new
FakeConnectorContext());
+ ConnectorMetadataCache<List<ConnectorPartitionInfo>> cache =
connector.partitionViewCacheForTest();
+ Assertions.assertFalse(cache.isEnabled(), ttlKey);
+ int[] loads = {0};
+ ConnectorTableKey key = new ConnectorTableKey("db", "table", -1L,
-1L);
+ cache.get(key, () -> {
+ loads[0]++;
+ return Collections.emptyList();
+ });
+ cache.get(key, () -> {
+ loads[0]++;
+ return Collections.emptyList();
+ });
+ Assertions.assertEquals(2, loads[0], ttlKey + " must not keep a
stale partition view");
+ }
+ }
+
@Test
public void refreshHooksInvalidatePartitionViewCache() {
// The REFRESH hooks must clear cache A too (else external DDL/writes
stay invisible beyond the TTL):
diff --git
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
index 2d3d39282a9..e45741a1246 100644
---
a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
+++
b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
@@ -216,7 +216,7 @@ public class HiveScanBatchModeTest {
});
HiveTableHandle handle =
parquetHandle(PART_KEYS).toBuilder().prunedPartitions(Arrays.asList(
part("year=2024/month=01"),
part("year=2024/month=02"))).build();
- FakeSession session = new
FakeSession(Collections.singletonMap("num_partitions_in_batch_mode", "1"),
+ FakeSession session = new
FakeSession(Collections.singletonMap("split_partition_batch_size", "1"),
scheduler);
try (ConnectorSplitSource source = provider.streamSplits(session,
ConnectorScanRequest.builder(handle,
Collections.emptyList()).build())) {
@@ -262,6 +262,40 @@ public class HiveScanBatchModeTest {
Assertions.assertEquals(Collections.singleton("year=2024/month=02"),
lister.callsPerLocation.keySet());
}
+ @Test
+ public void partitionMetadataBatchSizeDoesNotFollowBatchModeThreshold()
throws Exception {
+ List<Integer> requestedBatchSizes = new ArrayList<>();
+ FakeHmsClient hms = new FakeHmsClient() {
+ @Override
+ public HmsPartitionBatchResult getExistingPartitionsWithStats(
+ String dbName, String tableName, List<String>
partitionNames) {
+ requestedBatchSizes.add(partitionNames.size());
+ return super.getExistingPartitionsWithStats(dbName, tableName,
partitionNames);
+ }
+ };
+ CountingLister lister = new CountingLister();
+ HiveScanPlanProvider provider = provider(hms, lister);
+ List<String> names = Arrays.asList("year=2024/month=01",
"year=2024/month=02",
+ "year=2024/month=03", "year=2024/month=04",
"year=2024/month=05");
+ Map<String, String> settings = new HashMap<>();
+ settings.put("num_partitions_in_batch_mode", "1");
+ settings.put("split_partition_batch_size", "2");
+ ConnectorScanRequest request = ConnectorScanRequest.builder(
+ parquetHandle(PART_KEYS),
Collections.emptyList()).requiredPartitions(names).build();
+
+ int ranges = 0;
+ try (ConnectorSplitSource source = provider.streamSplits(new
FakeSession(settings), request)) {
+ while (source.hasNext()) {
+ source.next();
+ ranges++;
+ }
+ }
+
+ Assertions.assertEquals(5, ranges);
+ Assertions.assertEquals(Arrays.asList(2, 2, 1), requestedBatchSizes);
+ Assertions.assertEquals(5, lister.totalCalls);
+ }
+
@Test
public void sharedCompressionNamesMapToNativeCodecs() {
for (FileCompression codec : FileCompression.values()) {
@@ -360,7 +394,7 @@ public class HiveScanBatchModeTest {
.prunedPartitions(Collections.singletonList(part("year=2023/month=12"))).build();
Map<String, String> settings = new HashMap<>();
settings.put("file_split_size", "8");
- settings.put("num_partitions_in_batch_mode", "1");
+ settings.put("split_partition_batch_size", "1");
FakeSession session = new FakeSession(settings);
ConnectorScanRequest request =
ConnectorScanRequest.builder(handle, Collections.emptyList())
.requiredPartitions(Collections.singletonList("year=2023/month=12")).build();
@@ -437,6 +471,15 @@ public class HiveScanBatchModeTest {
Assertions.assertTrue(provider.supportsBatchScan(new FakeSession(),
handle));
}
+ @Test
+ public void
partitionBatchSizePreservesParallelListingAndBoundsHmsRequests() {
+ HiveScanPlanProvider provider = provider(null, new CountingLister());
+ Assertions.assertEquals(16, provider.partitionBatchSize(new
FakeSession(), 1000, 64));
+ Assertions.assertEquals(5, provider.partitionBatchSize(new
FakeSession(), 9, 2));
+ Assertions.assertEquals(2, provider.partitionBatchSize(
+ new
FakeSession(Collections.singletonMap("split_partition_batch_size", "2")), 1000,
64));
+ }
+
@Test
public void supportsBatchScanIsFalseForUnpartitionedTable() {
HiveScanPlanProvider provider = provider(null, new CountingLister());
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java
index 0d2203d44a3..a4965b3db2c 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProvider.java
@@ -222,6 +222,16 @@ public interface ConnectorScanPlanProvider {
return false;
}
+ /**
+ * Number of partitions assigned to one split-production batch. The
batch-mode threshold and
+ * the amount of work in a batch serve different purposes: the former
selects the asynchronous
+ * path, while this size bounds each connector request. Connectors with
cheap, incremental
+ * partition resolution may choose a smaller size than the legacy
read-session batch.
+ */
+ default int partitionBatchSize(ConnectorSession session, int
partitionCount, int parallelism) {
+ return 1024;
+ }
+
/**
* Whether this connector's scan ranges carry meaningful byte lengths
* ({@link ConnectorScanRange#getLength()}) so the engine can apply {@code
TABLESAMPLE} by
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
index f0e0415c7f7..4a7679858e0 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
+++
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/ConnectorPluginSurfaceTest.java
@@ -91,8 +91,9 @@ public class ConnectorPluginSurfaceTest {
// Major 20 adds deleteUsingMetadata to the connector write surface;
major 21 adds
// typed filesystem credential descriptions and statement-scoped
metadata.
// Major 22 adds the metadata-only DDL handle path; major 23 adds the
provider's
- // ConnectorSecuritySurface declaration; major 24 adds the scan task
scheduler.
- Assertions.assertEquals("24.0", version.getProperty("api.version"));
+ // ConnectorSecuritySurface declaration; major 24 adds the scan task
scheduler;
+ // major 25 adds connector-sized partition batches.
+ Assertions.assertEquals("25.0", version.getProperty("api.version"));
}
/** Root entry points plus provider/handle types returned to connector
plugins. */
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProviderBatchScanTest.java
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProviderBatchScanTest.java
index 2f7f521c83c..f9f5ae31267 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProviderBatchScanTest.java
+++
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/scan/ConnectorScanPlanProviderBatchScanTest.java
@@ -111,6 +111,12 @@ public class ConnectorScanPlanProviderBatchScanTest {
Assertions.assertFalse(provider.supportsBatchScan(null, null));
}
+ @Test
+ public void testPartitionBatchSizeDefaultsToLegacyReadSessionSize() {
+ ConnectorScanPlanProvider provider = new RecordingProvider();
+ Assertions.assertEquals(1024, provider.partitionBatchSize(null, 1000,
64));
+ }
+
@Test
public void testStreamingSplitEstimateDefaultsNegative() {
// FIX-M3: default MUST be < 0 so no connector silently enters
file-count streaming without opting in
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
index 3cdbe7dc9f8..af4ce869463 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
+++
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
@@ -188,6 +188,7 @@
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#getScanNodePropert
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#ignorePartitionPruneShortCircuit():boolean
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#planScan(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.scan.ConnectorScanRequest):java.util.List
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#planScanForPartitionBatch(org.apache.doris.connector.spi.ConnectorSession,org.apache.doris.connector.spi.scan.ConnectorScanRequest,java.util.List):java.util.List
+org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#partitionBatchSize(org.apache.doris.connector.spi.ConnectorSession,int,int):int
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#populateScanLevelParams(org.apache.doris.thrift.TFileScanRangeParams,java.util.Map):void
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#releaseReadTransaction(java.lang.String):void
org.apache.doris.connector.spi.scan.ConnectorScanPlanProvider#scannedPartitionCount(java.util.List):java.util.OptionalLong
diff --git a/fe/fe-connector/pom.xml b/fe/fe-connector/pom.xml
index ad60cbf0d69..ba5f9d2ef1d 100644
--- a/fe/fe-connector/pom.xml
+++ b/fe/fe-connector/pom.xml
@@ -55,7 +55,7 @@ under the License.
of the latter two means bumping this property as well (and
fe-extension-spi means bumping
all five families).
-->
- <connector.plugin.api.version>24.0</connector.plugin.api.version>
+ <connector.plugin.api.version>25.0</connector.plugin.api.version>
</properties>
<modules>
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
index f10aa7066e6..5eed364f973 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FileQueryScanNode.java
@@ -33,6 +33,7 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.NotImplementedException;
+import org.apache.doris.common.Status;
import org.apache.doris.common.UserException;
import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.common.util.FileFormatUtils;
@@ -418,7 +419,18 @@ public abstract class FileQueryScanNode extends
FileScanNode {
splitAssignment = new SplitAssignment(backendPolicy, this,
this::splitToScanRange,
locationProperties, pathPartitionKeys, admissionResult,
sessionVariable.getSplitBatchSize(),
sessionVariable.getSplitQueueCapacity());
- splitAssignment.init();
+ if (executor != null) {
+ // Cancellation can arrive before a coordinator exists, while
the first producer is in
+ // remote IO. Closing the task group alone does not wake the
first-split planning wait.
+
executor.registerQueryCancellationCallback(splitAssignment::stop);
+ }
+ // The first split can require a remote read session (for example,
MaxCompute). Keep
+ // the planning wait bounded by the query's effective timeout
instead of a fixed 30 seconds.
+ splitAssignment.init(ConnectContext.get().getExecTimeoutS() *
1000L);
+ Status cancelReason = executor == null ? null :
executor.getPendingCancelReason();
+ if (cancelReason != null) {
+ throw new UserException(cancelReason.getErrorMsg());
+ }
if (executor != null) {
executor.getSummaryProfile().setGetSplitsFinishTime();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
index 89dccb471dc..3d1f3947232 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
@@ -2086,7 +2086,7 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
}
/**
- * Asynchronously generates splits in batches of {@code
num_partitions_in_batch_mode} partitions,
+ * Asynchronously generates splits in connector-sized partition batches,
* streaming each batch into {@link #splitAssignment}. Mirrors legacy
* {@code MaxComputeScanNode.startSplit}: one read session per partition
batch (built by the
* connector via {@link
ConnectorScanPlanProvider#planScanForPartitionBatch}) on the shared
@@ -2142,9 +2142,12 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
sources =
Collections.singletonList(createStreamingBatchSource(scanProvider,
request.withRequiredPartitions(requiredPartitions),
summaryProfile));
} else {
+ int partitionCount = selectedPartitions.selectedPartitions.size();
+ int batchSize = onPluginClassLoader(scanProvider, () ->
scanProvider.partitionBatchSize(
+ connectorSession, partitionCount,
Config.external_scan_split_threads));
sources = createPartitionBatchSources(scanProvider, request,
new
ArrayList<>(selectedPartitions.selectedPartitions.keySet()),
- sessionVariable.getNumPartitionsInBatchMode(),
Config.external_scan_split_threads);
+ batchSize, Config.external_scan_split_threads);
}
startSplitProducers(scanProvider, sources,
summaryProfile == null ? null :
summaryProfile.getExecutionSummary());
@@ -2277,7 +2280,7 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
private SplitProducer.BatchSource
createStreamingBatchSource(ConnectorScanPlanProvider scanProvider,
ConnectorScanRequest request, SummaryProfile summaryProfile) {
RuntimeProfile generationProfile = new RuntimeProfile("Split
Generation " + getId());
- generationProfile.addInfoString("BatchSize",
String.valueOf(sessionVariable.getSplitBatchSize()));
+ generationProfile.addInfoString("BatchSize",
String.valueOf(splitAssignment.getSplitBatchSize()));
generationProfile.addInfoString("QueueCapacity",
String.valueOf(sessionVariable.getSplitQueueCapacity()));
Counter generatedSplits =
generationProfile.addCounter("GeneratedSplits", TUnit.UNIT,
RuntimeProfile.ROOT_COUNTER);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitAssignment.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitAssignment.java
index 2f0a44c960f..d5ab601ecbb 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitAssignment.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitAssignment.java
@@ -29,15 +29,15 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.Closeable;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Deque;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -71,21 +71,103 @@ public class SplitAssignment {
private final ConcurrentHashMap<Long, Set<Runnable>> queueSpaceListeners =
new ConcurrentHashMap<>();
private volatile Runnable moreSplitsRequestedCallback = () -> {};
- /** Data readiness and terminal state use the same per-backend wait
channel. */
- private static final class AssignmentQueue {
- private final BlockingQueue<Collection<TScanRangeLocations>> batches;
- // Written under this channel's monitor; producers avoid that monitor
when nobody waits.
- private volatile int waitingReaders;
+ /** Per-backend scan ranges; bulk operations amortize synchronization
without reserving an entire batch. */
+ static final class AssignmentQueue {
+ private final Deque<TScanRangeLocations> splits = new ArrayDeque<>();
+ private final int capacity;
+ private int waitingReaders;
+ private int waitingProducers;
private AssignmentQueue(int capacity) {
- batches = new LinkedBlockingQueue<>(capacity);
+ this.capacity = capacity;
}
- private void signalReaders() {
- if (waitingReaders > 0) {
- synchronized (this) {
- notifyAll();
+ synchronized int offer(List<TScanRangeLocations> batch, int offset) {
+ Preconditions.checkArgument(!batch.isEmpty() && offset >= 0 &&
offset < batch.size(),
+ "Invalid split batch offset: %s", offset);
+ int accepted = Math.min(batch.size() - offset, capacity -
splits.size());
+ for (int i = offset; i < offset + accepted; i++) {
+ splits.addLast(batch.get(i));
+ }
+ if (accepted > 0 && waitingReaders > 0) {
+ notifyAll();
+ }
+ return accepted;
+ }
+
+ synchronized int offer(List<TScanRangeLocations> batch, int offset,
long timeout, TimeUnit unit)
+ throws InterruptedException {
+ Preconditions.checkArgument(!batch.isEmpty() && offset >= 0 &&
offset < batch.size(),
+ "Invalid split batch offset: %s", offset);
+ long remaining = unit.toNanos(timeout);
+ long deadline = System.nanoTime() + remaining;
+ while (splits.size() == capacity) {
+ if (remaining <= 0) {
+ return 0;
+ }
+ waitingProducers++;
+ try {
+ TimeUnit.NANOSECONDS.timedWait(this, remaining);
+ } finally {
+ waitingProducers--;
}
+ remaining = deadline - System.nanoTime();
+ }
+ return offer(batch, offset);
+ }
+
+ synchronized TScanRangeLocations poll() {
+ TScanRangeLocations split = splits.pollFirst();
+ if (split != null && waitingProducers > 0) {
+ notifyAll();
+ }
+ return split;
+ }
+
+ synchronized int drainTo(Collection<TScanRangeLocations> target, int
maxSplits) {
+ int count = 0;
+ while (count < maxSplits && !splits.isEmpty()) {
+ target.add(splits.removeFirst());
+ count++;
+ }
+ if (count > 0 && waitingProducers > 0) {
+ notifyAll();
+ }
+ return count;
+ }
+
+ synchronized TScanRangeLocations awaitSplit(long waitNanos,
SplitAssignment assignment)
+ throws InterruptedException {
+ waitingReaders++;
+ try {
+ if (splits.isEmpty() && assignment.needMoreSplit()) {
+ TimeUnit.NANOSECONDS.timedWait(this, waitNanos);
+ }
+ return poll();
+ } finally {
+ waitingReaders--;
+ }
+ }
+
+ synchronized boolean isEmpty() {
+ return splits.isEmpty();
+ }
+
+ synchronized int size() {
+ return splits.size();
+ }
+
+ synchronized int remainingCapacity() {
+ return capacity - splits.size();
+ }
+
+ synchronized TScanRangeLocations peek() {
+ return splits.peekFirst();
+ }
+
+ synchronized void signalWaiters() {
+ if (waitingReaders > 0 || waitingProducers > 0) {
+ notifyAll();
}
}
}
@@ -107,27 +189,34 @@ public class SplitAssignment {
this.locationProperties = locationProperties;
this.pathPartitionKeys = pathPartitionKeys;
this.fileCacheAdmission = fileCacheAdmission;
- this.splitBatchSize = splitBatchSize;
+ this.splitBatchSize = Math.min(splitBatchSize, queueCapacity);
this.queueCapacity = queueCapacity;
}
- public void init() throws UserException {
+ public void init(long timeoutMillis) throws UserException {
+ Preconditions.checkArgument(timeoutMillis > 0, "First split timeout
must be positive");
+ if (!needMoreSplit()) {
+ if (exception != null) {
+ throw exception;
+ }
+ return;
+ }
splitGenerator.startSplit(backendPolicy.numBackends());
+ long deadlineNanos = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
synchronized (readinessLock) {
- final int waitIntervalTimeMillis = 100;
- final int initTimeoutMillis = 30000; // 30s
- int waitTotalTime = 0;
while (sampleSplit == null && needMoreSplit()) {
+ long remainingNanos = deadlineNanos - System.nanoTime();
+ if (remainingNanos <= 0) {
+ throw new UserException("Failed to get first split after
waiting for "
+ + timeoutMillis + " ms.");
+ }
try {
- readinessLock.wait(waitIntervalTimeMillis);
+ TimeUnit.NANOSECONDS.timedWait(readinessLock,
+ Math.min(remainingNanos,
TimeUnit.MILLISECONDS.toNanos(100)));
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new UserException(e.getMessage(), e);
}
- waitTotalTime += waitIntervalTimeMillis;
- if (waitTotalTime > initTimeoutMillis) {
- throw new UserException("Failed to get first split after
waiting for "
- + (waitTotalTime / 1000) + " seconds.");
- }
}
}
if (exception != null) {
@@ -141,14 +230,12 @@ public class SplitAssignment {
private void appendBatch(Multimap<Backend, Split> batch) throws
UserException {
for (Backend backend : batch.keySet()) {
- Collection<TScanRangeLocations> locations =
createScanRanges(backend, batch.get(backend));
- while (needMoreSplit()) {
+ List<TScanRangeLocations> locations = createScanRanges(backend,
batch.get(backend));
+ int offset = 0;
+ while (offset < locations.size() && needMoreSplit()) {
AssignmentQueue queue = getOrCreateAssignmentQueue(backend);
try {
- if (queue.batches.offer(locations, 100,
TimeUnit.MILLISECONDS)) {
- queue.signalReaders();
- break;
- }
+ offset += queue.offer(locations, offset, 100,
TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
setException(new UserException("Failed to offer batch
split by interrupted", e));
@@ -164,7 +251,7 @@ public class SplitAssignment {
}
}
- Collection<TScanRangeLocations> createScanRanges(Backend backend,
Collection<Split> splits)
+ List<TScanRangeLocations> createScanRanges(Backend backend,
Collection<Split> splits)
throws UserException {
List<TScanRangeLocations> locations = new ArrayList<>(splits.size());
for (Split split : splits) {
@@ -181,30 +268,30 @@ public class SplitAssignment {
return locations;
}
- boolean offerBatch(Backend backend, Collection<TScanRangeLocations>
locations, Runnable resume) {
+ int offerBatch(Backend backend, List<TScanRangeLocations> locations, int
offset, Runnable resume) {
if (!needMoreSplit()) {
- return false;
+ return 0;
}
AssignmentQueue queue = getOrCreateAssignmentQueue(backend);
- if (queue.batches.offer(locations)) {
- queue.signalReaders();
+ int accepted = queue.offer(locations, offset);
+ if (accepted > 0) {
removeQueueSpaceListener(backend, resume);
- return true;
+ return accepted;
}
synchronized (queueSpaceListeners) {
// An error/stop may race range conversion. Do not register a
waiter after its wakeup.
if (!needMoreSplit()) {
- return false;
+ return 0;
}
queueSpaceListeners.computeIfAbsent(backend.getId(), key -> new
HashSet<>()).add(resume);
// A consumer with no registered waiter skips this monitor. Retry
after registration
// so capacity released between the first offer and registration
cannot be missed.
- if (queue.batches.offer(locations)) {
- queue.signalReaders();
+ accepted = queue.offer(locations, offset);
+ if (accepted > 0) {
removeQueueSpaceListener(backend, resume);
- return true;
+ return accepted;
}
- return false;
+ return 0;
}
}
@@ -265,35 +352,23 @@ public class SplitAssignment {
synchronized (readinessLock) {
readinessLock.notifyAll();
}
- assignment.values().forEach(AssignmentQueue::signalReaders);
+ assignment.values().forEach(AssignmentQueue::signalWaiters);
}
private AssignmentQueue getOrCreateAssignmentQueue(Backend backend) {
return assignment.computeIfAbsent(backend.getId(), be -> new
AssignmentQueue(queueCapacity));
}
- Collection<TScanRangeLocations> awaitBatch(Backend backend, long
waitNanos) throws InterruptedException {
+ TScanRangeLocations awaitSplit(Backend backend, long waitNanos) throws
InterruptedException {
AssignmentQueue queue = getOrCreateAssignmentQueue(backend);
- synchronized (queue) {
- queue.waitingReaders++;
- try {
- // Register before checking both predicates. An offer or
terminal transition either
- // precedes this check or signals after wait releases this
channel's monitor.
- if (queue.batches.isEmpty() && needMoreSplit()) {
- TimeUnit.NANOSECONDS.timedWait(queue, waitNanos);
- }
- return queue.batches.poll();
- } finally {
- queue.waitingReaders--;
- }
- }
+ return queue.awaitSplit(waitNanos, this);
}
- public BlockingQueue<Collection<TScanRangeLocations>>
getAssignedSplits(Backend backend) throws UserException {
+ AssignmentQueue getAssignedSplits(Backend backend) throws UserException {
if (exception != null) {
throw exception;
}
- BlockingQueue<Collection<TScanRangeLocations>> splits =
getOrCreateAssignmentQueue(backend).batches;
+ AssignmentQueue splits = getOrCreateAssignmentQueue(backend);
if (scheduleFinished.get() && splits.isEmpty() || isStopped.get()) {
return null;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitProducer.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitProducer.java
index 8fe29b198c4..a51017e37ab 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitProducer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitProducer.java
@@ -71,7 +71,8 @@ public class SplitProducer implements Runnable, Closeable {
private volatile boolean cancelled;
private Iterator<Map.Entry<Backend, Collection<Split>>> pending =
Collections.emptyIterator();
private Backend pendingBackend;
- private Collection<TScanRangeLocations> pendingRanges;
+ private List<TScanRangeLocations> pendingRanges;
+ private int pendingRangeOffset;
public SplitProducer(SplitAssignment assignment, Executor executor,
BatchSource source,
Consumer<Exception> completion) {
@@ -180,14 +181,19 @@ public class SplitProducer implements Runnable, Closeable
{
Map.Entry<Backend, Collection<Split>> entry = pending.next();
pendingBackend = entry.getKey();
pendingRanges = assignment.createScanRanges(pendingBackend,
entry.getValue());
+ pendingRangeOffset = 0;
}
- if (!assignment.offerBatch(pendingBackend, pendingRanges, resume))
{
+ int accepted = assignment.offerBatch(pendingBackend,
pendingRanges, pendingRangeOffset, resume);
+ if (accepted == 0) {
if (cancelled || !assignment.needMoreSplit()) {
complete(null);
}
return false;
}
- pendingRanges = null;
+ pendingRangeOffset += accepted;
+ if (pendingRangeOffset == pendingRanges.size()) {
+ pendingRanges = null;
+ }
}
complete(null);
return false;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
index 1ca7614242a..1488bfd25ed 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/split/SplitSource.java
@@ -24,11 +24,8 @@ import org.apache.doris.thrift.TScanRangeLocations;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
-import java.util.Iterator;
import java.util.List;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -44,6 +41,14 @@ import java.util.concurrent.atomic.AtomicLong;
* the split source when its related scan node is GC.
*/
public class SplitSource {
+ /** The producer is still running, but this fetch did not receive a split
within its RPC wait window. */
+ public static class SplitPendingException extends UserException {
+ public SplitPendingException(long maxWaitTime) {
+ super("Timeout waiting for the next split after " + maxWaitTime +
" ms; production is still running",
+ null, false, false);
+ }
+ }
+
private static final AtomicLong UNIQUE_ID_GENERATOR = new AtomicLong(0);
private final long uniqueId;
@@ -53,13 +58,13 @@ public class SplitSource {
private final AtomicBoolean firstBatchRequested = new AtomicBoolean();
private final long maxWaitTime;
private final int initialBatchSize;
- private Iterator<TScanRangeLocations> pendingSplits =
Collections.emptyIterator();
public SplitSource(Backend backend, SplitAssignment splitAssignment, long
maxWaitTime) {
this(backend, splitAssignment, maxWaitTime, Integer.MAX_VALUE);
}
public SplitSource(Backend backend, SplitAssignment splitAssignment, long
maxWaitTime, int initialBatchSize) {
+ Preconditions.checkArgument(maxWaitTime > 0, "Split fetch wait time
must be positive");
Preconditions.checkArgument(initialBatchSize > 0, "Initial split batch
size must be positive");
this.initialBatchSize = initialBatchSize;
this.uniqueId = UNIQUE_ID_GENERATOR.getAndIncrement();
@@ -94,49 +99,27 @@ public class SplitSource {
long waitNanos = TimeUnit.MILLISECONDS.toNanos(maxWaitTime);
while (scanRanges.size() < maxBatchSize) {
// Check producer errors even while draining a previously fetched
batch.
- BlockingQueue<Collection<TScanRangeLocations>> splits =
splitAssignment.getAssignedSplits(backend);
+ SplitAssignment.AssignmentQueue splits =
splitAssignment.getAssignedSplits(backend);
if (splitAssignment.isStop()) {
- pendingSplits = Collections.emptyIterator();
+ scanRanges.clear();
isLastBatch.set(true);
break;
}
- while (pendingSplits.hasNext() && scanRanges.size() <
maxBatchSize) {
- scanRanges.add(pendingSplits.next());
- }
- if (scanRanges.size() == maxBatchSize) {
- return FileScanRangeCompactor.compactSplitBatch(scanRanges);
- }
- pendingSplits = Collections.emptyIterator();
- // A finished, empty queue is EOF only after the pending batch has
been drained.
if (splits == null) {
isLastBatch.set(true);
break;
}
- // Each queued batch contains at most splitBatchSize ranges. Drain
only batches
- // that fit this response, bounding the local buffer and
amortizing the queue lock.
- int batchesToDrain = Math.min(64,
- (maxBatchSize - scanRanges.size()) /
splitAssignment.getSplitBatchSize());
- if (batchesToDrain > 1 && splits.size() > 1) {
- List<Collection<TScanRangeLocations>> readyBatches = new
ArrayList<>(batchesToDrain);
- splits.drainTo(readyBatches, batchesToDrain);
+ // Drain only the requested amount; the queue stores individual
splits, but takes the lock once.
+ int drained = splits.drainTo(scanRanges, Math.min(64, maxBatchSize
- scanRanges.size()));
+ if (drained > 0) {
splitAssignment.notifyQueueSpace(backend);
// Resuming a producer may publish an error or cancellation
before we return.
splitAssignment.getAssignedSplits(backend);
if (splitAssignment.isStop()) {
+ scanRanges.clear();
isLastBatch.set(true);
break;
}
- for (Collection<TScanRangeLocations> readyBatch :
readyBatches) {
- for (TScanRangeLocations range : readyBatch) {
- scanRanges.add(range);
- }
- }
- continue;
- }
- Collection<TScanRangeLocations> batch = splits.poll();
- if (batch != null) {
- splitAssignment.notifyQueueSpace(backend);
- pendingSplits = batch.iterator();
continue;
}
if (!scanRanges.isEmpty()) {
@@ -144,13 +127,31 @@ public class SplitSource {
}
long remainingNanos = waitNanos - (System.nanoTime() - startTime);
if (remainingNanos <= 0) {
- throw new UserException("Timeout. Max wait time(ms): " +
maxWaitTime);
+ // A temporary gap between producer batches is neither EOF nor
a query failure.
+ // Let the BE retry after this bounded long poll rather than
failing a slow listing.
+ SplitAssignment.AssignmentQueue pending =
splitAssignment.getAssignedSplits(backend);
+ if (splitAssignment.isStop()) {
+ isLastBatch.set(true);
+ break;
+ }
+ if (pending == null) {
+ isLastBatch.set(true);
+ break;
+ }
+ throw new SplitPendingException(maxWaitTime);
}
try {
- batch = splitAssignment.awaitBatch(backend, remainingNanos);
- if (batch != null) {
+ TScanRangeLocations split =
splitAssignment.awaitSplit(backend, remainingNanos);
+ if (split != null) {
splitAssignment.notifyQueueSpace(backend);
- pendingSplits = batch.iterator();
+ // The awakened producer may fail or stop before this
one-split fetch returns.
+ splitAssignment.getAssignedSplits(backend);
+ if (splitAssignment.isStop()) {
+ scanRanges.clear();
+ isLastBatch.set(true);
+ break;
+ }
+ scanRanges.add(split);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 069a18a4cc8..c2e8e606351 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -596,6 +596,8 @@ public class SessionVariable implements Serializable,
Writable {
public static final String NUM_PARTITIONS_IN_BATCH_MODE =
"num_partitions_in_batch_mode";
+ public static final String SPLIT_PARTITION_BATCH_SIZE =
"split_partition_batch_size";
+
public static final String NUM_FILES_IN_BATCH_MODE =
"num_files_in_batch_mode";
public static final String SPLIT_BATCH_SIZE = "split_batch_size";
@@ -2622,22 +2624,27 @@ public class SessionVariable implements Serializable,
Writable {
description = "If the number of partitions exceeds the threshold,
scan ranges will be got through batch "
+ "mode.",
needForward = true)
- public int numPartitionsInBatchMode = 1024;
+ public int numPartitionsInBatchMode = 1;
+
+ @VarAttrDef.VarAttr(name = SPLIT_PARTITION_BATCH_SIZE, checker =
"checkSplitPartitionBatchSize",
+ needForward = true, description = "Maximum number of Hive
partitions resolved per split producer "
+ + "batch. This does not control when batch mode is
enabled.")
+ public int splitPartitionBatchSize = 16;
@VarAttrDef.VarAttr(
name = NUM_FILES_IN_BATCH_MODE,
fuzzy = true,
description = "If the number of files exceeds the threshold, scan
ranges will be got through batch mode.",
needForward = true)
- public int numFilesInBatchMode = 1024;
+ public int numFilesInBatchMode = 1;
@VarAttrDef.VarAttr(name = SPLIT_BATCH_SIZE, checker =
"checkSplitBatchSize", needForward = true,
description = "Maximum number of splits in a producer batch.")
- public int splitBatchSize = 128;
+ public int splitBatchSize = 8;
@VarAttrDef.VarAttr(name = SPLIT_QUEUE_CAPACITY, checker =
"checkSplitQueueCapacity", needForward = true,
- description = "Maximum number of queued split batches per
backend.")
- public int splitQueueCapacity = 32;
+ description = "Maximum number of queued splits per backend.")
+ public int splitQueueCapacity = 1024;
@VarAttrDef.VarAttr(name = SPLIT_INITIAL_BATCH_SIZE, checker =
"checkSplitInitialBatchSize", needForward = true,
description = "Maximum number of splits returned by the first
fetch per backend; "
@@ -2652,16 +2659,26 @@ public class SessionVariable implements Serializable,
Writable {
checkFieldValue(SPLIT_BATCH_SIZE, 1, value);
}
+ public void checkSplitPartitionBatchSize(String value) throws Exception {
+ checkFieldValue(SPLIT_PARTITION_BATCH_SIZE, 1, value);
+ }
+
public void checkSplitQueueCapacity(String value) throws Exception {
checkFieldValue(SPLIT_QUEUE_CAPACITY, 1, value);
}
@VarAttrDef.VarAttr(
name = FETCH_SPLITS_MAX_WAIT_TIME,
- description = "The max wait time of getting splits in batch mode.",
+ checker = "checkFetchSplitsMaxWaitTime",
+ description = "Maximum wait time per split-fetch RPC in batch
mode. A running producer can be retried "
+ + "after this interval; completion, cancellation and
production errors are reported separately.",
needForward = true)
public long fetchSplitsMaxWaitTime = 1000;
+ public void checkFetchSplitsMaxWaitTime(String value) throws Exception {
+ checkFieldLongValue(FETCH_SPLITS_MAX_WAIT_TIME, 1, value);
+ }
+
@VarAttrDef.VarAttr(
name = ENABLE_PARQUET_LAZY_MAT,
fuzzy = true,
@@ -5018,6 +5035,10 @@ public class SessionVariable implements Serializable,
Writable {
return numPartitionsInBatchMode;
}
+ public int getSplitPartitionBatchSize() {
+ return splitPartitionBatchSize;
+ }
+
public int getNumFilesInBatchMode() {
return numFilesInBatchMode;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 5078dc05228..dffc27be335 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -1119,6 +1119,10 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
result.setSplits(locations);
result.status = new TStatus(TStatusCode.OK);
return result;
+ } catch (SplitSource.SplitPendingException e) {
+ // The backend will retry; an empty OK batch would incorrectly
mean EOF.
+ result.status = new TStatus(TStatusCode.INCOMPLETE);
+ return result;
} catch (Exception e) {
LOG.warn("failed to fetch split batch with source id {}",
request.getSplitSourceId(), e);
result.status = new TStatus(TStatusCode.INTERNAL_ERROR);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java
index e2bbaae189b..4db437c8a6f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeBatchModeTest.java
@@ -168,6 +168,7 @@ public class PluginDrivenScanNodeBatchModeTest {
// Non-count agg so countPushdown=false; sessionVariable needed by the
partition fallback arg eval.
Mockito.doReturn(TPushAggOp.NONE).when(node).getPushDownAggNoGroupingOp();
SessionVariable sv = Mockito.mock(SessionVariable.class);
+ Mockito.when(sv.getEnableExternalTableBatchMode()).thenReturn(true);
Mockito.when(sv.getNumPartitionsInBatchMode()).thenReturn(THRESHOLD);
Mockito.when(sv.getEnableExternalTableBatchMode()).thenReturn(true);
Deencapsulation.setField(node, "sessionVariable", sv);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeHiveStreamingTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeHiveStreamingTest.java
index 694d3916905..05a5a95e9c2 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeHiveStreamingTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeHiveStreamingTest.java
@@ -110,6 +110,21 @@ class PluginDrivenScanNodeHiveStreamingTest {
}
}
+ @Test
+ @Timeout(30)
+ void partitionBatchUpperBoundStillStartsIndependentListingSources() throws
Exception {
+ try (Fixture fixture = new Fixture(Collections.emptySet())) {
+ ConnectorSession session = Deencapsulation.getField(fixture.node,
"connectorSession");
+ Mockito.when(session.getProperty("split_partition_batch_size",
Integer.class)).thenReturn(16);
+ fixture.start();
+ Assertions.assertTrue(fixture.entered.await(10, TimeUnit.SECONDS));
+ Assertions.assertEquals(2, fixture.listings.get());
+ fixture.release.countDown();
+ Assertions.assertEquals(expectedRanges(Collections.emptySet()),
new HashSet<>(fixture.drain()));
+ Assertions.assertEquals(2, fixture.peakListings.get());
+ }
+ }
+
@Test
@Timeout(30)
void fileCountStreamingDrainsThroughTheSharedProducerFinalizer() throws
Exception {
@@ -240,11 +255,11 @@ class PluginDrivenScanNodeHiveStreamingTest {
Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE);
FairTaskExecutor.Group group = executor.newGroup(1);
ConnectorTaskScheduler scheduler =
Mockito.mock(ConnectorTaskScheduler.class, Mockito.CALLS_REAL_METHODS);
- Mockito.when(scheduler.tryExecute(Mockito.any(), Mockito.any(),
Mockito.any()))
- .thenAnswer(call -> group.tryExecute(call.getArgument(0),
call.getArgument(1), call.getArgument(2)));
- Mockito.when(scheduler.tryExecuteContinuation(Mockito.any(),
Mockito.any(), Mockito.any()))
- .thenAnswer(call -> group.tryExecuteContinuation(
- call.getArgument(0), call.getArgument(1),
call.getArgument(2)));
+ Mockito.doAnswer(call -> group.tryExecute(call.getArgument(0),
call.getArgument(1), call.getArgument(2)))
+ .when(scheduler).tryExecute(Mockito.any(), Mockito.any(),
Mockito.any());
+ Mockito.doAnswer(call -> group.tryExecuteContinuation(
+ call.getArgument(0), call.getArgument(1),
call.getArgument(2)))
+ .when(scheduler).tryExecuteContinuation(Mockito.any(),
Mockito.any(), Mockito.any());
Mockito.doAnswer(call -> {
group.removeSpaceListener(call.getArgument(0));
return null;
@@ -252,6 +267,7 @@ class PluginDrivenScanNodeHiveStreamingTest {
Mockito.when(session.getTaskScheduler(ConnectorTaskScheduler.Stage.SPLIT)).thenReturn(scheduler);
Mockito.when(session.getProperty("file_split_size",
String.class)).thenReturn("1");
Mockito.when(session.getProperty("num_partitions_in_batch_mode",
Integer.class)).thenReturn(2);
+ Mockito.when(session.getProperty("split_partition_batch_size",
Integer.class)).thenReturn(2);
// The old implementation chooses its single-source streaming path
for this same Hive scan.
Assertions.assertTrue(provider.streamingSplitEstimate(session,
handle, Optional.empty(), false) >= 0);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
index 36323a963f5..e6c044bc74e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodePartitionBatchTest.java
@@ -57,7 +57,6 @@ class PluginDrivenScanNodePartitionBatchTest {
fixture.start(100, 1, 2);
fixture.executor.runUntilIdle();
Assertions.assertEquals(2, fixture.loaded.size(), "Paused producers
must retain only two partition lists");
- Assertions.assertEquals(1,
fixture.assignment.getAssignedSplits(fixture.backend).size());
Assertions.assertEquals(0, fixture.finished.get());
List<Long> actual = fixture.drain();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitAssignmentTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitAssignmentTest.java
index 267737b2a4b..f50e86f2f66 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitAssignmentTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitAssignmentTest.java
@@ -31,12 +31,10 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -105,7 +103,7 @@ public class SplitAssignmentTest {
ExecutorService workers = Executors.newFixedThreadPool(3);
try {
Future<?> init = workers.submit(() -> {
- splitAssignment.init();
+ splitAssignment.init(1000);
return null;
});
Assertions.assertTrue(initStarted.await(5, TimeUnit.SECONDS));
@@ -180,7 +178,7 @@ public class SplitAssignmentTest {
// ==================== init() method tests ====================
@Test
- void testInitSuccess() throws Exception {
+ void testInitAcceptsDelayedFirstSplitWithinCallerDeadline() throws
Exception {
Multimap<Backend, Split> batch = ArrayListMultimap.create();
batch.put(mockBackend, mockSplit);
@@ -202,7 +200,7 @@ public class SplitAssignmentTest {
splitGeneratorThread.start();
// Test
- Assertions.assertDoesNotThrow(() -> splitAssignment.init());
+ Assertions.assertDoesNotThrow(() -> splitAssignment.init(1000));
// Verify sample split is set
Assertions.assertNotNull(splitAssignment.getSampleSplit());
@@ -216,7 +214,7 @@ public class SplitAssignmentTest {
try {
Future<UserException> result = executor.submit(() -> {
Thread.currentThread().interrupt();
- return Assertions.assertThrows(UserException.class,
splitAssignment::init);
+ return Assertions.assertThrows(UserException.class, () ->
splitAssignment.init(1000));
});
Assertions.assertInstanceOf(InterruptedException.class,
result.get(5, TimeUnit.SECONDS).getCause());
Mockito.verify(mockSplitGenerator).startSplit(mockBackendPolicy.numBackends());
@@ -226,13 +224,44 @@ public class SplitAssignmentTest {
}
}
+ @Test
+ void testInitUsesCallerDeadlineBeforeFirstSplit() throws Exception {
+ UserException error = Assertions.assertThrows(UserException.class, ()
-> splitAssignment.init(20));
+ Assertions.assertTrue(error.getMessage().contains("20 ms"),
error.getMessage());
+
Mockito.verify(mockSplitGenerator).startSplit(mockBackendPolicy.numBackends());
+ }
+
+ @Test
+ void testStopWakesFirstSplitPlanningWait() throws Exception {
+ CountDownLatch producerStarted = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ producerStarted.countDown();
+ return null;
+ }).when(mockSplitGenerator).startSplit(Mockito.anyInt());
+ ExecutorService worker = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> planning = worker.submit(() -> {
+ splitAssignment.init(10000);
+ return null;
+ });
+ Assertions.assertTrue(producerStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertThrows(TimeoutException.class, () ->
planning.get(50, TimeUnit.MILLISECONDS));
+ splitAssignment.stop();
+ planning.get(1, TimeUnit.SECONDS);
+ } finally {
+ splitAssignment.stop();
+ worker.shutdownNow();
+ Assertions.assertTrue(worker.awaitTermination(5,
TimeUnit.SECONDS));
+ }
+ }
+
@Test
void testInitWithPreExistingException() throws Exception {
UserException preException = new UserException("Pre-existing error");
splitAssignment.setException(preException);
// Test & Verify
- UserException exception = Assertions.assertThrows(UserException.class,
() -> splitAssignment.init());
+ UserException exception = Assertions.assertThrows(UserException.class,
() -> splitAssignment.init(1000));
Assertions.assertTrue(exception.getMessage().contains(" Pre-existing
error"), exception.getMessage());
}
@@ -263,17 +292,36 @@ public class SplitAssignmentTest {
Assertions.assertEquals(mockSplit, splitAssignment.getSampleSplit());
// Verify assignment queue is created and contains data
- BlockingQueue<Collection<TScanRangeLocations>> queue =
splitAssignment.getAssignedSplits(mockBackend);
+ SplitAssignment.AssignmentQueue queue =
splitAssignment.getAssignedSplits(mockBackend);
Assertions.assertNotNull(queue);
}
@Test
void testConsumerCreatesBoundedQueueFirst() throws Exception {
- BlockingQueue<Collection<TScanRangeLocations>> queue =
splitAssignment.getAssignedSplits(mockBackend);
+ SplitAssignment.AssignmentQueue queue =
splitAssignment.getAssignedSplits(mockBackend);
Assertions.assertEquals(32, queue.remainingCapacity());
Assertions.assertSame(queue,
splitAssignment.getAssignedSplits(mockBackend));
}
+ @Test
+ void queueCapacityBelowBatchSizeReducesProducerBatchSize() throws
Exception {
+ SplitAssignment smallQueueAssignment = new
SplitAssignment(mockBackendPolicy, mockSplitGenerator,
+ mockSplitToScanRange, locationProperties, pathPartitionKeys,
true, 32, 2);
+ Assertions.assertEquals(2, smallQueueAssignment.getSplitBatchSize());
+ Multimap<Backend, Split> batch = ArrayListMultimap.create();
+ batch.put(mockBackend, mockSplit);
+ batch.put(mockBackend, mockSplit);
+
Mockito.when(mockBackendPolicy.computeScanRangeAssignment(Mockito.any())).thenReturn(batch);
+ Mockito.when(mockSplitToScanRange.getScanRange(Mockito.any(),
Mockito.any(), Mockito.any(), Mockito.any(),
+ Mockito.anyBoolean())).thenReturn(mockScanRangeLocations);
+ smallQueueAssignment.addToQueue(List.of(mockSplit, mockSplit));
+ SplitAssignment.AssignmentQueue queue =
smallQueueAssignment.getAssignedSplits(mockBackend);
+ Assertions.assertEquals(2, queue.size());
+ Assertions.assertEquals(0, queue.remainingCapacity());
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> queue.offer(List.of(mockScanRangeLocations), 1));
+ }
+
@Test
void testSamplePublishedAfterScanParametersAreInitialized() throws
Exception {
Multimap<Backend, Split> batch = ArrayListMultimap.create();
@@ -315,20 +363,20 @@ public class SplitAssignmentTest {
splitAssignment.addToQueue(Collections.singletonList(mockSplit));
- BlockingQueue<Collection<TScanRangeLocations>> queue =
splitAssignment.getAssignedSplits(mockBackend);
+ SplitAssignment.AssignmentQueue queue =
splitAssignment.getAssignedSplits(mockBackend);
Assertions.assertEquals(1, queue.size());
Assertions.assertEquals(31, queue.remainingCapacity());
- Assertions.assertSame(mockScanRangeLocations,
queue.element().iterator().next());
+ Assertions.assertSame(mockScanRangeLocations, queue.peek());
}
@Test
void testConsumerCreatedQueueBlocksProducerUntilSpaceAvailable() throws
Exception {
- BlockingQueue<Collection<TScanRangeLocations>> queue =
splitAssignment.getAssignedSplits(mockBackend);
- Collection<TScanRangeLocations> queuedBatch =
Collections.singletonList(mockScanRangeLocations);
+ SplitAssignment.AssignmentQueue queue =
splitAssignment.getAssignedSplits(mockBackend);
+ List<TScanRangeLocations> queuedBatch =
Collections.singletonList(mockScanRangeLocations);
for (int i = 0; i < 32; i++) {
- Assertions.assertTrue(queue.offer(queuedBatch));
+ Assertions.assertEquals(1, queue.offer(queuedBatch, 0));
}
- Assertions.assertFalse(queue.offer(queuedBatch));
+ Assertions.assertEquals(0, queue.offer(queuedBatch, 0));
Multimap<Backend, Split> batch = ArrayListMultimap.create();
batch.put(mockBackend, mockSplit);
@@ -350,7 +398,7 @@ public class SplitAssignmentTest {
Assertions.assertThrows(TimeoutException.class, () ->
producer.get(200, TimeUnit.MILLISECONDS));
Assertions.assertEquals(32, queue.size());
- Assertions.assertSame(queuedBatch, queue.poll());
+ Assertions.assertSame(mockScanRangeLocations, queue.poll());
producer.get(5, TimeUnit.SECONDS);
Assertions.assertEquals(32, queue.size());
} finally {
@@ -446,13 +494,13 @@ public class SplitAssignmentTest {
splitsProvider.start();
// Test init - should succeed once splits are added
- Assertions.assertDoesNotThrow(() -> splitAssignment.init());
+ Assertions.assertDoesNotThrow(() -> splitAssignment.init(1000));
// Verify
Assertions.assertNotNull(splitAssignment.getSampleSplit());
Assertions.assertEquals(mockSplit, splitAssignment.getSampleSplit());
- BlockingQueue<Collection<TScanRangeLocations>> queue =
splitAssignment.getAssignedSplits(mockBackend);
+ SplitAssignment.AssignmentQueue queue =
splitAssignment.getAssignedSplits(mockBackend);
Assertions.assertNotNull(queue);
splitsProvider.join(1000);
@@ -464,7 +512,8 @@ public class SplitAssignmentTest {
splitAssignment.stop(); // This should make needMoreSplit() return
false
// Init should complete immediately without waiting
- Assertions.assertDoesNotThrow(() -> splitAssignment.init());
+ Assertions.assertDoesNotThrow(() -> splitAssignment.init(1000));
+ Mockito.verify(mockSplitGenerator,
Mockito.never()).startSplit(Mockito.anyInt());
}
@Test
@@ -473,6 +522,6 @@ public class SplitAssignmentTest {
splitAssignment.finishSchedule();
// Init should complete immediately without waiting
- Assertions.assertDoesNotThrow(() -> splitAssignment.init());
+ Assertions.assertDoesNotThrow(() -> splitAssignment.init(1000));
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
index 890b9075f2a..fde370bc637 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitProducerTest.java
@@ -105,6 +105,28 @@ class SplitProducerTest {
Assertions.assertEquals(1, fixture.completions.get());
}
+ @Test
+ void partiallyOfferedBatchResumesWithoutLosingOrRepeatingSplits() throws
Exception {
+ Fixture fixture = new Fixture(3, 3);
+ fixture.assignment.addToQueue(Arrays.asList(split(0), split(2)));
+ ManualExecutor executor = new ManualExecutor();
+ RecordingSource source = new RecordingSource(3, 4, 6, 8);
+ fixture.start(executor, source);
+ executor.runUntilIdle();
+
+ Assertions.assertEquals(3,
fixture.assignment.getAssignedSplits(fixture.backend).size());
+ Assertions.assertEquals(0, fixture.completions.get());
+ SplitSource consumer = fixture.source();
+ Assertions.assertEquals(Arrays.asList(0L, 2L, 4L),
identities(consumer.getNextBatch(3)));
+ executor.runUntilIdle();
+ Assertions.assertEquals(Arrays.asList(6L, 8L),
identities(consumer.getNextBatch(3)));
+ Assertions.assertTrue(consumer.getNextBatch(1).isEmpty());
+ Assertions.assertEquals(1, fixture.conversions.get(4L));
+ Assertions.assertEquals(1, fixture.conversions.get(6L));
+ Assertions.assertEquals(1, fixture.conversions.get(8L));
+ Assertions.assertEquals(1, fixture.completions.get());
+ }
+
@Test
void queryCloseReleasesAProducerWaitingForTaskQueueCapacity() throws
Exception {
Fixture fixture = new Fixture(1, 1);
@@ -231,8 +253,8 @@ class SplitProducerTest {
}).when(second.generator).startSplit(Mockito.anyInt());
try {
Future<?> initialized = planner.submit(() -> {
- first.assignment.init();
- second.assignment.init();
+ first.assignment.init(1000);
+ second.assignment.init(1000);
return null;
});
initialized.get(5, TimeUnit.SECONDS);
@@ -257,9 +279,9 @@ class SplitProducerTest {
@Test
void partialMultiBackendSubmissionDoesNotReplayOrReassignTheBatch() throws
Exception {
- Fixture fixture = new Fixture(2, 1);
+ Fixture fixture = new Fixture(2, 2);
ManualExecutor executor = new ManualExecutor();
- fixture.assignment.addToQueue(Collections.singletonList(split(1)));
+ fixture.assignment.addToQueue(Arrays.asList(split(1), split(5)));
RecordingSource source = new RecordingSource(2, 0, 3);
fixture.start(executor, source);
executor.runNext();
@@ -270,6 +292,7 @@ class SplitProducerTest {
SplitSource other = new SplitSource(fixture.otherBackend,
fixture.assignment, 1000);
Assertions.assertEquals(Collections.singletonList(1L),
identities(other.getNextBatch(1)));
executor.runUntilIdle();
+ Assertions.assertEquals(Collections.singletonList(5L),
identities(other.getNextBatch(1)));
Assertions.assertEquals(Collections.singletonList(3L),
identities(other.getNextBatch(1)));
Assertions.assertTrue(other.getNextBatch(1).isEmpty());
Assertions.assertTrue(fixture.source().getNextBatch(1).isEmpty());
@@ -287,12 +310,12 @@ class SplitProducerTest {
fixture.assignment.addToQueue(Collections.singletonList(split(0)));
SplitSource consumer = fixture.source();
Mockito.doAnswer(call -> {
- boolean offered = (boolean) call.callRealMethod();
- if (!offered) {
+ int accepted = (int) call.callRealMethod();
+ if (accepted == 0) {
Assertions.assertEquals(Collections.singletonList(0L),
identities(consumer.getNextBatch(1)));
}
- return offered;
- }).when(fixture.assignment).offerBatch(Mockito.any(), Mockito.any(),
Mockito.any());
+ return accepted;
+ }).when(fixture.assignment).offerBatch(Mockito.any(), Mockito.any(),
Mockito.anyInt(), Mockito.any());
ManualExecutor executor = new ManualExecutor();
RecordingSource source = new RecordingSource(1, 2);
fixture.start(executor, source);
@@ -438,12 +461,12 @@ class SplitProducerTest {
if (beforeOffer) {
fixture.assignment.setException(expected);
}
- boolean offered = (boolean) call.callRealMethod();
+ int accepted = (int) call.callRealMethod();
if (!beforeOffer) {
fixture.assignment.setException(expected);
}
- return offered;
- }).when(fixture.assignment).offerBatch(Mockito.any(),
Mockito.any(), Mockito.any());
+ return accepted;
+ }).when(fixture.assignment).offerBatch(Mockito.any(),
Mockito.any(), Mockito.anyInt(), Mockito.any());
ManualExecutor executor = new ManualExecutor();
RecordingSource source = new RecordingSource(1, 2);
fixture.start(executor, source);
@@ -506,10 +529,10 @@ class SplitProducerTest {
Fixture fixture = new Fixture(1, 100);
fixture.assignment = Mockito.spy(fixture.assignment);
Mockito.doAnswer(call -> {
- boolean result = (boolean) call.callRealMethod();
+ int result = (int) call.callRealMethod();
fixture.assignment.stop();
return result;
- }).when(fixture.assignment).offerBatch(Mockito.any(), Mockito.any(),
Mockito.any());
+ }).when(fixture.assignment).offerBatch(Mockito.any(), Mockito.any(),
Mockito.anyInt(), Mockito.any());
ManualExecutor executor = new ManualExecutor();
RecordingSource source = new RecordingSource(1, 0, 2, 4);
fixture.start(executor, source);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
index 5ed216e9b6a..9b089eaf2ec 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitSourceTest.java
@@ -45,12 +45,10 @@ import java.lang.reflect.Field;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Set;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -95,7 +93,7 @@ public class SplitSourceTest {
}
assignment = new SplitAssignment(policy,
Mockito.mock(SplitGenerator.class),
(be, properties, split, keys, admission) ->
locations.get(split),
- Collections.emptyMap(), Collections.emptyList(), true, 3, 3);
+ Collections.emptyMap(), Collections.emptyList(), true, 3, 8);
}
private Object capacityMonitor() throws Exception {
@@ -112,8 +110,8 @@ public class SplitSourceTest {
synchronized (capacityMonitor()) {
Future<?> task = executor.submit(() -> {
if (produce) {
- Assertions.assertTrue(assignment.offerBatch(backend,
- Collections.singletonList(ranges.get(0)), ()
-> {}));
+ Assertions.assertEquals(1,
assignment.offerBatch(backend,
+ Collections.singletonList(ranges.get(0)), 0,
() -> {}));
} else {
assignment.notifyQueueSpace(backend);
}
@@ -127,16 +125,16 @@ public class SplitSourceTest {
@Test
void capacityReleasedBeforeWaiterRegistrationIsRetried() throws Exception {
- BlockingQueue<Collection<TScanRangeLocations>> queue =
assignment.getAssignedSplits(backend);
+ SplitAssignment.AssignmentQueue queue =
assignment.getAssignedSplits(backend);
while (queue.remainingCapacity() > 0) {
- queue.add(Collections.singletonList(ranges.get(0)));
+ Assertions.assertEquals(1,
queue.offer(Collections.singletonList(ranges.get(0)), 0));
}
AtomicInteger resumed = new AtomicInteger();
AtomicReference<Throwable> failure = new AtomicReference<>();
- AtomicReference<Boolean> offered = new AtomicReference<>();
+ AtomicReference<Integer> offered = new AtomicReference<>();
Thread producer = new Thread(() -> {
try {
- offered.set(assignment.offerBatch(backend,
Collections.singletonList(ranges.get(1)),
+ offered.set(assignment.offerBatch(backend,
Collections.singletonList(ranges.get(1)), 0,
resumed::incrementAndGet));
} catch (Throwable t) {
failure.set(t);
@@ -157,25 +155,46 @@ public class SplitSourceTest {
producer.join(5000);
Assertions.assertFalse(producer.isAlive());
Assertions.assertNull(failure.get());
- Assertions.assertEquals(Boolean.TRUE, offered.get());
+ Assertions.assertEquals(1, offered.get());
Assertions.assertEquals(0, queue.remainingCapacity());
assignment.notifyQueueSpace(backend);
Assertions.assertEquals(0, resumed.get(), "the successful retry must
remove its capacity callback");
}
+ @Test
+ void partialOfferFillsRemainingCapacity() throws Exception {
+ SplitAssignment.AssignmentQueue queue =
assignment.getAssignedSplits(backend);
+ AtomicInteger resumed = new AtomicInteger();
+ Runnable callback = resumed::incrementAndGet;
+ Assertions.assertEquals(3, assignment.offerBatch(backend,
ranges.subList(0, 3), 0, callback));
+ Assertions.assertEquals(3, assignment.offerBatch(backend,
ranges.subList(3, 6), 0, callback));
+ Assertions.assertEquals(6, queue.size());
+ Assertions.assertEquals(2, queue.remainingCapacity());
+ Assertions.assertEquals(2, assignment.offerBatch(backend,
ranges.subList(0, 3), 0, callback));
+ Assertions.assertEquals(0, queue.remainingCapacity());
+ Assertions.assertEquals(0, assignment.offerBatch(backend,
ranges.subList(0, 3), 2, callback));
+ Assertions.assertEquals(ranges.get(0), queue.poll());
+ assignment.notifyQueueSpace(backend);
+ Assertions.assertEquals(1, resumed.get());
+ Assertions.assertEquals(1, assignment.offerBatch(backend,
ranges.subList(0, 3), 2, callback));
+ Assertions.assertEquals(0, queue.remainingCapacity());
+ }
+
@Test
void backendChannelsUseStableIdsAcrossEndpointChanges() throws Exception {
Backend first = new Backend(11L, "127.0.0.1", 9050);
Backend second = new Backend(12L, "127.0.0.2", 9050);
- BlockingQueue<Collection<TScanRangeLocations>> firstQueue =
assignment.getAssignedSplits(first);
- BlockingQueue<Collection<TScanRangeLocations>> secondQueue =
assignment.getAssignedSplits(second);
+ SplitAssignment.AssignmentQueue firstQueue =
assignment.getAssignedSplits(first);
+ SplitAssignment.AssignmentQueue secondQueue =
assignment.getAssignedSplits(second);
AtomicInteger resumed = new AtomicInteger();
Runnable callback = resumed::incrementAndGet;
int capacity = firstQueue.remainingCapacity();
for (int i = 0; i < capacity; i++) {
- Assertions.assertTrue(assignment.offerBatch(first,
Collections.singletonList(ranges.get(0)), callback));
+ Assertions.assertEquals(1, assignment.offerBatch(first,
+ Collections.singletonList(ranges.get(0)), 0, callback));
}
- Assertions.assertFalse(assignment.offerBatch(first,
Collections.singletonList(ranges.get(0)), callback));
+ Assertions.assertEquals(0, assignment.offerBatch(first,
+ Collections.singletonList(ranges.get(0)), 0, callback));
first.setAlive(true);
first.setBePort(9061);
Backend sameId = new Backend(11L, "127.0.0.3", 9150);
@@ -196,13 +215,13 @@ public class SplitSourceTest {
void bulkDrainRespectsRequestAndResumesBackpressuredProducer() throws
Exception {
assignment.addToQueue(splits);
AtomicInteger resumed = new AtomicInteger();
- Assertions.assertFalse(assignment.offerBatch(backend,
Collections.singletonList(ranges.get(0)),
+ Assertions.assertEquals(0, assignment.offerBatch(backend,
Collections.singletonList(ranges.get(0)), 0,
resumed::incrementAndGet));
SplitSource source = new SplitSource(backend, assignment, 1000);
Assertions.assertEquals(ranges.subList(0, 6), source.getNextBatch(6));
Assertions.assertEquals(1, resumed.get());
- Assertions.assertEquals(1,
assignment.getAssignedSplits(backend).size(),
- "batches beyond the BE request must stay in the bounded
queue");
+ Assertions.assertEquals(2,
assignment.getAssignedSplits(backend).size(),
+ "splits beyond the BE request must stay in the bounded queue");
assignment.finishSchedule();
Assertions.assertEquals(ranges.subList(6, 8), source.getNextBatch(6));
Assertions.assertTrue(source.getNextBatch(6).isEmpty());
@@ -212,7 +231,7 @@ public class SplitSourceTest {
@ValueSource(booleans = {false, true})
void bulkDrainObservesTerminalStatePublishedByCapacityCallback(boolean
fail) throws Exception {
assignment.addToQueue(splits);
- Assertions.assertFalse(assignment.offerBatch(backend,
Collections.singletonList(ranges.get(0)), () -> {
+ Assertions.assertEquals(0, assignment.offerBatch(backend,
Collections.singletonList(ranges.get(0)), 0, () -> {
if (fail) {
assignment.setException(new UserException("bulk drain
failure"));
} else {
@@ -229,6 +248,42 @@ public class SplitSourceTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void awaitedSingleSplitObservesTerminalStateBeforeReturning(boolean fail)
throws Exception {
+ assignment = Mockito.spy(assignment);
+ CountDownLatch waiting = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ waiting.countDown();
+ TScanRangeLocations split = (TScanRangeLocations)
invocation.callRealMethod();
+ if (fail) {
+ assignment.setException(new UserException("awaited split
failure"));
+ } else {
+ assignment.stop();
+ }
+ return split;
+ }).when(assignment).awaitSplit(Mockito.eq(backend), Mockito.anyLong());
+ SplitSource source = new SplitSource(backend, assignment, 1000, 1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<List<TScanRangeLocations>> fetch = executor.submit(() ->
source.getNextBatch(1));
+ Assertions.assertTrue(waiting.await(5, TimeUnit.SECONDS));
+ Assertions.assertEquals(1, assignment.offerBatch(backend,
+ Collections.singletonList(ranges.get(0)), 0, () -> {}));
+ if (fail) {
+ ExecutionException error =
Assertions.assertThrows(ExecutionException.class,
+ () -> fetch.get(5, TimeUnit.SECONDS));
+ Assertions.assertInstanceOf(UserException.class,
error.getCause());
+
Assertions.assertTrue(error.getCause().getMessage().contains("awaited split
failure"));
+ } else {
+ Assertions.assertTrue(fetch.get(5,
TimeUnit.SECONDS).isEmpty());
+ Assertions.assertTrue(source.getNextBatch(1).isEmpty());
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Test
void initialFetchLimitIsIndependentOfProductionAndSubsequentFetches()
throws Exception {
// All eight splits are already produced without a BE request,
including a partially drained batch.
@@ -280,7 +335,7 @@ public class SplitSourceTest {
}
@Test
- void drainPendingBatchBeforeFinishedQueueBecomesEof() throws Exception {
+ void finishedQueueDrainsRemainingSplitsBeforeEof() throws Exception {
assignment.addToQueue(splits);
assignment.finishSchedule();
SplitSource source = new SplitSource(backend, assignment, 1000);
@@ -298,12 +353,13 @@ public class SplitSourceTest {
@Test
void producerBatchesAndConsumerCreatedQueueRespectConfiguredLimits()
throws Exception {
- BlockingQueue<Collection<TScanRangeLocations>> queue =
assignment.getAssignedSplits(backend);
- Assertions.assertEquals(3, queue.remainingCapacity());
+ SplitAssignment.AssignmentQueue queue =
assignment.getAssignedSplits(backend);
+ Assertions.assertEquals(8, queue.remainingCapacity());
assignment.addToQueue(splits);
- Assertions.assertEquals(3, queue.size());
- Assertions.assertEquals(Arrays.asList(3, 3, 2),
-
queue.stream().map(Collection::size).collect(java.util.stream.Collectors.toList()));
+ Assertions.assertEquals(8, queue.size());
+ Assertions.assertEquals(ranges, Arrays.asList(queue.poll(),
queue.poll(), queue.poll(), queue.poll(),
+ queue.poll(), queue.poll(), queue.poll(), queue.poll()));
+ Assertions.assertEquals(8, queue.remainingCapacity());
}
@Test
@@ -322,7 +378,7 @@ public class SplitSourceTest {
SplitSource source = new SplitSource(backend, assignment, 1000);
Assertions.assertEquals(ranges.subList(0, 1),
source.getNextBatch(1));
producer.get(5, TimeUnit.SECONDS);
- Assertions.assertEquals(3,
assignment.getAssignedSplits(backend).size());
+ Assertions.assertEquals(8,
assignment.getAssignedSplits(backend).size());
} finally {
executor.shutdownNow();
Assertions.assertTrue(executor.awaitTermination(5,
TimeUnit.SECONDS));
@@ -338,10 +394,59 @@ public class SplitSourceTest {
}
@Test
- void timeoutIsAnErrorRatherThanEof() {
+ void waitWindowExpiryIsPendingRatherThanEof() {
SplitSource source = new SplitSource(backend, assignment, 20);
- UserException failure = Assertions.assertThrows(UserException.class,
() -> source.getNextBatch(1));
+ UserException failure =
Assertions.assertThrows(SplitSource.SplitPendingException.class,
+ () -> source.getNextBatch(1));
Assertions.assertTrue(failure.getMessage().contains("Timeout"));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
SplitSource(backend, assignment, 0));
+ SessionVariable variables = new SessionVariable();
+ Assertions.assertThrows(Exception.class, () ->
variables.checkFetchSplitsMaxWaitTime("0"));
+ }
+
+ @Test
+ void unfinishedListingAfterFirstSplitIsPendingNotEof() throws Exception {
+ SplitSource source = new SplitSource(backend, assignment, 20, 1);
+ assignment.addToQueue(splits.subList(0, 1));
+ Assertions.assertEquals(ranges.subList(0, 1), source.getNextBatch(1));
+
+ // Another partition is still listing beyond one fetch wait window.
+ Assertions.assertThrows(SplitSource.SplitPendingException.class, () ->
source.getNextBatch(1));
+ assignment.addToQueue(splits.subList(1, 2));
+ assignment.finishSchedule();
+ Assertions.assertEquals(ranges.subList(1, 2), source.getNextBatch(1));
+ Assertions.assertTrue(source.getNextBatch(1).isEmpty());
+ }
+
+ @Test
+ void completedListingReturnsEof() throws Exception {
+ SplitSource source = new SplitSource(backend, assignment, 20);
+ assignment.finishSchedule();
+ Assertions.assertTrue(source.getNextBatch(1).isEmpty());
+ }
+
+ @Test
+ void backendWithoutInitialAssignmentCanRetryUntilListingFinishes() throws
Exception {
+ Backend otherBackend = new Backend(2L, "127.0.0.2", 9050);
+ SplitSource first = new SplitSource(backend, assignment, 20, 1);
+ SplitSource second = new SplitSource(otherBackend, assignment, 20, 1);
+ assignment.addToQueue(splits.subList(0, 1));
+ Assertions.assertEquals(ranges.subList(0, 1), first.getNextBatch(1));
+
+ Assertions.assertThrows(SplitSource.SplitPendingException.class, () ->
second.getNextBatch(1));
+ Assertions.assertEquals(1, assignment.offerBatch(otherBackend,
ranges.subList(1, 2), 0, () -> {}));
+ assignment.finishSchedule();
+ Assertions.assertEquals(ranges.subList(1, 2), second.getNextBatch(1));
+ Assertions.assertTrue(second.getNextBatch(1).isEmpty());
+ }
+
+ @Test
+ void producerFailureAfterPendingFetchIsStillAnError() throws Exception {
+ SplitSource source = new SplitSource(backend, assignment, 20);
+ Assertions.assertThrows(SplitSource.SplitPendingException.class, () ->
source.getNextBatch(1));
+ UserException failure = new UserException("listing failed");
+ assignment.setException(failure);
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class, () -> source.getNextBatch(1)));
}
@Test
@@ -468,7 +573,7 @@ public class SplitSourceTest {
Assertions.assertTrue(started.await(5, TimeUnit.SECONDS));
awaitConsumerWait(firstThread.get());
awaitConsumerWait(secondThread.get());
- Assertions.assertTrue(assignment.offerBatch(backend,
ranges.subList(0, 1), () -> {}));
+ Assertions.assertEquals(1, assignment.offerBatch(backend,
ranges.subList(0, 1), 0, () -> {}));
Assertions.assertEquals(ranges.subList(0, 1), firstResult.get(5,
TimeUnit.SECONDS));
Assertions.assertFalse(secondResult.isDone());
assignment.finishSchedule();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitStreamingConcurrencyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitStreamingConcurrencyTest.java
index f51636717d6..09846423de5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitStreamingConcurrencyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/split/SplitStreamingConcurrencyTest.java
@@ -36,12 +36,10 @@ import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -61,7 +59,7 @@ class SplitStreamingConcurrencyTest {
private void verifyConcurrentStreaming(boolean consumerFirst) throws
Exception {
int producerBatchSize = 7;
- int queueCapacity = 2;
+ int queueCapacity = 5;
int splitCount = 10003;
int[] requestedSizes = {1, 10, 16};
List<Backend> backends = new ArrayList<>();
@@ -81,8 +79,8 @@ class SplitStreamingConcurrencyTest {
CountDownLatch queuesFilled = new CountDownLatch(1);
SplitAssignment assignment = new SplitAssignment(policy,
Mockito.mock(SplitGenerator.class),
(backend, properties, split, keys, admission) -> {
- // Reaching batch three proves all queues received the
first two batches.
- if (split.getStart() >= producerBatchSize * queueCapacity)
{
+ // The producer has started filling the per-backend queues
before consumers run.
+ if (split.getStart() >= 10) {
queuesFilled.countDown();
}
// Preserve a unique split identity through assignment and
residual batches.
@@ -98,7 +96,7 @@ class SplitStreamingConcurrencyTest {
for (int i = 0; i < splitCount; i++) {
splits.add(new FileSplit(path, i, 1, splitCount, 0, null,
Collections.emptyList()));
}
- List<BlockingQueue<Collection<TScanRangeLocations>>> queues = new
ArrayList<>();
+ List<SplitAssignment.AssignmentQueue> queues = new ArrayList<>();
if (consumerFirst) {
for (Backend backend : backends) {
queues.add(assignment.getAssignedSplits(backend));
@@ -117,10 +115,8 @@ class SplitStreamingConcurrencyTest {
queues.add(assignment.getAssignedSplits(backend));
}
}
- for (BlockingQueue<Collection<TScanRangeLocations>> queue :
queues) {
- Assertions.assertEquals(queueCapacity, queue.size());
- Assertions.assertEquals(0, queue.remainingCapacity());
- verifyQueue(queue, producerBatchSize, queueCapacity);
+ for (SplitAssignment.AssignmentQueue queue : queues) {
+ verifyQueue(queue, queueCapacity);
}
List<Future<Set<Long>>> consumers = new ArrayList<>();
for (int i = 0; i < backends.size(); i++) {
@@ -129,7 +125,7 @@ class SplitStreamingConcurrencyTest {
consumers.add(executor.submit(() -> {
Set<Long> identities = new HashSet<>();
while (true) {
- verifyQueue(queues.get(backendIndex),
producerBatchSize, queueCapacity);
+ verifyQueue(queues.get(backendIndex), queueCapacity);
List<TScanRangeLocations> batch =
source.getNextBatch(requestedSizes[backendIndex]);
Assertions.assertTrue(batch.size() <=
requestedSizes[backendIndex]);
if (batch.isEmpty()) {
@@ -163,11 +159,9 @@ class SplitStreamingConcurrencyTest {
}
}
- private void verifyQueue(BlockingQueue<Collection<TScanRangeLocations>>
queue,
- int producerBatchSize, int queueCapacity) {
+ private void verifyQueue(SplitAssignment.AssignmentQueue queue, int
queueCapacity) {
+ Assertions.assertTrue(queue.remainingCapacity() >= 0);
+ Assertions.assertTrue(queue.remainingCapacity() <= queueCapacity);
Assertions.assertTrue(queue.size() <= queueCapacity);
- for (Collection<TScanRangeLocations> batch : queue) {
- Assertions.assertTrue(batch.size() <= producerBatchSize);
- }
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index 4c496334771..223a525d771 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -257,6 +257,10 @@ public class SessionVariablesTest extends
TestWithFeService {
public void testExternalTableBatchModeDefaultsAndFuzzyAttribute() throws
Exception {
SessionVariable sessionVar = new SessionVariable();
Assertions.assertTrue(sessionVar.getEnableExternalTableBatchMode());
+ Assertions.assertEquals(1, sessionVar.getNumPartitionsInBatchMode());
+ Assertions.assertEquals(1, sessionVar.getNumFilesInBatchMode());
+ Assertions.assertEquals(8, sessionVar.getSplitBatchSize());
+ Assertions.assertEquals(1024, sessionVar.getSplitQueueCapacity());
Field field =
SessionVariable.class.getDeclaredField("enableExternalTableBatchMode");
VarAttrDef.VarAttr varAttr =
field.getAnnotation(VarAttrDef.VarAttr.class);
@@ -481,7 +485,10 @@ public class SessionVariablesTest extends
TestWithFeService {
@Test
public void testSplitBatchAndQueueLimitsUseTheirOwnVariableCheckers()
throws Exception {
SessionVariable variable = new SessionVariable();
- for (String name : new String[] {SessionVariable.SPLIT_BATCH_SIZE,
SessionVariable.SPLIT_QUEUE_CAPACITY}) {
+ Assertions.assertEquals(16, variable.getSplitPartitionBatchSize());
+ Assertions.assertEquals(1024, variable.getSplitQueueCapacity());
+ for (String name : new String[] {SessionVariable.SPLIT_BATCH_SIZE,
+ SessionVariable.SPLIT_PARTITION_BATCH_SIZE,
SessionVariable.SPLIT_QUEUE_CAPACITY}) {
Field field = VariableMgr.getVarContext(name).getField();
for (int value : new int[] {1, 9}) {
VariableMgr.setVar(variable, new SetVar(SetType.SESSION, name,
new IntLiteral(value)));
diff --git
a/regression-test/suites/external_table_p0/hive/test_hive_streaming_splits.groovy
b/regression-test/suites/external_table_p0/hive/test_hive_streaming_splits.groovy
index 6584101f101..d0b59c08cfd 100644
---
a/regression-test/suites/external_table_p0/hive/test_hive_streaming_splits.groovy
+++
b/regression-test/suites/external_table_p0/hive/test_hive_streaming_splits.groovy
@@ -31,7 +31,8 @@ suite("test_hive_streaming_splits", "p0,external") {
'type' = 'hms', 'hive.metastore.uris' =
'thrift://${host}:${port}')"""
}
def settings = ["enable_external_table_batch_mode",
"num_files_in_batch_mode",
- "num_partitions_in_batch_mode", "split_batch_size",
"split_queue_capacity",
+ "num_partitions_in_batch_mode",
"split_partition_batch_size",
+ "split_batch_size", "split_queue_capacity",
"split_initial_batch_size"]
def saved = settings.collectEntries { name -> [(name): sql("show variables
like '${name}'")[0][1]] }
try {
@@ -57,6 +58,7 @@ suite("test_hive_streaming_splits", "p0,external") {
sql "set enable_external_table_batch_mode=true"
sql "set num_files_in_batch_mode=1"
sql "set num_partitions_in_batch_mode=1"
+ sql "set split_partition_batch_size=2"
qt_empty_unpartitioned "select count(*), sum(id) from
streaming_unpartitioned"
qt_empty_partitioned "select count(*), sum(id) from
streaming_partitioned"
// Separate inserts create enough files to cross several small
producer batches.
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
index 4c96e96b455..2e9189dd9b9 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_deletion_vector.groovy
@@ -624,19 +624,19 @@ s3.path-style-access=true
// Cross-reader comparison protects against Doris accepting a
DV/equality-delete result that
// diverges from Spark, Trino, or Iceberg's Java reader.
explain {
- sql("SELECT count(*) FROM dv_test;")
+ sql("SELECT /*+ SET_VAR(enable_external_table_batch_mode=false) */
count(*) FROM dv_test;")
verbose(true)
contains "deleteFileNum"
}
explain {
- sql("SELECT count(*) FROM dv_test_orc;")
+ sql("SELECT /*+ SET_VAR(enable_external_table_batch_mode=false) */
count(*) FROM dv_test_orc;")
verbose(true)
contains "deleteFileNum"
}
explain {
- sql("SELECT count(*) FROM dv_test_v2;")
+ sql("SELECT /*+ SET_VAR(enable_external_table_batch_mode=false) */
count(*) FROM dv_test_v2;")
verbose(true)
contains "deleteFileNum"
}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_delete.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_delete.groovy
index 381f38439b6..d1fe13af79c 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_delete.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_position_delete.groovy
@@ -170,6 +170,13 @@ suite("test_iceberg_position_delete", "p0,external") {
explain {
sql ("select name from iceberg_position_gen_data where id !=
5;")
verbose (true)
+ contains "(approximate)inputSplitNum"
+ }
+ explain {
+ // Full per-backend delete-file counts require complete,
synchronous split planning.
+ sql ("select /*+
SET_VAR(enable_external_table_batch_mode=false) */ name "
+ + "from iceberg_position_gen_data where id != 5;")
+ verbose (true)
contains "deleteFileNum"
}
}
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector_oss.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector_oss.groovy
index 71d8781b244..4c3e9469e57 100644
---
a/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector_oss.groovy
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_deletion_vector_oss.groovy
@@ -55,7 +55,7 @@ suite("test_paimon_deletion_vector_oss", "p0,external") {
def test_explain_verbose = {
explain {
- sql ("select * from deletion_vector_orc;")
+ sql ("select /*+
SET_VAR(enable_external_table_batch_mode=false) */ * from deletion_vector_orc;")
verbose (true)
contains "deleteFileNum"
}
@@ -69,4 +69,4 @@ suite("test_paimon_deletion_vector_oss", "p0,external") {
sql """set force_jni_scanner=false"""
}
-}
\ No newline at end of file
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
