This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 07009d25633 branch-4.1: [fix](sqlcache) Do not replay the MySQL sql
cache on an Arrow Flight connection #67381 (#67556)
07009d25633 is described below
commit 07009d25633aaecc82bd24f4d047b63542a46e7f
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Sep 8 11:09:14 2026 +0800
branch-4.1: [fix](sqlcache) Do not replay the MySQL sql cache on an Arrow
Flight connection #67381 (#67556)
Cherry-picked from #67381
### Backport notes
One conflict, in `SessionVariable`: branch-4.1 still declares session
variables with
`@VariableMgr.VarAttr` rather than `@VarAttrDef.VarAttr`, so
`RETURN_OBJECT_DATA_AS_BINARY` keeps
that annotation here. The `affectQueryResultInExecution` attribute
exists on this branch's
annotation and `NereidsSqlCacheManager.usedVariablesChanged` compares
the same
`SqlCacheContext.computeAffectQueryResultVariables()` string, so the
behaviour is identical.
`ConnectProcessor`, `StmtExecutor` and both regression suites applied
cleanly.
### Local verification
`./build.sh --fe` on this branch: **BUILD SUCCESS**, no errors,
checkstyle clean on every module (`fe-common` and `fe-core` included).
Regression suites were not run locally.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01DzUYFcGHQH3bnLGCjpncVj
---
.../java/org/apache/doris/qe/ConnectProcessor.java | 10 +-
.../java/org/apache/doris/qe/SessionVariable.java | 7 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 6 +
.../test_sql_cache_over_arrow_flight.groovy | 168 +++++++++++++++++++++
.../query_p0/cache/sql_cache_object_type.groovy | 107 +++++++++++++
5 files changed, 296 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
index 90e0f6e8a62..fa6f646f5e8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
@@ -269,7 +269,15 @@ public abstract class ConnectProcessor {
ctx.setSqlHash(sqlHash);
SessionVariable sessionVariable = ctx.getSessionVariable();
- boolean wantToParseSqlFromSqlCache =
CacheAnalyzer.canUseSqlCache(sessionVariable);
+ // The sql cache keeps the result rows in MySQL wire format and
replays them through a
+ // MysqlChannel (StmtExecutor.sendCachedValues -> sendFields), which
only exists on a MySQL
+ // connection. An Arrow Flight SQL connection has no channel and needs
Arrow batches built by
+ // the BE, and the cached rows would be wrong for it anyway (object
types such as HLL /
+ // BITMAP / QUANTILE_STATE were serialized as NULL under
return_object_data_as_binary=false).
+ // So a non-MySQL connection must always re-execute the query instead
of replaying the cache.
+ // The cache is never populated by such a connection either, see
StmtExecutor.handleQueryStmt.
+ boolean wantToParseSqlFromSqlCache =
connectType.equals(ConnectType.MYSQL)
+ && CacheAnalyzer.canUseSqlCache(sessionVariable);
List<StatementBase> stmts = null;
long parseSqlStartTime = System.currentTimeMillis();
List<StatementBase> cachedStmts = null;
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 7bcb664f9e6..a4fe894e927 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
@@ -2000,7 +2000,12 @@ public class SessionVariable implements Serializable,
Writable {
@VariableMgr.VarAttr(name = ENABLE_INFER_PREDICATE)
private boolean enableInferPredicate = true;
- @VariableMgr.VarAttr(name = RETURN_OBJECT_DATA_AS_BINARY)
+ // Forwarded to the BE as a query option and read by the MySQL result
writer: when it is false
+ // the object types (HLL / BITMAP / QUANTILE_STATE) are serialized as NULL
instead of their raw
+ // bytes. It therefore changes the result rows the sql cache stores, and
must take part in the
+ // cache key, otherwise a session that turns it on replays the NULLs
cached by a session that
+ // had it off. It only affects execution, not the plan, so it does not
force forwarding.
+ @VariableMgr.VarAttr(name = RETURN_OBJECT_DATA_AS_BINARY,
affectQueryResultInExecution = true)
private boolean returnObjectDataAsBinary = false;
@VariableMgr.VarAttr(name = BLOCK_ENCRYPTION_MODE, affectQueryResultInPlan
= true)
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 0384ddd07da..b1a80513b43 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
@@ -1352,6 +1352,12 @@ public class StmtExecutor {
LogicalPlanAdapter logicalPlanAdapter = (LogicalPlanAdapter)
parsedStmt;
LogicalPlan logicalPlan = logicalPlanAdapter.getLogicalPlan();
if (logicalPlan instanceof
org.apache.doris.nereids.trees.plans.algebra.SqlCache) {
+ // sendCachedValues replays MySQL protocol packets, so it
needs a MysqlChannel.
+ // ConnectProcessor.executeQuery only looks the sql cache up
for a MySQL connection,
+ // so a cached plan must never reach another protocol here.
+ Preconditions.checkState(channel != null,
+ "sql cache can only be replayed on a MySQL connection,
but connect type is %s",
+ context.getConnectType());
NereidsPlanner nereidsPlanner = (NereidsPlanner) planner;
PhysicalSqlCache physicalSqlCache = (PhysicalSqlCache)
nereidsPlanner.getPhysicalPlan();
sendCachedValues(channel, physicalSqlCache.getCacheValues(),
logicalPlanAdapter, false, true);
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_sql_cache_over_arrow_flight.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_sql_cache_over_arrow_flight.groovy
new file mode 100644
index 00000000000..577665b275b
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_sql_cache_over_arrow_flight.groovy
@@ -0,0 +1,168 @@
+// 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 org.apache.doris.regression.util.JdbcUtils
+
+// Regression for https://github.com/apache/doris/issues/67364
+//
+// The FE sql cache is shared by every protocol, but its rows are MySQL wire
protocol packets that
+// StmtExecutor.sendCachedValues replays through a MysqlChannel, and an Arrow
Flight SQL connection
+// has none. Replaying an entry created by an identical MySQL query used to
fail
+// Preconditions.checkState(connectType == MYSQL) in StmtExecutor.sendFields()
and reach the client
+// as "INTERNAL ... IllegalStateException, msg: null", for any result type.
The issue was reported
+// on raw HLL / QUANTILE_STATE columns only because that sql text happened to
be the one primed
+// through the MySQL control session.
+//
+// Two setup details decide whether this test can reproduce the bug at all --
get either wrong and
+// it stays green on a broken FE:
+//
+// 1. The flight statements are sent on the raw flight connection.
Suite.arrow_flight_sql()
+// prepends "USE <db>;" to the statement, which changes the sql text and
therefore the cache
+// key (NereidsSqlCacheManager.generateCacheKey is
"<catalog>.<db>:<user>:<sql text>").
+// 2. Both sessions must agree on every session variable the cache compares
+// (NereidsSqlCacheManager.usedVariablesChanged compares the whole
affectQueryResult* set).
+// The MySQL JDBC driver adds STRICT_TRANS_TABLES to sql_mode at connect
time while the Arrow
+// Flight JDBC driver does not, and sql_mode is affectQueryResultInPlan,
so an unaligned
+// sql_mode alone makes every flight lookup miss.
+suite("test_sql_cache_over_arrow_flight") {
+ def mysqlConn = context.getConn()
+ def flightConn = context.getArrowFlightSqlConnection()
+
+ def runOnMysql = { String stmt ->
+ def (result, meta) = JdbcUtils.executeToList(mysqlConn, stmt)
+ return result
+ }
+ def runOnFlight = { String stmt ->
+ def (result, meta) = JdbcUtils.executeToList(flightConn, stmt)
+ return result
+ }
+
+ def hasSqlCache = { String stmt ->
+ def (explainRows, meta) = JdbcUtils.executeToList(mysqlConn, "explain
physical plan " + stmt)
+ return explainRows.collect { row -> row.get(0).toString()
}.join("\n").contains("PhysicalSqlCache")
+ }
+
+ // Create the cache entry on the MySQL connection, and wait until an
identical statement is
+ // actually served from it, so the flight query below really runs against
a populated cache.
+ def primeSqlCacheOnMysql = { String stmt ->
+ for (int i = 0; i < 60; ++i) {
+ runOnMysql(stmt)
+ if (hasSqlCache(stmt)) {
+ return
+ }
+ sleep(1000)
+ }
+ throw new IllegalStateException("failed to create sql cache for: " +
stmt)
+ }
+
+ // JdbcUtils renders a binary column as an "0x.." hex string, but falls
back to the raw object
+ // when the driver does not implement getBytes().
+ def isNonEmptyBinary = { value ->
+ if (value == null) {
+ return false
+ }
+ if (value instanceof byte[]) {
+ return ((byte[]) value).length > 0
+ }
+ return value.toString().length() > "0x".length()
+ }
+
+ withGlobalLock("cache_last_version_interval_second") {
+ runOnMysql "ADMIN SET ALL FRONTENDS CONFIG
('cache_last_version_interval_second' = '0')"
+
+ def dbName = context.dbName
+ runOnMysql "USE `${dbName}`"
+ runOnFlight "USE `${dbName}`"
+ runOnMysql "set enable_sql_cache=true"
+ runOnFlight "set enable_sql_cache=true"
+ // See note 2 above: without this the flight lookup always misses and
the test is toothless.
+ runOnMysql "set sql_mode='ONLY_FULL_GROUP_BY'"
+ runOnFlight "set sql_mode='ONLY_FULL_GROUP_BY'"
+
+ // The cache key is the catalog, the database, the user and the sql
text, and the lookup
+ // additionally compares the session variables that affect the result.
The statements below
+ // are byte identical on both connections, so assert the rest of the
inputs match too.
+ assertEquals(runOnMysql("select database()")[0][0],
runOnFlight("select database()")[0][0])
+ assertEquals(runOnMysql("select current_user()")[0][0],
runOnFlight("select current_user()")[0][0])
+ assertEquals(runOnMysql("select @@sql_mode")[0][0],
runOnFlight("select @@sql_mode")[0][0])
+
+ // 1. A constant result, cached in the FE itself
(PhysicalOneRowRelation.computeResultInFe
+ // -> tryAddFeSqlCache). This replays through the resultSet branch of
sendCachedValues,
+ // needs no table and no quiet window, and is the cheapest way to hit
the bug.
+ def constantSql = "select 1 as c, 'x' as s"
+ primeSqlCacheOnMysql(constantSql)
+ def constantOnFlight = runOnFlight(constantSql)
+ assertEquals(1, constantOnFlight.size())
+ assertEquals(1, constantOnFlight[0][0] as int)
+ assertEquals("x", constantOnFlight[0][1].toString())
+
+ def tblName = "test_sql_cache_over_arrow_flight_tbl"
+ runOnMysql "DROP TABLE IF EXISTS ${tblName}"
+ runOnMysql """
+ CREATE TABLE ${tblName} (
+ k INT,
+ h HLL HLL_UNION,
+ q QUANTILE_STATE QUANTILE_UNION
+ ) AGGREGATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES("replication_num"="1")
+ """
+ runOnMysql "INSERT INTO ${tblName} SELECT 1, HLL_HASH('x'),
TO_QUANTILE_STATE(1, 2048)"
+
+ // 2. A plain scalar result read from a table, cached on the BE. The
failure was protocol
+ // specific, not type specific.
+ def scalarSql = "select k from ${tblName} order by k"
+ primeSqlCacheOnMysql(scalarSql)
+ def scalarOnFlight = runOnFlight(scalarSql)
+ assertEquals(1, scalarOnFlight.size())
+ assertEquals(1, scalarOnFlight[0][0] as int)
+
+ // 3. The raw aggregate state columns from the issue. HLL and
QUANTILE_STATE are carried as
+ // arrow binary (be/src/format/arrow/arrow_row_batch.cpp), so flight
returns the serialized
+ // state, while the MySQL protocol keeps showing NULL under
+ // return_object_data_as_binary=false. Asserting both at once also
proves the flight result
+ // is produced by the BE rather than replayed from the MySQL rows
sitting in the cache.
+ def rawStateSql = "select h, q from ${tblName}"
+ primeSqlCacheOnMysql(rawStateSql)
+ def rawStateOnMysql = runOnMysql(rawStateSql)
+ assertEquals(1, rawStateOnMysql.size())
+ assertNull(rawStateOnMysql[0][0])
+ assertNull(rawStateOnMysql[0][1])
+ def rawStateOnFlight = runOnFlight(rawStateSql)
+ assertEquals(1, rawStateOnFlight.size())
+ assertTrue(isNonEmptyBinary(rawStateOnFlight[0][0]),
+ "expect a non empty HLL state over arrow flight, but got: " +
rawStateOnFlight[0][0])
+ assertTrue(isNonEmptyBinary(rawStateOnFlight[0][1]),
+ "expect a non empty QUANTILE_STATE over arrow flight, but got:
" + rawStateOnFlight[0][1])
+
+ // 4. The server side conversions the issue used as a workaround.
+ def convertedSql = "select hll_cardinality(h) as c,
quantile_percent(q, 0.5) as p from ${tblName}"
+ primeSqlCacheOnMysql(convertedSql)
+ def convertedOnFlight = runOnFlight(convertedSql)
+ assertEquals(1, convertedOnFlight.size())
+ assertEquals(1L, convertedOnFlight[0][0] as long)
+ assertEquals(1.0d, convertedOnFlight[0][1] as double, 1e-9)
+
+ // The flight queries must not have consumed the cache: a cached plan
reaching a non MySQL
+ // connection is exactly the crash this test guards against, and the
entries must still be
+ // there for the MySQL session afterwards.
+ assertTrue(hasSqlCache(constantSql))
+ assertTrue(hasSqlCache(scalarSql))
+ assertTrue(hasSqlCache(rawStateSql))
+ assertTrue(hasSqlCache(convertedSql))
+ }
+}
diff --git a/regression-test/suites/query_p0/cache/sql_cache_object_type.groovy
b/regression-test/suites/query_p0/cache/sql_cache_object_type.groovy
new file mode 100644
index 00000000000..6300840d20f
--- /dev/null
+++ b/regression-test/suites/query_p0/cache/sql_cache_object_type.groovy
@@ -0,0 +1,107 @@
+// 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 org.apache.doris.regression.util.JdbcUtils
+
+// return_object_data_as_binary decides whether the BE's MySQL result writer
serializes HLL /
+// BITMAP / QUANTILE_STATE as their raw bytes or as NULL, so it changes the
very rows the sql cache
+// stores. It must therefore take part in the cache key comparison
+// (NereidsSqlCacheManager.usedVariablesChanged over
SessionVariable.affectQueryResultFields),
+// otherwise a session that turns it on replays the NULLs cached by a session
that had it off.
+suite("sql_cache_object_type") {
+ def conn = context.getConn()
+ def run = { String stmt ->
+ def (result, meta) = JdbcUtils.executeToList(conn, stmt)
+ return result
+ }
+ def hasSqlCache = { String stmt ->
+ def (rows, meta) = JdbcUtils.executeToList(conn, "explain physical
plan " + stmt)
+ return rows.collect { row -> row.get(0).toString()
}.join("\n").contains("PhysicalSqlCache")
+ }
+ def primeSqlCache = { String stmt ->
+ for (int i = 0; i < 60; ++i) {
+ run(stmt)
+ if (hasSqlCache(stmt)) {
+ return
+ }
+ sleep(1000)
+ }
+ throw new IllegalStateException("failed to create sql cache for: " +
stmt)
+ }
+ def isNonEmpty = { value ->
+ if (value == null) {
+ return false
+ }
+ if (value instanceof byte[]) {
+ return ((byte[]) value).length > 0
+ }
+ return !value.toString().isEmpty()
+ }
+
+ withGlobalLock("cache_last_version_interval_second") {
+ run "ADMIN SET ALL FRONTENDS CONFIG
('cache_last_version_interval_second' = '0')"
+ run "set enable_sql_cache=true"
+
+ def tblName = "sql_cache_object_type_tbl"
+ run "DROP TABLE IF EXISTS ${tblName}"
+ run """
+ CREATE TABLE ${tblName} (
+ k INT,
+ h HLL HLL_UNION,
+ b BITMAP BITMAP_UNION
+ ) AGGREGATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES("replication_num"="1")
+ """
+ run "INSERT INTO ${tblName} SELECT 1, HLL_HASH('x'), TO_BITMAP(1)"
+
+ def objectSql = "select h, b from ${tblName}"
+
+ // With the default (false) the object columns come back as NULL, and
that is what lands in
+ // the cache.
+ run "set return_object_data_as_binary=false"
+ primeSqlCache(objectSql)
+ def asNull = run(objectSql)
+ assertEquals(1, asNull.size())
+ assertNull(asNull[0][0])
+ assertNull(asNull[0][1])
+
+ // Turning it on must not be served the cached NULLs: it is a
different result, so it is a
+ // different cache key and the query has to be executed again.
+ run "set return_object_data_as_binary=true"
+ assertFalse(hasSqlCache(objectSql),
+ "return_object_data_as_binary=true must not reuse the entry
cached with it off")
+ def asBinary = run(objectSql)
+ assertEquals(1, asBinary.size())
+ assertTrue(isNonEmpty(asBinary[0][0]),
+ "expect the raw HLL bytes, but got: " + asBinary[0][0])
+ assertTrue(isNonEmpty(asBinary[0][1]),
+ "expect the raw BITMAP bytes, but got: " + asBinary[0][1])
+
+ // The two settings keep their own entries, and each still serves its
own result.
+ primeSqlCache(objectSql)
+ def asBinaryCached = run(objectSql)
+ assertTrue(isNonEmpty(asBinaryCached[0][0]))
+ assertTrue(isNonEmpty(asBinaryCached[0][1]))
+
+ run "set return_object_data_as_binary=false"
+ assertTrue(hasSqlCache(objectSql))
+ def asNullAgain = run(objectSql)
+ assertNull(asNullAgain[0][0])
+ assertNull(asNullAgain[0][1])
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
