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 0796543f6a0 [fix](arrow-flight) Forward a statement to the master FE
without touching the MySQL channel (#67569)
0796543f6a0 is described below
commit 0796543f6a025a16948d507ecc161761da574d58
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Sep 9 09:52:43 2026 +0800
[fix](arrow-flight) Forward a statement to the master FE without touching
the MySQL channel (#67569)
### What problem does this PR solve?
Related PR: #61050
Problem Summary:
**1. A Flight session could not forward a statement at all.**
An Arrow Flight SQL session has no `MysqlChannel`:
`FlightSqlConnectContext` overrides
`getMysqlChannel()` to throw `getMysqlChannel not in mysql connection`.
Since #61050,
`FEOpExecutor.buildStmtForwardParams()` reads the client's
`CLIENT_DEPRECATE_EOF` capability
straight off that channel, so on a multi-FE deployment every statement a
Flight connection has to
forward to the master fails with that message before the request is even
sent:
- any DDL issued over Arrow Flight SQL to a follower or observer FE;
- any statement at all when `force_forward_all_queries` is on.
`CLIENT_DEPRECATE_EOF` is a MySQL protocol capability, so only read it
for a MySQL connection. The
thrift field is `optional` with no IDL default, and leaving it unset for
other protocols puts the
master back on the packet layout it used before #61050 —
`ConnectProcessor` applies it only when
`isSetClientDeprecatedEOF() && isClientDeprecatedEOF()`, so unset is
equivalent to set-false for
every reader in every rolling-upgrade direction.
**2. Once the forward succeeds, nothing carries the master's answer back
to the Flight client.**
`ConnectProcessor.finalizeCommand()` is the only place a forwarded
statement's status and result set
are replayed, and it is MySQL-only: it opens with
`Preconditions.checkState(connectType.equals(ConnectType.MYSQL))`.
Nothing else reads
`getProxyStatusCode()` / `getShowResultSet()` / `getOutputPacket()`
except the audit log. So after
the RPC, `ctx.getState()` stays at the `OK` that `executeQuery()` set
with `reset()`, the
`FlightSqlChannel` stays empty, and `DorisFlightSqlProducer` answers
with `addOKResult()`'s
synthesized one-row `StatusResult = 0`. With a client connected over
Arrow Flight SQL to a follower
or observer:
- a DDL that **failed** on the master was reported to the client as
success — only the follower's
audit event recorded the real error;
- a forwarded statement that returns rows (`SHOW FRONTENDS`, `SHOW
LOAD`, … — about 33
`ShowCommand` subclasses both forward and return rows) returned that
single `StatusResult` row
instead of the master's rows. The same statement run against the master
returns real rows, because
`StmtExecutor.sendResultSet()` already has a working Arrow Flight
branch.
This gap is older than #61050 (`finalizeCommand()` was already
MySQL-only before it), so fixing only
part 1 would have turned a loud, diagnosable error into a silent wrong
answer, which the
repository's "Error Means Failure" invariant does not allow. Part 2 is
therefore fixed here too:
- `ConnectProcessor.carryForwardedOutcomeToFlightSession()` is the Arrow
Flight counterpart of
`finalizeCommand()`'s forwarded-statement branch. It copies a non-zero
master status into
`ctx.getState()` (and logs it at WARN — the follower used to say
nothing), and otherwise replays
the master's result set through the existing `sendResultSet()` Arrow
Flight branch.
- It is called from the branch of `executeQuery()` that is **already**
scoped to
`ARROW_FLIGHT_SQL`, so it structurally cannot fire on a MySQL connection
and
`finalizeCommand()`'s `getStateType() != ERR` condition is untouched.
- `StmtExecutor` now refuses to forward a *query* on an Arrow Flight SQL
connection, before the RPC.
A query result comes back as MySQL wire packets in
`TMasterOpResult.queryResultBufList`, which
only `finalizeCommand()` can replay and which cannot be converted to
Arrow batches; answering it
with a synthesized empty success would be the same silent wrong answer.
This is reachable only
under `force_forward_all_queries`, which `Config` documents as "For
testing purposes".
The MySQL path is byte-for-byte unchanged.
---
.../java/org/apache/doris/qe/ConnectProcessor.java | 33 ++++++
.../java/org/apache/doris/qe/FEOpExecutor.java | 8 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 9 ++
.../ConnectProcessorFlightForwardOutcomeTest.java | 121 +++++++++++++++++++++
.../doris/qe/FEOpExecutorFlightForwardTest.java | 120 ++++++++++++++++++++
.../test_arrow_flight_forward_to_master.groovy | 97 +++++++++++++++++
6 files changed, 387 insertions(+), 1 deletion(-)
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 55f5d2c9c6c..18b6a4d64ad 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
@@ -86,6 +86,7 @@ import org.apache.doris.thrift.TMasterOpResult;
import org.apache.doris.thrift.TUniqueId;
import org.apache.doris.transaction.TransactionEntry;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
@@ -382,6 +383,9 @@ public abstract class ConnectProcessor {
}
}
} else if
(connectType.equals(ConnectType.ARROW_FLIGHT_SQL)) {
+ if (executor.hasForwardedToMaster()) {
+ carryForwardedOutcomeToFlightSession(executor);
+ }
if (!ctx.isReturnResultFromLocal()) {
returnResultFromRemoteExecutor.add(executor);
}
@@ -685,6 +689,35 @@ public abstract class ConnectProcessor {
LOG.debug("End finalizing command for query {}",
DebugUtil.printId(ctx.queryId));
}
+ // Arrow Flight SQL counterpart of the forwarded-statement branch of
finalizeCommand(). That
+ // method is the only place a forwarded statement's status and result set
are replayed to the
+ // client, and it is MySQL-only: it opens with
+ // Preconditions.checkState(connectType.equals(ConnectType.MYSQL)).
Without this method a
+ // forwarded statement leaves ctx.getState() at the OK that executeQuery()
set with reset() and
+ // leaves the FlightSqlChannel empty, so DorisFlightSqlProducer answers
with addOKResult()'s
+ // synthesized StatusResult=0 -- reporting success for a statement that
failed on the master,
+ // and an empty status row instead of the rows a forwarded SHOW produced.
+ @VisibleForTesting
+ void carryForwardedOutcomeToFlightSession(StmtExecutor executor) throws
IOException {
+ if (executor.getProxyStatusCode() != 0) {
+ // The master rejected the statement, e.g. CREATE TABLE on a table
that already exists.
+ // TMasterOpResult carries the master's error code as a plain int
and ErrorCode has no
+ // reverse lookup, so the master's code travels in the message
instead.
+ String errMsg = "forwarded statement failed on master FE, error
code: "
+ + executor.getProxyStatusCode() + ", error message: " +
executor.getProxyErrMsg();
+ LOG.warn(errMsg);
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, errMsg);
+ return;
+ }
+ // Set exactly when the forwarded statement produced rows:
proxyExecute() fills
+ // TMasterOpResult.resultSet from getProxyShowResultSet(). A forwarded
DDL produces none,
+ // and the synthesized StatusResult=0 is the right answer for it.
+ ShowResultSet resultSet = executor.getShowResultSet();
+ if (resultSet != null) {
+ executor.sendResultSet(resultSet);
+ }
+ }
+
public TMasterOpResult proxyExecute(TMasterOpRequest request) throws
TException {
ctx.setDatabase(request.db);
ctx.setEnv(Env.getCurrentEnv());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
index cb1f6d5da9e..414a9725c70 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
@@ -26,6 +26,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.datasource.DelegatedCredential;
import org.apache.doris.mysql.MysqlCommand;
+import org.apache.doris.qe.ConnectContext.ConnectType;
import org.apache.doris.thrift.FrontendService;
import org.apache.doris.thrift.TExpr;
import org.apache.doris.thrift.TExprNode;
@@ -225,7 +226,12 @@ public class FEOpExecutor {
// Propagate the client's CLIENT_DEPRECATE_EOF capability so the
master FE
// generates packets matching the original client's protocol
expectations.
-
params.setClientDeprecatedEOF(ctx.getMysqlChannel().clientDeprecatedEOF());
+ // Only a MySQL connection negotiates this capability and owns a
MysqlChannel;
+ // an Arrow Flight SQL session has none, and leaving the field unset
keeps the
+ // master on its default packet layout.
+ if (ctx.getConnectType() == ConnectType.MYSQL) {
+
params.setClientDeprecatedEOF(ctx.getMysqlChannel().clientDeprecatedEOF());
+ }
return params;
}
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 396ec17a08c..ae4bc830fd8 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
@@ -915,6 +915,15 @@ public class StmtExecutor {
if (context.getCommand() == MysqlCommand.COM_STMT_PREPARE) {
throw new UserException("Forward master command is not
supported for prepare statement");
}
+ if (context.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) {
+ // The master returns a query result as MySQL wire packets
in
+ // TMasterOpResult.queryResultBufList, which only
ConnectProcessor.finalizeCommand()
+ // can replay and which cannot be converted to Arrow
batches. Refuse here, before
+ // the RPC, rather than let the master build a result set
this FE would discard
+ // and answer the client with a synthesized empty success.
+ throw new UserException("Forwarding a query to the master
FE is not supported on an"
+ + " Arrow Flight SQL connection. Connect to the
master FE to run this query.");
+ }
if (isProxy) {
// This is already a stmt forwarded from other FE.
// If we goes here, means we can't find a valid Master
FE(some error happens).
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorFlightForwardOutcomeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorFlightForwardOutcomeTest.java
new file mode 100644
index 00000000000..e7824798cc5
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/ConnectProcessorFlightForwardOutcomeTest.java
@@ -0,0 +1,121 @@
+// 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.qe;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.service.arrowflight.sessions.FlightSqlConnectContext;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+
+/**
+ * ConnectProcessor.finalizeCommand() is the only place a forwarded
statement's status and result set
+ * are replayed to the client, and it is MySQL-only (it opens with a
Preconditions.checkState on the
+ * connect type). carryForwardedOutcomeToFlightSession() is its Arrow Flight
SQL counterpart. Without
+ * it, ctx.getState() stays at the OK that executeQuery() set with reset() and
the FlightSqlChannel
+ * stays empty, so DorisFlightSqlProducer answers with addOKResult()'s
synthesized StatusResult=0:
+ * success for a statement that failed on the master, and that same row
instead of the rows a
+ * forwarded SHOW produced.
+ */
+public class ConnectProcessorFlightForwardOutcomeTest {
+ private boolean savedRunningUnitTest;
+ private FlightSqlConnectContext context;
+ private ConnectProcessor processor;
+
+ @BeforeEach
+ public void setUp() {
+ savedRunningUnitTest = FeConstants.runningUnitTest;
+ // ConnectContext.init() registers the session with Env unless running
as a unit test.
+ FeConstants.runningUnitTest = true;
+ context = new FlightSqlConnectContext("test-peer-identity");
+ processor = new TestConnectProcessor(context);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ FeConstants.runningUnitTest = savedRunningUnitTest;
+ }
+
+ @Test
+ public void testMasterFailureIsReportedToTheFlightClient() throws
Exception {
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ // e.g. CREATE TABLE on a table that already exists.
+ Mockito.when(executor.getProxyStatusCode()).thenReturn(1050);
+ Mockito.when(executor.getProxyErrMsg()).thenReturn("Table 'tbl'
already exists");
+
+ processor.carryForwardedOutcomeToFlightSession(executor);
+
+ Assertions.assertEquals(QueryState.MysqlStateType.ERR,
context.getState().getStateType());
+ Assertions.assertEquals(ErrorCode.ERR_UNKNOWN_ERROR,
context.getState().getErrorCode());
+ // The master's own error code has no ErrorCode enum on this side, so
it travels in the text.
+
Assertions.assertTrue(context.getState().getErrorMessage().contains("1050"),
+ context.getState().getErrorMessage());
+
Assertions.assertTrue(context.getState().getErrorMessage().contains("Table
'tbl' already exists"),
+ context.getState().getErrorMessage());
+ // A statement that failed must not also replay a result set.
+ Mockito.verify(executor, Mockito.never()).sendResultSet(Mockito.any());
+ }
+
+ @Test
+ public void testForwardedResultSetIsReplayedToTheFlightChannel() throws
Exception {
+ ShowResultSet resultSet = new ShowResultSet(
+ ShowResultSetMetaData.builder()
+ .addColumn(new Column("JobId",
ScalarType.createVarchar(20)))
+ .build(),
+ Lists.<List<String>>newArrayList(Lists.newArrayList("10086")));
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Mockito.when(executor.getProxyStatusCode()).thenReturn(0);
+ Mockito.when(executor.getShowResultSet()).thenReturn(resultSet);
+
+ processor.carryForwardedOutcomeToFlightSession(executor);
+
+ Assertions.assertNotEquals(QueryState.MysqlStateType.ERR,
context.getState().getStateType());
+ // sendResultSet()'s ARROW_FLIGHT_SQL branch puts the rows into the
FlightSqlChannel, which is
+ // what DorisFlightSqlProducer hands back instead of the synthesized
StatusResult row.
+ Mockito.verify(executor).sendResultSet(resultSet);
+ }
+
+ @Test
+ public void testForwardedDdlKeepsTheSynthesizedOkResult() throws Exception
{
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+ Mockito.when(executor.getProxyStatusCode()).thenReturn(0);
+ // A forwarded DDL carries no result set, and StatusResult=0 is the
right answer for it.
+ Mockito.when(executor.getShowResultSet()).thenReturn(null);
+
+ processor.carryForwardedOutcomeToFlightSession(executor);
+
+ Assertions.assertNotEquals(QueryState.MysqlStateType.ERR,
context.getState().getStateType());
+ Assertions.assertEquals(0L, context.getFlightSqlChannel().resultNum());
+ Mockito.verify(executor, Mockito.never()).sendResultSet(Mockito.any());
+ }
+
+ private static class TestConnectProcessor extends ConnectProcessor {
+ private TestConnectProcessor(ConnectContext context) {
+ super(context);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorFlightForwardTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorFlightForwardTest.java
new file mode 100644
index 00000000000..af39b6f936d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/FEOpExecutorFlightForwardTest.java
@@ -0,0 +1,120 @@
+// 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.qe;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.service.arrowflight.sessions.FlightSqlConnectContext;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TMasterOpRequest;
+import org.apache.doris.thrift.TNetworkAddress;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+/**
+ * Building the forward request for the master FE must not reach for a
MysqlChannel unconditionally:
+ * an Arrow Flight SQL session has none, so every statement a Flight
connection forwards (any DDL on
+ * a non-master FE, or anything at all under force_forward_all_queries) used
to fail with
+ * "getMysqlChannel not in mysql connection". CLIENT_DEPRECATE_EOF is a MySQL
protocol capability and
+ * is only carried for MySQL connections.
+ */
+public class FEOpExecutorFlightForwardTest {
+ private boolean savedRunningUnitTest;
+
+ @BeforeEach
+ public void setUp() {
+ savedRunningUnitTest = FeConstants.runningUnitTest;
+ // ConnectContext.init() registers the session with Env unless running
as a unit test.
+ FeConstants.runningUnitTest = true;
+ }
+
+ @AfterEach
+ public void tearDown() {
+ FeConstants.runningUnitTest = savedRunningUnitTest;
+ }
+
+ @Test
+ public void testFlightSessionForwardsWithoutMysqlChannel() throws
Exception {
+ try (MockedStatic<Env> mockedEnv = mockSelfNode()) {
+ FlightSqlConnectContext context = new
FlightSqlConnectContext("test-peer-identity");
+ prepare(context);
+
+ TMasterOpRequest request = new TestFEOpExecutor(context).build();
+
+ Assertions.assertFalse(request.isSetClientDeprecatedEOF());
+ Assertions.assertEquals("select 1", request.getSql());
+ }
+ }
+
+ @Test
+ public void testMysqlSessionKeepsCarryingDeprecatedEof() throws Exception {
+ try (MockedStatic<Env> mockedEnv = mockSelfNode()) {
+ ConnectContext context = new ConnectContext();
+ prepare(context);
+ context.getMysqlChannel().setClientDeprecatedEOF();
+
+ TMasterOpRequest request = new TestFEOpExecutor(context).build();
+
+ Assertions.assertTrue(request.isSetClientDeprecatedEOF());
+ Assertions.assertTrue(request.isClientDeprecatedEOF());
+ }
+ }
+
+ @Test
+ public void testMysqlSessionWithoutDeprecatedEof() throws Exception {
+ try (MockedStatic<Env> mockedEnv = mockSelfNode()) {
+ ConnectContext context = new ConnectContext();
+ prepare(context);
+
+ TMasterOpRequest request = new TestFEOpExecutor(context).build();
+
+ Assertions.assertTrue(request.isSetClientDeprecatedEOF());
+ Assertions.assertFalse(request.isClientDeprecatedEOF());
+ }
+ }
+
+ private static MockedStatic<Env> mockSelfNode() {
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getSelfNode()).thenReturn(new
SystemInfoService.HostInfo("127.0.0.1", 9010));
+ MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class);
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ return mockedEnv;
+ }
+
+ private static void prepare(ConnectContext context) {
+
context.setCurrentUserIdentity(UserIdentity.createAnalyzedUserIdentWithIp("alice",
"%"));
+ context.setRemoteIP("127.0.0.1");
+ }
+
+ private static class TestFEOpExecutor extends FEOpExecutor {
+ private TestFEOpExecutor(ConnectContext context) {
+ super(new TNetworkAddress("127.0.0.1", 9010), new
OriginStatement("select 1", 0), context, true);
+ }
+
+ private TMasterOpRequest build() throws AnalysisException {
+ return buildStmtForwardParams();
+ }
+ }
+}
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_forward_to_master.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_forward_to_master.groovy
new file mode 100644
index 00000000000..7927272a428
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_forward_to_master.groovy
@@ -0,0 +1,97 @@
+// 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.suite.ClusterOptions
+
+// A statement issued over Arrow Flight SQL to a non-master FE has to be
forwarded to the master.
+// This needs more than one FE, so it can only be exercised in a docker
cluster: on a single-FE
+// deployment StmtExecutor.shouldForwardToMaster() returns false immediately.
+//
+// The group is 'docker' and not 'arrow_flight_sql' on purpose.
SuiteContext.useArrowFlightSql()
+// keys off the suite group, so keeping it out means `sql` resolves to the
thread-local connection
+// this suite opens itself (the follower's Flight port) instead of the
globally configured one.
+suite("test_arrow_flight_forward_to_master", "docker") {
+ def options = new ClusterOptions()
+ options.setFeNum(2)
+ options.connectToFollower = true
+
+ docker(options) {
+ def follower = cluster.getOneFollowerFe()
+ assertNotNull(follower, "a follower FE is required to exercise the
forward-to-master path")
+
+ // Every docker FE serves Arrow Flight SQL on the same fixed port, see
+ // FE_ARROW_FLIGHT_SQL_PORT in docker/runtime/doris-compose/cluster.py.
+ Class.forName("org.apache.arrow.driver.jdbc.ArrowFlightJdbcDriver")
+ def flightUrl =
"jdbc:arrow-flight-sql://${follower.host}:8070/catalog=${context.dbName}" +
+ "?useServerPrepStmts=false&useSSL=false&useEncryption=false"
+ logger.info("connect to follower over arrow flight sql:
${flightUrl}".toString())
+
+ sql "DROP TABLE IF EXISTS test_arrow_flight_forward_to_master"
+
+ connect('root', '', flightUrl) {
+ // 1. A DDL is a Redirect command, so on a follower it is
forwarded to the master.
+ // Before the CLIENT_DEPRECATE_EOF capability was read only for
MySQL connections
+ // this failed with "getMysqlChannel not in mysql connection".
+ sql """
+ CREATE TABLE test_arrow_flight_forward_to_master (k1 int)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+
+ // 2. The same DDL again fails on the master. The follower replays
a forwarded
+ // statement's outcome only in
ConnectProcessor.finalizeCommand(), which is
+ // MySQL-only, so without
carryForwardedOutcomeToFlightSession() the master's error
+ // was dropped and the client was told StatusResult=0 --
success for a statement
+ // that failed.
+ test {
+ sql """
+ CREATE TABLE test_arrow_flight_forward_to_master (k1 int)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ exception "already exists"
+ }
+
+ // 3. SHOW FRONTENDS is forwarded too (forward_to_master defaults
to true) and it
+ // carries a result set. Without the fix the master's rows were
dropped and the
+ // client got the synthesized single column named StatusResult
instead.
+ def frontends = sql_return_maparray "SHOW FRONTENDS"
+ assertFalse(frontends.isEmpty(), "SHOW FRONTENDS returned nothing")
+ assertFalse(frontends[0].containsKey("StatusResult"),
+ "got the synthesized status row instead of the master's
result set: " + frontends[0])
+ assertTrue(frontends.size() >= 2, "expected both FEs, got " +
frontends.size())
+ assertEquals(1, frontends.count { it.IsMaster == "true" })
+ }
+
+ // The forwarded DDL really took effect on the master.
CreateTableCommand forwards with
+ // sync, so by now the follower has replayed it too.
+ def tables = sql_return_maparray "SHOW TABLES LIKE
'test_arrow_flight_forward_to_master'"
+ assertEquals(1, tables.size())
+
+ // 4. A forwarded *query* returns its result as MySQL wire packets,
which cannot be turned
+ // into Arrow batches. That must be refused explicitly rather than
answered with the
+ // synthesized empty success. Use a fresh Flight session so
force_forward_all_queries
+ // cannot leak into the assertions above.
+ connect('root', '', flightUrl) {
+ sql "SET force_forward_all_queries = true"
+ test {
+ sql "SELECT k1 FROM test_arrow_flight_forward_to_master"
+ exception "not supported on an Arrow Flight SQL connection"
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
