[refactor](session) Replace the execution layer's ConnectType branches with protocol capabilities

fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java

@@ -197,13 +197,11 @@ private FlightInfo executeQueryStatement(String peerIdentity, ConnectContext con
         try {
             Preconditions.checkState(null != connectContext);
             Preconditions.checkState(!query.isEmpty());
-            // Finalize the previous query's coordinator on this connection whose close was
-            // deferred (Arrow Flight keeps it alive across GetFlightInfo -> DoGet so the BE can
-            // fetch external-table splits during DoGet). By now the previous DoGet is done. #62259
-            connectContext.closeFlightSqlDeferredExecutors();
-            // After the previous query was executed, there was no getStreamStatement to take away the result.
-            connectContext.getFlightSqlChannel().reset();
-            connectContext.clearFlightSqlEndpointsLocations();
+            // Drops what the previous request left on the session: its deferred coordinator (Arrow
+            // Flight keeps it alive across GetFlightInfo -> DoGet so the BE can fetch external-table
+            // splits during DoGet, and by now that DoGet is done, #62259), a result no
+            // getStreamStatement took away, and its endpoints.
+            FlightProtocolAdapter.of(connectContext).beginRequest();
             try (FlightSqlConnectProcessor flightSQLConnectProcessor = new FlightSqlConnectProcessor(connectContext)) {
                 flightSQLConnectProcessor.handleQuery(query);
                 if (connectContext.getState().getStateType() == MysqlStateType.ERR) {

fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlConnectProcessor.java

@@ -64,7 +64,6 @@ public class FlightSqlConnectProcessor extends ConnectProcessor implements AutoC
     public FlightSqlConnectProcessor(ConnectContext context) {
         super(context);
         context.setThreadLocalInfo();
-        context.setReturnResultFromLocal(true);
     }
 
     public Schema getArrowSchema() {

fe/fe-core/src/main/java/org/apache/doris/arrowflight/protocol/FlightProtocolAdapter.java

@@ -30,6 +30,7 @@
 import org.apache.doris.qe.ShowResultSet;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.qe.protocol.ProtocolAdapter;
+import org.apache.doris.thrift.TMasterOpRequest;
 import org.apache.doris.thrift.TResultSinkType;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -66,6 +67,10 @@ public class FlightProtocolAdapter implements ProtocolAdapter {
     private final Map<String, String> preparedQuerys = new HashMap<>();
     private String runningQuery;
     private final List<FlightSqlEndpointsLocation> endpointsLocations = Lists.newArrayList();
+    // Whether the result of the statement being executed is on this frontend (a SHOW, a SET, an
+    // EXPLAIN: cached on the channel for the client's DoGet) or on the backends the coordinator
+    // ran the query on, registered in endpointsLocations for the client to pull from. Set by the
+    // statement lifecycle hooks below.
     private boolean returnResultFromLocal = true;
     // Executors of already-planned queries whose results are produced on the BE and pulled later
     // during the DoGet phase. Their coordinators must stay alive until the BE finishes scanning:
@@ -129,6 +134,80 @@ public boolean supportsSqlCacheReplay() {
         return false;
     }
 
+    /**
+     * The master returns a query result as MySQL wire packets, which cannot be turned into the
+     * Arrow batches a Flight client needs. The executor refuses to forward a query rather than
+     * let the master build a result set this frontend would discard and answer the client with a
+     * synthesized empty success.
+     */
+    @Override
+    public boolean canReplayForwardedQueryResult() {
+        return false;
+    }
+
+    /**
+     * A result this frontend materializes is cached with every column as a Utf8 vector, whatever
+     * its type ({@link FlightResultSender}). That is acceptable for the text a SHOW or an EXPLAIN
+     * produces, not for a SELECT a client expects typed Arrow data from, so a query the planner
+     * could answer here is run on a backend until the sender types its vectors.
+     */
+    @Override
+    public boolean supportsFeSideResult() {
+        return false;
+    }
+
+    /**
+     * The short circuit produces no Arrow result at either end. PointQueryExecutor is not a
+     * Coordinator, and Coordinator/NereidsCoordinator are the only places that register a
+     * FlightSqlEndpointsLocation, so GetFlightInfo found none and failed the query with
+     * "no FlightSqlEndpointsLocations"; the backend side cannot be pointed at either, since the
+     * lookup rpc serializes with VMysqlResultWriter into PTabletKeyLookupResponse.row_batch and
+     * never creates the ArrowFlightResultBlockBuffer that fetch_arrow_flight_schema looks up.
+     * Arrow Flight SQL stays on the normal execution path. See #67368.
+     */
+    @Override
+    public boolean supportsShortCircuitPointQuery() {
+        return false;
+    }
+
+    /**
+     * A Flight session does not retry a failed query: the backend endpoints the failed attempt
+     * registered would have to be withdrawn first, and nothing does that yet.
+     */
+    @Override
+    public boolean canRetryQuery(ConnectContext ctx) {
+        return false;
+    }
+
+    /** A statement's result is on this frontend until {@link #beforeQuery} says otherwise. */
+    @Override
+    public void beforeStatement(ConnectContext ctx) {
+        returnResultFromLocal = true;
+    }
+
+    /**
+     * The query's result stays on the backends for the client to pull with DoGet; the
+     * coordinator registers where ({@link #addEndpointsLocation}) instead of fetching the rows.
+     */
+    @Override
+    public void beforeQuery(ConnectContext ctx) {
+        returnResultFromLocal = false;
+    }
+
+    @Override
+    public boolean returnsResultFromLocal(ConnectContext ctx) {
+        return returnResultFromLocal;
+    }
+
+    /**
+     * The master's response is consumed here as a status and, for a SHOW, a result set (see
+     * {@link #carryForwardedOutcome}); it is never replayed to the client as packets, so the
+     * master needs to know nothing about the client.
+     */
+    @Override
+    public void fillForwardRequest(ConnectContext ctx, TMasterOpRequest request) {
+    }
+
     @Override
     public ConnectPoolMgr connectPool(ConnectScheduler scheduler) {
         return scheduler.getFlightSqlConnectPoolMgr();
@@ -232,16 +311,18 @@ public List<FlightSqlEndpointsLocation> getEndpointsLocations() {
         return endpointsLocations;
     }
 
-    public void clearEndpointsLocations() {
+    /**
+     * Starts a request of the session: whatever the previous request left behind is dropped.
+     * Its query's coordinator, if its close was deferred, is finalized now -- the previous DoGet
+     * is done by the time the next request arrives (#62259); the result it may have cached and
+     * never pulled with DoGet is released; its endpoints are forgotten; and the new request's
+     * result is on this frontend until a query is run for it.
+     */
+    public void beginRequest() {
+        closeDeferredExecutors();
+        channel.reset();
         endpointsLocations.clear();
-    }
-
-    public void setReturnResultFromLocal(boolean returnResultFromLocal) {
-        this.returnResultFromLocal = returnResultFromLocal;
-    }
-
-    public boolean isReturnResultFromLocal() {
-        return returnResultFromLocal;
+        returnResultFromLocal = true;
     }
 
     public void addDeferredExecutor(StmtExecutor executor) {

fe/fe-core/src/main/java/org/apache/doris/arrowflight/protocol/FlightResultSender.java

@@ -49,10 +49,6 @@ public class FlightResultSender implements ResultSender {
     @Override
     public void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos, boolean binaryRows) {
         adapter.getChannel().addResult(DebugUtil.printId(ctx.queryId()), adapter.getRunningQuery(), resultSet);
-        // The statement's result is on this frontend, whatever the query path decided earlier: an
-        // EXPLAIN goes through the query path, which marks the result as coming from the backend
-        // before it knows the statement will not run there.
-        adapter.setReturnResultFromLocal(true);
     }
 
     @Override
@@ -70,6 +66,6 @@ public void sendRow(ByteBuffer row) {
     @Override
     public void reset() {
         // Results are cached per query id and the cache is cleared when the next request of the
-        // session starts (DorisFlightSqlProducer.executeQueryStatement); nothing is pending here.
+        // session starts (FlightProtocolAdapter.beginRequest); nothing is pending here.
     }
 }

fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlProtocolAdapter.java

@@ -20,13 +20,15 @@
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.mysql.MysqlCapability;
 import org.apache.doris.mysql.MysqlChannel;
+import org.apache.doris.mysql.MysqlCommand;
 import org.apache.doris.mysql.MysqlCursorFetchCompatibility;
 import org.apache.doris.mysql.MysqlHandshakePacket;
 import org.apache.doris.mysql.MysqlPacket;
 import org.apache.doris.mysql.MysqlResultSetEndPacket;
 import org.apache.doris.mysql.MysqlSerializer;
 import org.apache.doris.mysql.MysqlServerStatusFlag;
 import org.apache.doris.mysql.MysqlSslContext;
+import org.apache.doris.mysql.ProxyMysqlChannel;
 import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.stats.StatsErrorEstimator;
 import org.apache.doris.qe.ConnectContext;
@@ -38,13 +40,15 @@
 import org.apache.doris.qe.ShowResultSet;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.qe.protocol.ProtocolAdapter;
+import org.apache.doris.thrift.TMasterOpRequest;
 import org.apache.doris.thrift.TResultSinkType;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.List;
 
 /**
  * The MySQL protocol side of a connection: the channel to the client, the capabilities negotiated
@@ -111,6 +115,100 @@ public boolean supportsSqlCacheReplay() {
         return true;
     }
 
+    @Override
+    public boolean canReplayForwardedQueryResult() {
+        return true;
+    }
+
+    @Override
+    public boolean supportsFeSideResult() {
+        return true;
+    }
+
+    @Override
+    public boolean supportsShortCircuitPointQuery() {
+        return true;
+    }
+
+    /**
+     * The packets of a failed attempt are dropped when the next attempt resets the channel, so a
+     * retry is invisible to the client as long as none of them was flushed to the socket yet.
+     */
+    @Override
+    public boolean canRetryQuery(ConnectContext ctx) {
+        return !channel.isSend();
+    }
+
+    /**
+     * Clears the send flag and whatever the previous statement of the request left in the send
+     * buffer. Between the statements of a request run for a client that did not negotiate
+     * CLIENT_MULTI_STATEMENTS nothing is flushed (see {@link #finishStatement}), so what a
+     * statement wrote is still in the buffer when the next one starts.
+     */
+    @Override
+    public void beforeStatement(ConnectContext ctx) {
+        channel.reset();
+    }
+
+    @Override
+    public void beforeQuery(ConnectContext ctx) {
+        // The rows are relayed through the channel as the coordinator fetches them.
+    }
+
+    @Override
+    public boolean returnsResultFromLocal(ConnectContext ctx) {
+        return true;
+    }
+
+    /**
+     * The master encodes the response of a forwarded statement for this connection's client, so
+     * it needs the client's negotiated capabilities, and for a forwarded COM_STMT_EXECUTE the
+     * execute packet and whether it asked for a cursor. {@link #restoreFromForwardRequest} is the
+     * master's side.
+     */
+    @Override
+    public void fillForwardRequest(ConnectContext ctx, TMasterOpRequest request) {
+        if (ctx.getCommand() == MysqlCommand.COM_STMT_EXECUTE) {
+            if (prepareExecuteBuffer != null) {
+                request.setPrepareExecuteBuffer(prepareExecuteBuffer);
+            }
+            request.setCursorFetchRequested(cursorFetchRequested);
+        }
+        request.setClientDeprecatedEOF(channel.clientDeprecatedEOF());
+        request.setMysqlCapability(capability.getFlags());
+    }
+
+    /**
+     * On the master, gives the proxy context of a forwarded statement the capabilities of the
+     * client it is answering, as {@link #fillForwardRequest} put them in the request. A request
+     * from an old frontend carries neither the capability flags nor the cursor flag; it gets the
+     * default capabilities without CLIENT_DEPRECATE_EOF, plus that flag when set separately, and
+     * its ordinary prepared statements are not rejected.
+     */
+    public void restoreFromForwardRequest(ConnectContext ctx, TMasterOpRequest request) {
+        int flags = request.isSetMysqlCapability() ? request.getMysqlCapability()
+                : MysqlCapability.DEFAULT_CAPABILITY.getFlags()
+                        & ~MysqlCapability.Flag.CLIENT_DEPRECATE_EOF.getFlagBit();
+        if (request.isSetClientDeprecatedEOF() && request.isClientDeprecatedEOF()) {
+            flags |= MysqlCapability.Flag.CLIENT_DEPRECATE_EOF.getFlagBit();
+        }
+        MysqlCapability restored = new MysqlCapability(flags);
+        setCapability(restored);
+        channel.getSerializer().setCapability(restored);
+        if (restored.isDeprecatedEOF()) {
+            channel.setClientDeprecatedEOF();
+        }
+        cursorFetchRequested = request.isSetCursorFetchRequested() && request.isCursorFetchRequested();
+    }
+
+    /**
+     * On the master, the packets the forwarded statement produced, collected by the proxy
+     * context's channel to be handed back to the frontend the client is connected to.
+     */
+    public List<ByteBuffer> proxyResultPackets() {
+        return ((ProxyMysqlChannel) channel).getProxyResultBufferList();
+    }
+
     @Override
     public ConnectPoolMgr connectPool(ConnectScheduler scheduler) {
         return scheduler.getConnectPoolMgr();
@@ -120,9 +218,10 @@ public ConnectPoolMgr connectPool(ConnectScheduler scheduler) {
      * Between the statements of a multi-statement request the intermediate response carries
      * SERVER_MORE_RESULTS_EXISTS, and is sent right away if the client negotiated
      * CLIENT_MULTI_STATEMENTS. Here Doris differs from MySQL: a client that did not negotiate it
-     * gets the request run as several statements anyway, but only the last result is delivered
-     * (the next query resets the channel, see {@link MysqlResultSender#reset}). The response of the
-     * last statement is the response of the command, sent by {@link #finishCommand}.
+     * gets the request run as several statements anyway, but only the last statement's outcome
+     * is delivered (the next statement resets the channel, see {@link #beforeStatement}). The
+     * response of the last statement is the response of the command, sent by
+     * {@link #finishCommand}.
      */
     @Override
     public boolean finishStatement(ConnectContext ctx, StmtExecutor executor, int stmtIndex, int stmtCount)

fe/fe-core/src/main/java/org/apache/doris/mysql/protocol/MysqlResultSender.java

@@ -138,7 +138,7 @@ public void sendRow(ByteBuffer row) throws IOException {
         channel().sendOnePacket(row);
     }
 
-    /** Clears the send flag and whatever the previous statement left in the send buffer. */
+    /** Clears the send flag and whatever a failed attempt of the query left in the send buffer. */
     @Override
     public void reset() {
         channel().reset();

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/LogicalResultSinkToShortCircuitPointQuery.java

@@ -31,7 +31,6 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.qe.ConnectContext.ConnectType;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -69,16 +68,12 @@ boolean scanMatchShortCircuitCondition(LogicalOlapScan olapScan) {
         if (!connectContext.getSessionVariable().isEnableShortCircuitQuery()) {
             return false;
         }
-        // The short circuit produces no Arrow result at either end. PointQueryExecutor is not a
-        // Coordinator, and Coordinator/NereidsCoordinator are the only places that register a
-        // FlightSqlEndpointsLocation, so GetFlightInfo found none and failed the query with
-        // "no FlightSqlEndpointsLocations"; the BE side cannot be pointed at either, since the lookup rpc
-        // serializes with VMysqlResultWriter into PTabletKeyLookupResponse.row_batch and never creates the
-        // ArrowFlightResultBlockBuffer that fetch_arrow_flight_schema looks up. Keep Arrow Flight SQL on
-        // the normal execution path. This has to be decided here at plan time rather than when picking the
-        // executor: OlapScanNode.computeTabletInfo and several rewrite and property rules read
-        // StatementContext.isShortCircuitQuery() while building the plan. See #67368.
-        if (connectContext.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) {
+        // A protocol whose client pulls the result from the backend has no result to pull for a
+        // short circuit (see FlightProtocolAdapter.supportsShortCircuitPointQuery). This has to be
+        // decided here at plan time rather than when picking the executor: OlapScanNode.computeTabletInfo
+        // and several rewrite and property rules read StatementContext.isShortCircuitQuery() while
+        // building the plan. See #67368.
+        if (!connectContext.getProtocolAdapter().supportsShortCircuitPointQuery()) {
             return false;
         }
         // Lazy point-query pruning does not preserve explicit PARTITION/TABLET restrictions.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/StartTransactionCommand.java

@@ -32,9 +32,6 @@ public StartTransactionCommand() {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
-        if (ctx.getConnectType() == ConnectContext.ConnectType.MYSQL) {
-            ctx.getMysqlChannel().reset();
-        }
         // do nothing
         ctx.getState().setOk();
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/TransactionBeginCommand.java

@@ -64,10 +64,6 @@ private void validate(ConnectContext ctx) {
     }
 
     private void handleTransactionBegin(ConnectContext ctx) {
-        if (ctx.getConnectType() == ConnectContext.ConnectType.MYSQL) {
-            // Every time set no send flag and clean all data in buffer
-            ctx.getMysqlChannel().reset();
-        }
         ctx.getState().setOk(0, 0, "");
 
         if (ctx.isTxnModel()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/TransactionCommitCommand.java

@@ -45,10 +45,6 @@ public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
     }
 
     private void handleTransactionCommit(ConnectContext ctx) throws AnalysisException {
-        if (ctx.getConnectType() == ConnectContext.ConnectType.MYSQL) {
-            // Every time set no send flag and clean all data in buffer
-            ctx.getMysqlChannel().reset();
-        }
         ctx.getState().setOk(0, 0, "");
 
         if (ctx.getTxnEntry() != null && ctx.getTxnEntry().getRowsInTransaction() == 0