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
commit 39892d33393b81b2c47f0d6a950c9690958c8d4a 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]
