This is an automated email from the ASF dual-hosted git repository.
morningman 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 1250443134f [fix](sqlcache) Do not replay the MySQL sql cache on an
Arrow Flight connection (#67381)
1250443134f is described below
commit 1250443134f3d0baad2f6fb6f233b80d02d4b0ac
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Sep 2 18:09:09 2026 +0800
[fix](sqlcache) Do not replay the MySQL sql cache on an Arrow Flight
connection (#67381)
### What problem does this PR solve?
Issue Number: close #67364
Related PR: #65182, #65615
Problem Summary:
Reading anything over Arrow Flight SQL fails with `INTERNAL` /
`IllegalStateException` as soon as the query hits the FE sql cache.
The FE sql cache is keyed by `<catalog>.<db>:<user>:<sql text>`
(`NereidsSqlCacheManager.generateCacheKey`) and is shared by every
protocol, but its rows are MySQL wire protocol packets that
`StmtExecutor.sendCachedValues` replays through a `MysqlChannel`. An
Arrow Flight SQL connection has no channel, so an entry created by an
identical MySQL query makes `handleQueryStmt` take the cached-plan
branch and fail `Preconditions.checkState(connectType == MYSQL)` in
`sendFields()`. The client sees:
```
INTERNAL: get flight info statement failed, after executeQueryStatement
handleQuery,
error code: ERR_UNKNOWN_ERROR, error msg: IllegalStateException, msg: null
```
The `CacheAnalyzer` branch right below it is already gated on `channel
!= null` with a `// TODO support arrow flight sql`; only this
cached-plan replay was left unguarded.
**This is not about `HLL` / `QUANTILE_STATE`.** The issue was reported
on raw aggregate-state columns, but both the FE schema helper and the BE
map `HLL` / `BITMAP` / `QUANTILE_STATE` to Arrow `binary` and have Arrow
writers for them, and they read back correctly once the query is
actually executed. They only looked special because that sql text was
the one primed through the MySQL control session; `select 1` fails
exactly the same way.
Two conditions have to line up, which is why this is not seen more
often:
1. The same sql text must have been run on a MySQL connection first (an
Arrow Flight connection never populates the cache).
2. Both sessions must agree on every session variable the cache compares
(`NereidsSqlCacheManager.usedVariablesChanged` over the
`affectQueryResult*` set). The MySQL **JDBC driver** adds
`STRICT_TRANS_TABLES` to `sql_mode` at connect time while the Arrow
Flight JDBC driver does not, so a JDBC control session masks the bug --
a `mysql` CLI session, or any client that leaves `sql_mode` alone, does
not.
### What is changed
- `ConnectProcessor.executeQuery`: look the sql cache up only for a
MySQL connection. Any other protocol re-executes the query and gets its
result from the BE. Such a connection never populates the cache either,
so this only removes a broken read path; it does not change MySQL
behaviour.
- `StmtExecutor.handleQueryStmt`: assert the channel in the cached-plan
branch, so a future regression names the protocol instead of throwing a
bare `IllegalStateException`.
- `SessionVariable`: mark `return_object_data_as_binary` as
`affectQueryResultInExecution`. It is forwarded to the BE and decides
whether the MySQL result writer serializes `HLL` / `BITMAP` /
`QUANTILE_STATE` as their raw bytes or as NULL, so it changes the cached
rows and must take part in the cache key comparison. Without it, a
session that turns it on is served the NULLs cached by a session that
had it off -- a separate, pure-MySQL-protocol correctness bug in the
same family.
---
.../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 ed0cda0ab1f..4fe85154210 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
@@ -280,7 +280,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 c5b503e9f14..d9ace832267 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
@@ -1992,7 +1992,12 @@ public class SessionVariable implements Serializable,
Writable {
@VarAttrDef.VarAttr(name = GLOBAL_PARTITION_TOPN_THRESHOLD)
private double globalPartitionTopNThreshold = 100;
- @VarAttrDef.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.
+ @VarAttrDef.VarAttr(name = RETURN_OBJECT_DATA_AS_BINARY,
affectQueryResultInExecution = true)
private boolean returnObjectDataAsBinary = false;
@VarAttrDef.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 797140bd408..a9a38a884d5 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
@@ -1455,6 +1455,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]
