Skip to content

branch-4.2: [improvement](datasource) Bound and fairly schedule external scan planning (#69298) #69986

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 4 commits into from
Nov 4, 2026
Merged

branch-4.2: [improvement](datasource) Bound and fairly schedule external scan planning (#69298) #69986

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b52f1af
[improvement](datasource) Bound and fairly schedule external scan pla…
924060929 Nov 3, 2026
152e4c0
[test](fe) Exercise streaming delivery in Hive split estimate tests
924060929 Nov 3, 2026
e651f20
[test](regression) Use normal explain for cold Hive streaming planning
924060929 Nov 3, 2026
e878673
[fix](fe) Export fair external scan scheduler metrics
924060929 Nov 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2472,6 +2472,17 @@ public class Config extends ConfigBase {
@ConfField(mutable = false, masterOnly = false)
public static int max_external_cache_loader_thread_pool_size = 64;

@ConfField(callback = PositiveIntConfHandler.class, description = "Positive worker count for fair split production")
public static int external_scan_split_threads = 64;

@github-actions github-actions Bot Nov 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Validate these global scheduler values as positive at configuration load. A non-positive external_scan_split_threads or external_scan_metadata_threads is accepted by ConfigBase and then makes ExternalMetaCacheMgr construction throw in FairTaskExecutor; external_scan_task_queue_capacity <= 0 reaches newGroup() and fails the first external scan. Add startup range validation (and a regression for invalid fe.conf values) so a typo is rejected at the configuration boundary.

@924060929 924060929 Nov 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 663c74d. All three settings opt into a pure positive-integer config validator during both ordinary and custom startup-file loading. Zero/negative values are rejected with the config name before that invalid field is updated. The existing empty-entry default rule and runtime immutability are preserved; unrelated runtime callbacks with side effects are still not dispatched during startup. Scheduler constructor/group preconditions remain for direct programmatic callers. Config tests cover both startup paths, 0/-1, valid positive/whitespace values, previous-value preservation and runtime update rejection. All 147 related FE tests and full FE packaging/Checkstyle pass locally.

@ConfField(callback = PositiveIntConfHandler.class,
description = "Positive worker count for independent metadata tasks")
public static int external_scan_metadata_threads = 64;

@ConfField(callback = PositiveIntConfHandler.class,
description = "Positive maximum queued tasks per query in each external scan scheduler")
public static int external_scan_task_queue_capacity = 128;

/**
* Max cache num of external catalog's file
* Decrease this value if FE's memory is small
Expand Down
20 changes: 19 additions & 1 deletion fe/fe-common/src/main/java/org/apache/doris/common/ConfigBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,21 @@ private static String checkedOption(Field field, String confVal) throws ConfigEx
}
}

public static class PositiveIntConfHandler implements ConfHandler {
@Override
public void handle(Field field, String confVal) throws Exception {
setConfigField(field, checkedValue(field, confVal));
}

private static String checkedValue(Field field, String confVal) throws ConfigException {
int value = Integer.parseInt(confVal.trim());
if (value <= 0) {
throw new ConfigException("Config '" + field.getName() + "' must be greater than 0");
}
return Integer.toString(value);
}
}

static class CommaSeparatedIntersectConfHandler implements ConfHandler {
@Override
public void handle(Field field, String newVal) throws Exception {
Expand Down Expand Up @@ -296,13 +311,16 @@ private static void setFields(Properties props, boolean isLdapConfig) throws Exc
}

try {
// Runtime callbacks can have side effects; only options validation is safe here.
// Runtime callbacks can have side effects; only pure value validators run here.
if (anno.callback() == OptionsConfHandler.class) {
confVal = OptionsConfHandler.checkedOption(f, confVal);
}
if (confVal.isEmpty()) {
continue;
}
if (anno.callback() == PositiveIntConfHandler.class) {
confVal = PositiveIntConfHandler.checkedValue(f, confVal);
}
setConfigField(f, confVal);
} catch (Exception e) {
String msg = String.format("Failed to set config, name: %s, value: %s", f.getName(), confVal);
Expand Down
63 changes: 63 additions & 0 deletions fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,17 @@
import org.junit.Test;
import org.junit.jupiter.api.Assertions;

import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;

public class ConfigTest {
private static final String[] EXTERNAL_SCAN_SCHEDULER_CONFIGS = {
"external_scan_split_threads", "external_scan_metadata_threads", "external_scan_task_queue_capacity"
};

@BeforeClass
public static void setUp() throws Exception {
Config config = new Config();
Expand Down Expand Up @@ -153,6 +158,64 @@ public void testOptionsConfigIsValidatedInCustomStartupFile() throws Exception {
}
}

@Test
public void testExternalScanSchedulerRejectsNonPositiveStartupValues() throws Exception {
assertExternalScanSchedulerRejectsNonPositiveValues(false);
}

@Test
public void testExternalScanSchedulerRejectsNonPositiveCustomStartupValues() throws Exception {
assertExternalScanSchedulerRejectsNonPositiveValues(true);
}

private void assertExternalScanSchedulerRejectsNonPositiveValues(boolean custom) throws Exception {
Config config = new Config();
config.init(configFile("").toString());
for (String name : EXTERNAL_SCAN_SCHEDULER_CONFIGS) {
Field field = Config.class.getField(name);
int original = field.getInt(null);
try {
for (String value : new String[] {"0", "-1"}) {
Path invalid = configFile(name + "=" + value + "\n");
IllegalArgumentException error = Assertions.assertThrows(IllegalArgumentException.class, () -> {
if (custom) {
config.initCustom(invalid.toString());
} else {
config.init(invalid.toString());
}
});
Assertions.assertTrue(error.getMessage().contains(name));
Assertions.assertInstanceOf(ConfigException.class, error.getCause());
Assertions.assertTrue(error.getCause().getMessage().contains("must be greater than 0"));
Assertions.assertEquals(original, field.getInt(null));
}
} finally {
field.setInt(null, original);
}
}
}

@Test
public void testExternalScanSchedulerAcceptsPositiveStartupValuesAndStaysImmutable() throws Exception {
Config config = new Config();
for (String name : EXTERNAL_SCAN_SCHEDULER_CONFIGS) {
Field field = Config.class.getField(name);
int original = field.getInt(null);
try {
config.init(configFile(name + "=1\n").toString());
Assertions.assertEquals(1, field.getInt(null));
config.initCustom(configFile(name + "= 128 \n").toString());
Assertions.assertEquals(128, field.getInt(null));
config.init(configFile(name + "=\n").toString());
Assertions.assertEquals(128, field.getInt(null));
Assertions.assertThrows(ConfigException.class, () -> ConfigBase.setMutableConfig(name, "8"));
Assertions.assertEquals(128, field.getInt(null));
} finally {
field.setInt(null, original);
}
}
}

private static Path configFile(String contents) throws Exception {
Path file = Files.createTempFile("fe_ut_options_", ".conf");
file.toFile().deleteOnExit();
Expand Down
26 changes: 26 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/common/ThreadPoolManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@


import org.apache.doris.common.security.authentication.ExecutionAuthenticator;
import org.apache.doris.common.util.FairTaskExecutor;
import org.apache.doris.metric.Metric;
import org.apache.doris.metric.Metric.MetricUnit;
import org.apache.doris.metric.MetricLabel;
Expand Down Expand Up @@ -75,6 +76,7 @@ public class ThreadPoolManager {

private static final Logger LOG = LogManager.getLogger(ThreadPoolManager.class);
private static Map<String, ThreadPoolExecutor> nameToThreadPoolMap = Maps.newConcurrentMap();
private static final Map<String, FairTaskExecutor> nameToFairTaskExecutorMap = Maps.newConcurrentMap();

private static String[] poolMetricTypes = {"pool_size", "active_thread_num", "task_in_queue"};

Expand All @@ -85,6 +87,21 @@ public static void registerAllThreadPoolMetric() {
registerThreadPoolMetric(entry.getKey(), entry.getValue());
}
nameToThreadPoolMap.clear();
for (Map.Entry<String, FairTaskExecutor> entry : nameToFairTaskExecutorMap.entrySet()) {
registerThreadPoolMetric(entry.getKey(), entry.getValue());
}
nameToFairTaskExecutorMap.clear();
}

public static void registerThreadPoolMetric(String poolName, FairTaskExecutor executor) {
Metric.MetricType gauge = Metric.MetricType.GAUGE;
MetricUnit nounit = MetricUnit.NOUNIT;
registerMetric(poolName, "pool_size", gauge, nounit, executor::getPoolSize);
registerMetric(poolName, "active_thread_num", gauge, nounit, executor::getActiveCount);
registerMetric(poolName, "task_in_queue", gauge, nounit, executor::getQueuedTaskCount);
registerMetric(poolName, "completed_task_count", Metric.MetricType.COUNTER, nounit,
executor::getCompletedTaskCount);
registerMetric(poolName, "task_group_num", gauge, nounit, executor::getGroupCount);
}

public static void registerThreadPoolMetric(String poolName, ThreadPoolExecutor threadPool) {
Expand Down Expand Up @@ -117,6 +134,15 @@ public T getValue() {
MetricRepo.DORIS_METRIC_REGISTER.addMetrics(gauge);
}

public static FairTaskExecutor newDaemonFairTaskExecutor(int parallelism, String poolName,
boolean needRegisterMetric) {
FairTaskExecutor executor = new FairTaskExecutor(parallelism, namedThreadFactory(poolName));
if (needRegisterMetric) {
nameToFairTaskExecutorMap.put(poolName, executor);
}
return executor;
}

public static ThreadPoolExecutor newDaemonCacheThreadPool(int maxNumThread,
String poolName, boolean needRegisterMetric) {
return newDaemonThreadPool(0, maxNumThread, KEEP_ALIVE_TIME,
Expand Down
Loading
Loading