github-actions[bot] commented on code in PR #68281:
URL: https://github.com/apache/doris/pull/68281#discussion_r4058867890
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -4148,10 +4154,46 @@ public void setMaxExecutionTimeMS(String
maxExecutionTimeMS) {
}
public void setPipelineTaskNum(String value) throws Exception {
- int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0, value);
+ int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0,
Config.max_parallel_pipeline_task_num, value);
this.parallelPipelineTaskNum = val;
}
+ public void checkColocateMaxParallelNum(String value) throws Exception {
+ checkFieldValue(COLOCATE_MAX_PARALLEL_NUM, 1,
Config.max_colocate_parallel_num, value);
Review Comment:
[P1] Enforce the checker on map and forwarded session inputs
The new annotations protect the checker-aware `VariableMgr.setValue` path,
but streaming-job `session.*` properties use reflective `readFromMap`, and
`setForwardedSessionVariables` uses the raw
`VariableMgr.setValue(Object,String,Field,...)` overload. An external job or an
older/mismatched FE can therefore install an over-cap value and send it to
planning/Thrift. Route these inputs through one checker-aware path (with an
explicit mixed-version policy) or revalidate before installation/serialization.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -4148,10 +4154,46 @@ public void setMaxExecutionTimeMS(String
maxExecutionTimeMS) {
Review Comment:
[P2] Keep SET GLOBAL atomic across both checker calls
`setVarInternal` first mutates `defaultSessionVariable` and writes the edit
log, then checks/mutates the caller session in a second call. Because both
checks reread a mutable `Config.max_*` and the config update has no shared
lock/snapshot, an intervening ADMIN SET that lowers the cap can make the
statement fail after publishing the global value. Please validate both targets
against one effective-limit snapshot or make the operation rollback-safe before
logging.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -586,6 +586,27 @@ public class Config extends ConfigBase {
@ConfField(description = "The timeout of RPC for high-concurrency
short-circuit queries")
public static int point_query_timeout_ms = 10000; // 10s
+ @ConfField(mutable = true, description = "Upper limit for the
parallel_pipeline_task_num session variable. "
+ + "Zero still selects pipeline parallelism automatically.")
+ public static volatile int max_parallel_pipeline_task_num = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
colocate_max_parallel_num session variable.")
+ public static volatile int max_colocate_parallel_num = 256;
Review Comment:
[P1] Reconcile defaults and image state when a cap is lowered
Lowering this cap does not update `defaultSessionVariable`, so sessions
created afterward can clone an over-cap value (for example colocate=128 with
max=64). Image deserialization also assigns serialized ints without the
checker, and replay failures are caught/ignored, allowing the over-cap default
to survive restart. Please define lower-cap semantics and validate or clamp
defaults/image state before exposing them.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -4148,10 +4154,46 @@ public void setMaxExecutionTimeMS(String
maxExecutionTimeMS) {
}
public void setPipelineTaskNum(String value) throws Exception {
- int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0, value);
+ int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0,
Config.max_parallel_pipeline_task_num, value);
Review Comment:
[P2] Preflight default restoration after lowering a cap
`UNSET`/all-default restoration iterates startup-captured defaults and
mutates each field immediately. If an administrator lowers (for example) the
colocate cap below its captured default of 128, this checker throws partway
through the loop after earlier variables were already reset. Please preflight
the complete reset or update/clamp the captured defaults when limits change so
the operation is atomic.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -586,6 +586,27 @@ public class Config extends ConfigBase {
@ConfField(description = "The timeout of RPC for high-concurrency
short-circuit queries")
public static int point_query_timeout_ms = 10000; // 10s
+ @ConfField(mutable = true, description = "Upper limit for the
parallel_pipeline_task_num session variable. "
+ + "Zero still selects pipeline parallelism automatically.")
+ public static volatile int max_parallel_pipeline_task_num = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
colocate_max_parallel_num session variable.")
+ public static volatile int max_colocate_parallel_num = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
max_scanners_concurrency, "
+ + "min_scanners_concurrency, max_file_scanners_concurrency and
min_file_scanners_concurrency "
+ + "session variables.")
+ public static volatile int max_scanners_concurrency = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
parallel_scan_max_scanners_count session variable.")
+ public static volatile int max_parallel_scan_scanners_count = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
send_batch_parallelism session variable.")
+ public static volatile int max_send_batch_parallelism = 256;
+
+ @ConfField(mutable = true, description = "Upper limit for the
load_stream_per_node session variable.")
+ public static volatile int max_load_stream_per_node = 256;
Review Comment:
[P2] Reject invalid mutable cap values
These mutable limits are parsed as unrestricted integers. Setting
`max_load_stream_per_node` or `max_colocate_parallel_num` to 0 or a negative
value makes the corresponding checker (`min=1` and `value <= max`)
unsatisfiable, including `DEFAULT`, and can strand sessions until the config is
repaired. Validate each cap's floor at startup and ADMIN SET time before
publishing it.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -4148,10 +4154,46 @@ public void setMaxExecutionTimeMS(String
maxExecutionTimeMS) {
}
public void setPipelineTaskNum(String value) throws Exception {
- int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0, value);
+ int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0,
Config.max_parallel_pipeline_task_num, value);
this.parallelPipelineTaskNum = val;
}
+ public void checkColocateMaxParallelNum(String value) throws Exception {
+ checkFieldValue(COLOCATE_MAX_PARALLEL_NUM, 1,
Config.max_colocate_parallel_num, value);
+ }
+
+ public void checkMaxScannersConcurrency(String value) throws Exception {
+ // Non-positive scanner concurrency values select the BE defaults.
+ checkFieldValue(MAX_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMaxFileScannersConcurrency(String value) throws Exception
{
+ checkFieldValue(MAX_FILE_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMinScannersConcurrency(String value) throws Exception {
+ checkFieldValue(MIN_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMinFileScannersConcurrency(String value) throws Exception
{
+ checkFieldValue(MIN_FILE_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkParallelScanMaxScannersCount(String value) throws
Exception {
+ // Non-positive values select the number of CPU cores on the BE.
+ checkFieldValue(PARALLEL_SCAN_MAX_SCANNERS_COUNT, Integer.MIN_VALUE,
+ Config.max_parallel_scan_scanners_count, value);
+ }
+
+ public void checkSendBatchParallelism(String value) throws Exception {
+ // The tablet writer uses one sender for values less than or equal to
one.
Review Comment:
[P1] Bound the direct HTTP stream-load override
The SQL session checker does not cover the user `load_stream_per_node` HTTP
header: BE parses it into the request and
`FrontendServiceImpl.httpStreamPutImpl` copies any positive value directly into
fragment parameters, which `VTabletWriterV2` uses for per-node stream
allocation. Apply the same configured cap at this request/planning boundary
(and retain a BE-side sanity check for mixed-version requests).
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -4148,10 +4154,46 @@ public void setMaxExecutionTimeMS(String
maxExecutionTimeMS) {
}
public void setPipelineTaskNum(String value) throws Exception {
- int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0, value);
+ int val = checkFieldValue(PARALLEL_PIPELINE_TASK_NUM, 0,
Config.max_parallel_pipeline_task_num, value);
this.parallelPipelineTaskNum = val;
}
+ public void checkColocateMaxParallelNum(String value) throws Exception {
+ checkFieldValue(COLOCATE_MAX_PARALLEL_NUM, 1,
Config.max_colocate_parallel_num, value);
+ }
+
+ public void checkMaxScannersConcurrency(String value) throws Exception {
+ // Non-positive scanner concurrency values select the BE defaults.
+ checkFieldValue(MAX_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMaxFileScannersConcurrency(String value) throws Exception
{
+ checkFieldValue(MAX_FILE_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMinScannersConcurrency(String value) throws Exception {
+ checkFieldValue(MIN_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkMinFileScannersConcurrency(String value) throws Exception
{
+ checkFieldValue(MIN_FILE_SCANNERS_CONCURRENCY, Integer.MIN_VALUE,
Config.max_scanners_concurrency, value);
+ }
+
+ public void checkParallelScanMaxScannersCount(String value) throws
Exception {
+ // Non-positive values select the number of CPU cores on the BE.
+ checkFieldValue(PARALLEL_SCAN_MAX_SCANNERS_COUNT, Integer.MIN_VALUE,
+ Config.max_parallel_scan_scanners_count, value);
Review Comment:
[P1] Apply the send-batch cap to LOAD properties
`max_send_batch_parallelism` only guards SQL session assignment. LOAD checks
merely positivity and routine load accepts a positive long before narrowing to
int; both values reach `VTabletWriter` as sender thread-pool concurrency
without consulting this cap. Reuse the bound in those property validators,
reject out-of-range values before narrowing, and keep a BE-side check for
mixed-version requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
