This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new e55e803688d branch-4.2: [fix](point-query) Refresh stale prepared
short-circuit plans #63920 (#70084)
e55e803688d is described below
commit e55e803688dbd69fc9d681809c8acfb6e1b167dc
Author: morrySnow <[email protected]>
AuthorDate: Wed Nov 4 08:24:15 2026 +0800
branch-4.2: [fix](point-query) Refresh stale prepared short-circuit plans
#63920 (#70084)
picked from master #63920
Co-authored-by: HonestManXin <[email protected]>
---
.../nereids/trees/expressions/Placeholder.java | 8 ++
.../trees/plans/commands/ExecuteCommand.java | 111 ++++++++++++++++++++-
.../org/apache/doris/planner/OlapScanNode.java | 6 ++
.../apache/doris/qe/ShortCircuitQueryContext.java | 8 +-
.../trees/plans/commands/ExecuteCommandTest.java | 1 +
.../doris/qe/ShortCircuitQueryContextTest.java | 19 ++++
.../suites/point_query_p0/test_point_query.groovy | 101 +++++++++++++++++++
7 files changed, 252 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Placeholder.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Placeholder.java
index ce92a0e36c8..2806e9a5bdc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Placeholder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/Placeholder.java
@@ -93,6 +93,14 @@ public class Placeholder extends Expression implements
LeafExpression {
return MysqlColType.isUnsigned(mysqlTypeCode);
}
+ public boolean hasMysqlColType() {
+ return mysqlColType.isPresent();
+ }
+
+ public int getMysqlTypeCode() {
+ return mysqlTypeCode;
+ }
+
public MysqlColType getMysqlColType() {
return mysqlColType.get();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java
index 203deb626a8..e985132527f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java
@@ -17,14 +17,19 @@
package org.apache.doris.nereids.trees.plans.commands;
+import org.apache.doris.analysis.StatementBase;
import org.apache.doris.analysis.StmtType;
import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.catalog.MysqlColType;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundRelation;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Placeholder;
import org.apache.doris.nereids.trees.expressions.SubqueryExpr;
+import org.apache.doris.nereids.trees.plans.PlaceholderId;
import org.apache.doris.nereids.trees.plans.PlanType;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertOverwriteTableCommand;
@@ -41,9 +46,13 @@ import org.apache.doris.qe.ShortCircuitQueryContext;
import org.apache.doris.qe.StmtExecutor;
import com.google.common.base.Preconditions;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -51,6 +60,8 @@ import java.util.stream.Collectors;
* Prepared Statement
*/
public class ExecuteCommand extends Command {
+ private static final Logger LOG =
LogManager.getLogger(ExecuteCommand.class);
+
private final String stmtName;
private final PrepareCommand prepareCommand;
private final StatementContext statementContext;
@@ -168,8 +179,14 @@ public class ExecuteCommand extends Command {
}
}
// execute real statement
+ if (statementContext.isShortCircuitQuery() && hasShortCircuitContext
&& !shortCircuitContextReusable) {
+ statementContext = refreshPreparedPlan(preparedStmtCtx, executor,
prepareCommand, statementContext);
+ } else {
+ statementContext.setShortCircuitQueryContext(null);
+ }
+ // Drop the previously cached short-circuit context: either it was
reusable and returned
+ // early above, has just been refreshed here, or is stale and we are
about to re-plan.
preparedStmtCtx.shortCircuitQueryContext = Optional.empty();
- statementContext.setShortCircuitQueryContext(null);
executor.execute();
StatementContext executedStatementContext =
executor.getContext().getStatementContext();
ShortCircuitQueryContext shortCircuitQueryContext =
@@ -182,6 +199,98 @@ public class ExecuteCommand extends Command {
}
}
+ private StatementContext refreshPreparedPlan(PreparedStatementContext
preparedStmtCtx, StmtExecutor executor,
+ PrepareCommand currentCommand, StatementContext
currentStatementContext) {
+ Map<PlaceholderId, Expression> boundPlaceholderValues = new HashMap<>(
+ currentStatementContext.getIdToPlaceholderRealExpr());
+ StatementContext originalStatementContext =
executor.getContext().getStatementContext();
+ StatementBase originalParsedStmt = executor.getParsedStmt();
+ PrepareCommand originalCommand = preparedStmtCtx.command;
+ StatementContext originalPreparedStatementContext =
preparedStmtCtx.getStatementContext();
+ boolean refreshed = false;
+ try {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("refresh prepared statement plan for short-circuit
query, stmtName={}",
+ currentCommand.getName());
+ }
+ List<StatementBase> reparsedStatements = new
NereidsParser().parseSQL(
+ currentCommand.getOriginalStmt().originStmt,
executor.getContext().getSessionVariable());
+ if (reparsedStatements.size() <=
currentCommand.getOriginalStmt().idx) {
+ throw new AnalysisException("Nereids parse failed. Parser get
" + reparsedStatements.size()
+ + " statements, but we need at least " +
(currentCommand.getOriginalStmt().idx + 1)
+ + " statements.");
+ }
+ StatementBase reparsedStmt =
reparsedStatements.get(currentCommand.getOriginalStmt().idx);
+ if (!(reparsedStmt instanceof LogicalPlanAdapter)) {
+ throw new AnalysisException("Prepared statement must be parsed
as LogicalPlanAdapter, but get "
+ + reparsedStmt.getClass().getName());
+ }
+
+ LogicalPlanAdapter reparsedAdapter = (LogicalPlanAdapter)
reparsedStmt;
+ StatementContext reparsedStatementContext =
reparsedAdapter.getStatementContext();
+ List<Placeholder> reparsedPlaceholders =
reparsedStatementContext.getPlaceholders();
+ List<Placeholder> currentPlaceholders =
currentCommand.getPlaceholders();
+ if (reparsedPlaceholders.size() != currentPlaceholders.size()) {
+ throw new AnalysisException("Prepared statement placeholder
count changed after reparse, old: "
+ + currentPlaceholders.size() + ", new: " +
reparsedPlaceholders.size());
+ }
+
+ reparsedStatementContext.setConnectContext(executor.getContext());
+
reparsedStatementContext.setOriginStatement(currentCommand.getOriginalStmt());
+ reparsedStatementContext.getIdToPlaceholderRealExpr().clear();
+
+ List<Placeholder> refreshedPlaceholders = new
ArrayList<>(reparsedPlaceholders.size());
+ for (int i = 0; i < reparsedPlaceholders.size(); i++) {
+ Placeholder oldPlaceholder = currentPlaceholders.get(i);
+ Placeholder refreshedPlaceholder =
withMysqlType(reparsedPlaceholders.get(i), oldPlaceholder);
+ refreshedPlaceholders.add(refreshedPlaceholder);
+
+ Expression boundValue =
boundPlaceholderValues.get(oldPlaceholder.getPlaceholderId());
+ if (boundValue != null) {
+ reparsedStatementContext.getIdToPlaceholderRealExpr().put(
+ refreshedPlaceholder.getPlaceholderId(),
boundValue);
+ }
+ }
+ reparsedStatementContext.setPlaceholders(refreshedPlaceholders);
+
+ reparsedAdapter.setOrigStmt(currentCommand.getOriginalStmt());
+ executor.setStatementContext(reparsedStatementContext);
+ executor.setParsedStmt(reparsedAdapter);
+ PrepareCommand refreshedCommand = new
PrepareCommand(currentCommand.getName(),
+ reparsedAdapter.getLogicalPlan(), refreshedPlaceholders,
currentCommand.getOriginalStmt());
+ preparedStmtCtx.command = refreshedCommand;
+ preparedStmtCtx.setStatementContext(reparsedStatementContext);
+ refreshed = true;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("refreshed prepared statement plan for short-circuit
query, stmtName={}",
+ currentCommand.getName());
+ }
+ return reparsedStatementContext;
+ } finally {
+ if (!refreshed) {
+ executor.setStatementContext(originalStatementContext);
+ executor.setParsedStmt(originalParsedStmt);
+ preparedStmtCtx.command = originalCommand;
+
preparedStmtCtx.setStatementContext(originalPreparedStatementContext);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("rollback prepared statement plan refresh for
short-circuit query, stmtName={}",
+ currentCommand.getName());
+ }
+ }
+ }
+ }
+
+ private Placeholder withMysqlType(Placeholder refreshedPlaceholder,
Placeholder oldPlaceholder) {
+ if (!oldPlaceholder.hasMysqlColType()) {
+ return refreshedPlaceholder;
+ }
+ int mysqlTypeCode = oldPlaceholder.getMysqlTypeCode() &
MysqlColType.MYSQL_CODE_MASK;
+ if (oldPlaceholder.isUnsigned()) {
+ mysqlTypeCode |= MysqlColType.UNSIGNED_MASK;
+ }
+ return refreshedPlaceholder.withNewMysqlColType(mysqlTypeCode);
+ }
+
/**
* return the sql representation contains real expr instead of placeholders
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 36b9601fbad..ae928c706bd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -162,6 +162,7 @@ public class OlapScanNode extends ScanNode {
private String reasonOfPreAggregation = null;
private boolean forceOpenPreAgg = false;
private OlapTable olapTable = null;
+ private String tableNameInPlan = null;
private long totalTabletsNum = 0;
private long selectedIndexId = -1;
private Collection<Long> selectedPartitionIds = Lists.newArrayList();
@@ -228,6 +229,7 @@ public class OlapScanNode extends ScanNode {
public OlapScanNode(PlanNodeId id, TupleDescriptor desc, String
planNodeName, ScanContext scanContext) {
super(id, desc, planNodeName, scanContext,
StatisticalType.OLAP_SCAN_NODE);
olapTable = (OlapTable) desc.getTable();
+ tableNameInPlan = olapTable.getName();
distributionColumnIds = Sets.newTreeSet();
Set<String> distColumnName = getDistributionColumnNames();
@@ -357,6 +359,10 @@ public class OlapScanNode extends ScanNode {
return helper.toString();
}
+ public String getTableNameInPlan() {
+ return tableNameInPlan;
+ }
+
/**
* Init OlapScanNode, ONLY used for Nereids. Should NOT use this function
in anywhere else.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/ShortCircuitQueryContext.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ShortCircuitQueryContext.java
index 4cda24153d3..b48b196bb4e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ShortCircuitQueryContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ShortCircuitQueryContext.java
@@ -37,6 +37,7 @@ import org.apache.thrift.TSerializer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -57,6 +58,7 @@ public class ShortCircuitQueryContext {
public final OlapTable tbl;
private final long fileCacheQueryLimitBytes;
private final long partitionTopologyVersion;
+ public final String tableName;
public final OlapScanNode scanNode;
public final Queriable analzyedQuery;
@@ -109,6 +111,7 @@ public class ShortCircuitQueryContext {
this.cacheID = UUID.randomUUID();
this.scanNode = olapScanNode;
this.tbl = this.scanNode.getOlapTable();
+ this.tableName = this.scanNode.getTableNameInPlan();
this.schemaVersion = this.tbl.getBaseSchemaVersion();
this.partitionTopologyVersion = this.tbl.getPartitionTopologyVersion();
this.analzyedQuery = analzyedQuery;
@@ -122,6 +125,7 @@ public class ShortCircuitQueryContext {
this.serializedQueryOptions = ByteString.EMPTY;
this.cacheID = UUID.randomUUID();
this.tbl = tbl;
+ this.tableName = tbl.getName();
this.schemaVersion = schemaVersion;
this.fileCacheQueryLimitBytes = fileCacheQueryLimitBytes;
this.partitionTopologyVersion = tbl.getPartitionTopologyVersion();
@@ -130,7 +134,8 @@ public class ShortCircuitQueryContext {
}
public boolean isReusable(ConnectContext ctx) {
- return this.tbl.getBaseSchemaVersion() == this.schemaVersion
+ return !this.tbl.isDropped && Objects.equals(this.tableName,
this.tbl.getName())
+ && this.tbl.getBaseSchemaVersion() == this.schemaVersion
&& this.fileCacheQueryLimitBytes ==
ctx.getSessionVariable().fileCacheQueryLimitBytes
&& this.tbl.getPartitionTopologyVersion() ==
this.partitionTopologyVersion;
}
@@ -140,5 +145,6 @@ public class ShortCircuitQueryContext {
Preconditions.checkNotNull(serializedOutputExpr);
Preconditions.checkNotNull(cacheID);
Preconditions.checkNotNull(tbl);
+ Preconditions.checkNotNull(tableName);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommandTest.java
index 219ac4dfe2c..24495c6d9a4 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommandTest.java
@@ -322,6 +322,7 @@ public class ExecuteCommandTest {
Mockito.doReturn("tbl").when(table).getName();
Mockito.doReturn(10).when(table).getBaseSchemaVersion();
Mockito.when(scanNode.getOlapTable()).thenReturn(table);
+ Mockito.when(scanNode.getTableNameInPlan()).thenReturn("tbl");
Mockito.when(scanNode.getConjuncts()).thenReturn(Collections.emptyList());
Mockito.when(planner.getScanNodes()).thenReturn(Collections.singletonList(scanNode));
ShortCircuitQueryContext cachedPlan = new
ShortCircuitQueryContext(planner, Mockito.mock(Queriable.class));
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/ShortCircuitQueryContextTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/ShortCircuitQueryContextTest.java
index f5a13b13cdd..59ebd8c586c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/ShortCircuitQueryContextTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/ShortCircuitQueryContextTest.java
@@ -86,4 +86,23 @@ public class ShortCircuitQueryContextTest {
new RandomDistributionInfo(1)));
Assertions.assertFalse(context.isReusable(connectContext(-1)));
}
+
+ @Test
+ public void testDroppedTableCannotReuseCachedPlan() {
+ OlapTable table = table("tbl", 10);
+ ShortCircuitQueryContext context = new ShortCircuitQueryContext(table,
10, -1);
+ Assertions.assertTrue(context.isReusable(connectContext(-1)));
+ table.isDropped = true;
+ Assertions.assertFalse(context.isReusable(connectContext(-1)));
+ }
+
+ @Test
+ public void testRenamedTableCannotReuseCachedPlan() {
+ OlapTable table = table("tbl", 10);
+ ShortCircuitQueryContext context = new ShortCircuitQueryContext(table,
10, -1);
+ Assertions.assertTrue(context.isReusable(connectContext(-1)));
+ Mockito.doReturn("renamed").when(table).getName();
+ Assertions.assertFalse(context.isReusable(connectContext(-1)));
+ }
+
}
diff --git a/regression-test/suites/point_query_p0/test_point_query.groovy
b/regression-test/suites/point_query_p0/test_point_query.groovy
index 8bb21c743cc..def5c12d192 100644
--- a/regression-test/suites/point_query_p0/test_point_query.groovy
+++ b/regression-test/suites/point_query_p0/test_point_query.groovy
@@ -274,6 +274,107 @@ suite("test_point_query") {
}
}
+ sql "DROP TABLE IF EXISTS ${realDb}.tbl_point_query_meta_change"
+ sql "DROP TABLE IF EXISTS
${realDb}.tbl_point_query_meta_change_renamed"
+ sql "DROP TABLE IF EXISTS
${realDb}.tbl_point_query_meta_change_replace"
+ sql "DROP TABLE IF EXISTS ${realDb}.tbl_point_query_meta_change_swap"
+ sql "DROP TABLE IF EXISTS
${realDb}.tbl_point_query_meta_change_string_key"
+ sql "DROP TABLE IF EXISTS
${realDb}.tbl_point_query_meta_change_string_key_renamed"
+ def create_meta_change_table = { tbl ->
+ sql """
+ CREATE TABLE ${tbl} (
+ `k` int NOT NULL,
+ `v` varchar(30) NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "true",
+ "store_row_column" = "true"
+ )
+ """
+ }
+ def create_string_key_meta_change_table = { tbl ->
+ sql """
+ CREATE TABLE ${tbl} (
+ `k` varchar(30) NOT NULL,
+ `v` varchar(30) NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(`k`)
+ DISTRIBUTED BY HASH(`k`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "true",
+ "store_row_column" = "true"
+ )
+ """
+ }
+ def get_prepared_value = { stmt, key ->
+ stmt.setInt(1, key)
+ try (ResultSet rs = stmt.executeQuery()) {
+ assertTrue(rs.next())
+ def value = rs.getString(1)
+ assertFalse(rs.next())
+ return value
+ }
+ }
+ def get_prepared_string_key_value = { stmt, key ->
+ stmt.setString(1, key)
+ try (ResultSet rs = stmt.executeQuery()) {
+ assertTrue(rs.next())
+ def value = rs.getString(1)
+ assertFalse(rs.next())
+ return value
+ }
+ }
+ create_meta_change_table("${realDb}.tbl_point_query_meta_change")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change VALUES (1,
'origin')"
+ connect(user, password, prepare_url) {
+ def stmt = prepareStatement "select /*+
SET_VAR(enable_nereids_planner=true) */ v from
${realDb}.tbl_point_query_meta_change where k = ?"
+ assertEquals(stmt.class, com.mysql.cj.jdbc.ServerPreparedStatement)
+ assertEquals('origin', get_prepared_value(stmt, 1))
+ assertEquals('origin', get_prepared_value(stmt, 1))
+
+ nprep_sql "ALTER TABLE ${realDb}.tbl_point_query_meta_change
RENAME tbl_point_query_meta_change_renamed"
+ create_meta_change_table("${realDb}.tbl_point_query_meta_change")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change VALUES (1,
'renamed')"
+ assertEquals('renamed', get_prepared_value(stmt, 1))
+
+ sql "DROP TABLE ${realDb}.tbl_point_query_meta_change"
+ create_meta_change_table("${realDb}.tbl_point_query_meta_change")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change VALUES (1,
'recreated')"
+ assertEquals('recreated', get_prepared_value(stmt, 1))
+
+
create_meta_change_table("${realDb}.tbl_point_query_meta_change_replace")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change_replace
VALUES (1, 'replaced')"
+ sql "ALTER TABLE ${realDb}.tbl_point_query_meta_change REPLACE
WITH TABLE tbl_point_query_meta_change_replace PROPERTIES('swap' = 'false')"
+ assertEquals('replaced', get_prepared_value(stmt, 1))
+
+
create_meta_change_table("${realDb}.tbl_point_query_meta_change_swap")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change_swap VALUES
(1, 'swapped')"
+ sql "ALTER TABLE ${realDb}.tbl_point_query_meta_change REPLACE
WITH TABLE tbl_point_query_meta_change_swap PROPERTIES('swap' = 'true')"
+ assertEquals('swapped', get_prepared_value(stmt, 1))
+ stmt.close()
+ }
+
+
create_string_key_meta_change_table("${realDb}.tbl_point_query_meta_change_string_key")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change_string_key
VALUES ('key1', 'string_origin')"
+ connect(user, password, prepare_url) {
+ def stmt = prepareStatement "select /*+
SET_VAR(enable_nereids_planner=true) */ v from
${realDb}.tbl_point_query_meta_change_string_key where k = ?"
+ assertEquals(stmt.class, com.mysql.cj.jdbc.ServerPreparedStatement)
+ assertEquals('string_origin', get_prepared_string_key_value(stmt,
'key1'))
+
+ nprep_sql "ALTER TABLE
${realDb}.tbl_point_query_meta_change_string_key RENAME
tbl_point_query_meta_change_string_key_renamed"
+
create_string_key_meta_change_table("${realDb}.tbl_point_query_meta_change_string_key")
+ sql "INSERT INTO ${realDb}.tbl_point_query_meta_change_string_key
VALUES ('key1', 'string_renamed')"
+ assertEquals('string_renamed', get_prepared_string_key_value(stmt,
'key1'))
+ assertEquals('string_renamed', get_prepared_string_key_value(stmt,
'key1'))
+ stmt.close()
+ }
+
sql """ADMIN SET FRONTEND CONFIG ("enable_lightweight_lookup_request"
= "true")"""
try {
connectToDoris(user, password, prepare_url) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
