This is an automated email from the ASF dual-hosted git repository.
HappenLee 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 6e4d434e41f [fix](fe) Enable profiles for computed DML commands
(#64551)
6e4d434e41f is described below
commit 6e4d434e41f971a8ec25c9d12585cda64185c2a4
Author: HappenLee <[email protected]>
AuthorDate: Thu Jun 25 15:48:42 2026 +0800
[fix](fe) Enable profiles for computed DML commands (#64551)
Problem Summary:
Some DML statements that execute through the compute path were treated
as profile-unsafe statements, so `UPDATE`, `MERGE INTO`, and `DELETE
FROM ... USING` did not produce query/load profiles even when profiling
was enabled.
This PR marks `UpdateCommand`, `MergeIntoCommand`, and
`DeleteFromUsingCommand` as profile-safe. Regular `DeleteFromCommand` is
left unchanged because it should not produce a compute profile.
### Release note
Support profiles for `UPDATE`, `MERGE INTO`, and `DELETE FROM ... USING`
statements.
### Check List (For Author)
- Test: Regression test / Build
- `./build.sh --fe`
- `./run-regression-test.sh --run --conf
/tmp/doris-regression-conf-run-path.groovy -d query_profile -s
dml_profile_safe`
- Behavior changed: Yes
- `UPDATE`, `MERGE INTO`, and `DELETE FROM ... USING` now generate
profiles when profiling is enabled.
- Does this need documentation: No
---
.../plans/commands/DeleteFromUsingCommand.java | 7 +-
.../trees/plans/commands/SupportProfile.java | 47 ++++++
.../trees/plans/commands/UpdateCommand.java | 7 +-
.../plans/commands/merge/MergeIntoCommand.java | 8 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 6 +
...ptive_pipeline_task_serial_read_on_limit.groovy | 6 +-
.../suites/query_profile/dml_profile_safe.groovy | 181 +++++++++++++++++++++
7 files changed, 256 insertions(+), 6 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java
index 2364c690712..6d6b53d2254 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteFromUsingCommand.java
@@ -33,7 +33,7 @@ import java.util.Optional;
/**
* delete from unique key table.
*/
-public class DeleteFromUsingCommand extends DeleteFromCommand {
+public class DeleteFromUsingCommand extends DeleteFromCommand implements
SupportProfile {
private final Optional<LogicalPlan> cte;
private final boolean hasOrderByLimit;
@@ -75,6 +75,11 @@ public class DeleteFromUsingCommand extends
DeleteFromCommand {
return logicalQuery;
}
+ @Override
+ public List<String> getTargetTableNameParts() {
+ return nameParts;
+ }
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitDeleteFromUsingCommand(this, context);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/SupportProfile.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/SupportProfile.java
new file mode 100644
index 00000000000..cddca92a3bc
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/SupportProfile.java
@@ -0,0 +1,47 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.nereids.util.RelationUtil;
+import org.apache.doris.qe.ConnectContext;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Commands that can generate query profiles.
+ */
+public interface SupportProfile {
+
+ List<String> getTargetTableNameParts();
+
+ /**
+ * Check whether the target table is an OLAP table.
+ */
+ default boolean isTargetTableOlap(ConnectContext ctx) {
+ try {
+ List<String> qualifiedTableName =
RelationUtil.getQualifierName(ctx, getTargetTableNameParts());
+ TableIf table = RelationUtil.getTable(qualifiedTableName,
ctx.getEnv(), Optional.empty());
+ return table instanceof OlapTable;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateCommand.java
index 4e717bbb1d6..736577f8804 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateCommand.java
@@ -76,7 +76,7 @@ import javax.annotation.Nullable;
* =>
* insert into t1 (c1, c3) select t2.c1, t2.c3 * 100 from t1 join t2 inner
join t3 on t2.id = t3.id where t1.id = t2.id
*/
-public class UpdateCommand extends Command implements ForwardWithSync,
Explainable {
+public class UpdateCommand extends Command implements ForwardWithSync,
Explainable, SupportProfile {
private final List<EqualTo> assignments;
private final List<String> nameParts;
private final @Nullable String tableAlias;
@@ -295,6 +295,11 @@ public class UpdateCommand extends Command implements
ForwardWithSync, Explainab
return logicalQuery;
}
+ @Override
+ public List<String> getTargetTableNameParts() {
+ return nameParts;
+ }
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitUpdateCommand(this, context);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/merge/MergeIntoCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/merge/MergeIntoCommand.java
index 4a68f739651..e002e596c7d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/merge/MergeIntoCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/merge/MergeIntoCommand.java
@@ -54,6 +54,7 @@ import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.Command;
import org.apache.doris.nereids.trees.plans.commands.ForwardWithSync;
import org.apache.doris.nereids.trees.plans.commands.IcebergMergeCommand;
+import org.apache.doris.nereids.trees.plans.commands.SupportProfile;
import org.apache.doris.nereids.trees.plans.commands.UpdateCommand;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
@@ -84,7 +85,7 @@ import java.util.Optional;
/**
* merge into table
*/
-public class MergeIntoCommand extends Command implements ForwardWithSync,
Explainable {
+public class MergeIntoCommand extends Command implements ForwardWithSync,
Explainable, SupportProfile {
private static final String BRANCH_LABEL =
"__DORIS_MERGE_INTO_BRANCH_LABEL__";
private final List<String> targetNameParts;
@@ -153,6 +154,11 @@ public class MergeIntoCommand extends Command implements
ForwardWithSync, Explai
return RelationUtil.getTable(qualifiedTableName, ctx.getEnv(),
Optional.empty());
}
+ @Override
+ public List<String> getTargetTableNameParts() {
+ return targetNameParts;
+ }
+
private OlapTable getTargetTable(ConnectContext ctx) {
TableIf table = getTargetTableIf(ctx);
if (!(table instanceof OlapTable) || !((OlapTable)
table).getEnableUniqueKeyMergeOnWrite()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 62773925744..8d4f29cda86 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -102,6 +102,7 @@ import
org.apache.doris.nereids.trees.plans.commands.Forward;
import org.apache.doris.nereids.trees.plans.commands.LoadCommand;
import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
import org.apache.doris.nereids.trees.plans.commands.Redirect;
+import org.apache.doris.nereids.trees.plans.commands.SupportProfile;
import org.apache.doris.nereids.trees.plans.commands.TransactionCommand;
import org.apache.doris.nereids.trees.plans.commands.UpdateCommand;
import
org.apache.doris.nereids.trees.plans.commands.insert.BatchInsertIntoTableCommand;
@@ -1146,6 +1147,11 @@ public class StmtExecutor {
return true;
}
+ // Computed DML profiles are currently only supported for OLAP target
tables.
+ if (plan instanceof SupportProfile && ((SupportProfile)
plan).isTargetTableOlap(context)) {
+ return true;
+ }
+
// Generate profile for:
// 1. CreateTableCommand(mainly for create as select).
// 2. LoadCommand.
diff --git
a/regression-test/suites/query_profile/adaptive_pipeline_task_serial_read_on_limit.groovy
b/regression-test/suites/query_profile/adaptive_pipeline_task_serial_read_on_limit.groovy
index 4a36401b02b..4ccd45af04a 100644
---
a/regression-test/suites/query_profile/adaptive_pipeline_task_serial_read_on_limit.groovy
+++
b/regression-test/suites/query_profile/adaptive_pipeline_task_serial_read_on_limit.groovy
@@ -21,8 +21,8 @@ import groovy.json.StringEscapeUtils
import org.apache.doris.regression.action.ProfileAction
def verifyProfileContent = { suiteContext, stmt, serialReadOnLimit ->
- // Sleep 500ms to wait for the profile collection
- Thread.sleep(500)
+ // Sleep 1000ms to wait for the profile collection
+ Thread.sleep(1000)
// Get profile list by using getProfileList
def profileAction = new ProfileAction(suiteContext)
List profileData = profileAction.getProfileList()
@@ -122,4 +122,4 @@ suite('adaptive_pipeline_task_serial_read_on_limit') {
sql "set enable_adaptive_pipeline_task_serial_read_on_limit=false;"
sql """select "enable_adaptive_pipeline_task_serial_read_on_limit=false",
* from adaptive_pipeline_task_serial_read_on_limit limit 1000000;"""
assertTrue(verifyProfileContent(context, "select
\"enable_adaptive_pipeline_task_serial_read_on_limit=false\", * from
adaptive_pipeline_task_serial_read_on_limit limit 1000000;", false))
-}
\ No newline at end of file
+}
diff --git a/regression-test/suites/query_profile/dml_profile_safe.groovy
b/regression-test/suites/query_profile/dml_profile_safe.groovy
new file mode 100644
index 00000000000..17618df34d1
--- /dev/null
+++ b/regression-test/suites/query_profile/dml_profile_safe.groovy
@@ -0,0 +1,181 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import groovy.json.JsonSlurper
+
+def getProfileList = { masterHTTPAddr ->
+ def dst = 'http://' + masterHTTPAddr
+ def conn = new URL(dst + "/rest/v1/query_profile").openConnection()
+ conn.setRequestMethod("GET")
+ def password = context.config.feHttpPassword == null ? "" :
context.config.feHttpPassword
+ def encoding = Base64.getEncoder()
+ .encodeToString((context.config.feHttpUser + ":" +
password).getBytes("UTF-8"))
+ conn.setRequestProperty("Authorization", "Basic ${encoding}")
+ return conn.getInputStream().getText()
+}
+
+def getMasterHttpAddress = { allFrontends ->
+ for (def frontend : allFrontends) {
+ if (frontend[8] == "true") {
+ return frontend[1] + ":" + frontend[3]
+ }
+ }
+ throw new IllegalStateException("master frontend not found")
+}
+
+def hasProfile = { masterHTTPAddr, taskType, sqlToken ->
+ def wholeString = getProfileList(masterHTTPAddr)
+ def profileListData = new JsonSlurper().parseText(wholeString).data.rows
+ for (def profileList : profileListData) {
+ def profileTaskType = profileList["Task Type"].toString()
+ def stmt = profileList["Sql Statement"].toString().toLowerCase()
+ if (profileTaskType == taskType &&
stmt.contains(sqlToken.toLowerCase())) {
+ return true
+ }
+ }
+ return false
+}
+
+def waitForProfile = { masterHTTPAddr, taskType, sqlToken ->
+ for (int i = 0; i < 20; i++) {
+ if (hasProfile(masterHTTPAddr, taskType, sqlToken)) {
+ return true
+ }
+ Thread.sleep(500)
+ }
+ return false
+}
+
+suite("dml_profile_safe") {
+ sql """set enable_nereids_planner = true;"""
+ sql """set enable_fallback_to_original_planner = false;"""
+
+ sql """drop table if exists dml_profile_update_target;"""
+ sql """drop table if exists dml_profile_merge_target;"""
+ sql """drop table if exists dml_profile_merge_source;"""
+ sql """drop table if exists dml_profile_delete_using_target;"""
+ sql """drop table if exists dml_profile_delete_using_source;"""
+ sql """drop table if exists dml_profile_delete_command_target;"""
+
+ sql """
+ create table dml_profile_update_target (
+ k int,
+ v int
+ )
+ unique key(k)
+ distributed by hash(k) buckets 1
+ properties(
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ sql """
+ create table dml_profile_merge_target (
+ k int,
+ v int
+ )
+ unique key(k)
+ distributed by hash(k) buckets 1
+ properties(
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ sql """
+ create table dml_profile_merge_source (
+ k int,
+ v int
+ )
+ duplicate key(k)
+ distributed by hash(k) buckets 1
+ properties("replication_num" = "1");
+ """
+
+ sql """
+ create table dml_profile_delete_using_target (
+ k int,
+ v int
+ )
+ unique key(k)
+ distributed by hash(k) buckets 1
+ properties(
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true"
+ );
+ """
+
+ sql """
+ create table dml_profile_delete_using_source (
+ k int
+ )
+ duplicate key(k)
+ distributed by hash(k) buckets 1
+ properties("replication_num" = "1");
+ """
+
+ sql """
+ create table dml_profile_delete_command_target (
+ k int,
+ v int
+ )
+ duplicate key(k)
+ distributed by hash(k) buckets 1
+ properties("replication_num" = "1");
+ """
+
+ sql """insert into dml_profile_update_target values (1, 10), (2, 20);"""
+ sql """insert into dml_profile_merge_target values (1, 10);"""
+ sql """insert into dml_profile_merge_source values (1, 11), (2, 22);"""
+ sql """insert into dml_profile_delete_using_target values (1, 10), (2,
20);"""
+ sql """insert into dml_profile_delete_using_source values (1);"""
+ sql """insert into dml_profile_delete_command_target values (1, 10), (2,
20);"""
+ sql """sync;"""
+
+ def masterHTTPAddr = getMasterHttpAddress(sql """show frontends;""")
+
+ sql """clean all profile;"""
+ sql """set enable_profile = true;"""
+ sql """set profile_level = 2;"""
+
+ sql """update dml_profile_update_target set v = v + 1 where k = 1;"""
+ assertTrue(waitForProfile(masterHTTPAddr, "LOAD",
+ "update dml_profile_update_target set v = v + 1 where k = 1"))
+
+ sql """
+ merge into dml_profile_merge_target t
+ using dml_profile_merge_source s
+ on t.k = s.k
+ when matched then update set v = s.v
+ when not matched then insert values(s.k, s.v);
+ """
+ assertTrue(waitForProfile(masterHTTPAddr, "LOAD", "merge into
dml_profile_merge_target"))
+
+ sql """
+ delete from dml_profile_delete_using_target t
+ using dml_profile_delete_using_source s
+ where t.k = s.k;
+ """
+ assertTrue(waitForProfile(masterHTTPAddr, "LOAD", "delete from
dml_profile_delete_using_target"))
+
+ sql """clean all profile;"""
+ sql """delete from dml_profile_delete_command_target where k = 1;"""
+ Thread.sleep(1000)
+ assertFalse(hasProfile(masterHTTPAddr, "LOAD", "delete from
dml_profile_delete_command_target"))
+ assertFalse(hasProfile(masterHTTPAddr, "QUERY", "delete from
dml_profile_delete_command_target"))
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
