github-actions[bot] commented on code in PR #67966:
URL: https://github.com/apache/doris/pull/67966#discussion_r4011202246
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java:
##########
@@ -1034,15 +1046,28 @@ public long getFlightSqlDeferredExecutorsIdleTimeoutS()
{
? ((FlightProtocolAdapter)
protocolAdapter).getDeferredExecutorsIdleTimeoutS() : -1;
}
+ // When the oldest deferred query started; -1 when the connection has
nothing deferred.
+ public long getFlightSqlDeferredExecutorsStartTimeMs() {
+ return protocolAdapter instanceof FlightProtocolAdapter
+ ? ((FlightProtocolAdapter)
protocolAdapter).getDeferredExecutorsStartTimeMs() : -1;
+ }
+
// Called by the timeout checker for a sleeping connection that is not
past wait_timeout yet.
- private void reapIdleFlightSqlDeferredExecutors(long idleMs) {
+ // The bound counts from when the deferred query started, not from
startTime: the session's
+ // later commands (a session option, a metadata request) move startTime on
without finishing
+ // that query, and must not keep its coordinator alive either.
+ private void reapIdleFlightSqlDeferredExecutors(long now) {
long timeoutS = getFlightSqlDeferredExecutorsIdleTimeoutS();
- if (timeoutS < 0 || idleMs <= timeoutS * 1000L) {
+ if (timeoutS < 0) {
+ return;
+ }
+ long deferredMs = now - getFlightSqlDeferredExecutorsStartTimeMs();
Review Comment:
[P1] Reap each executor against its own frozen deadline
This combines `min(start_i)` with `max(timeout_i)` and then closes the whole
current list. A multi-statement Flight request can retain multiple
remote-result executors, while each `StmtExecutor.executeByNereids()` resets
the start time and each statement can freeze a different `SET_VAR` timeout, so
this aggregate can close a later executor before its own deadline. The three
operations are also separate snapshots: the checker can read old A's timeout,
race with `beginRequest()` clearing A, see start `-1`, and then close newly
added B using that stale decision. Please atomically select/remove exact
expired executors, pairing each start with its own `max(config, execution
timeout)`, then finalize that snapshot outside the lock; cover both unequal
per-statement lifetimes and the checker-versus-`beginRequest` interleaving.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1088,13 +1104,27 @@ public int getDeferredExecTimeoutS() {
return deferredExecTimeoutS;
}
+ // The query id the deferred query runs under; null when the query is not
deferred.
+ public TUniqueId getDeferredQueryId() {
+ return deferredQueryId;
+ }
+
+ // When the deferred query started, in epoch milliseconds; -1 when the
query is not deferred.
+ public long getDeferredStartTimeMs() {
+ return deferredStartTimeMs;
+ }
+
// Keep this query's coordinator alive past GetFlightInfo (see the gate in
executeAndSendResult)
// and hand it to the ConnectContext, which finalizes it later. Records
the execution timeout in
// effect right now: it floors the idle reaper's bound and must be the
value the query actually
- // ran with, not the session value left behind after SET_VAR hints are
reverted.
+ // ran with, not the session value left behind after SET_VAR hints are
reverted. Records the
+ // query id and the start time for the same reason: a statement the
session runs in the meantime
+ // replaces both on the context, and the query is finalized under its own.
void deferForArrowFlight() {
deferredForArrowFlight = true;
deferredExecTimeoutS = context.getExecTimeoutS();
+ deferredQueryId = context.queryId();
Review Comment:
[P2] Freeze the deferred query's profiling state too
Only the ID and start time are frozen here. A profiled query has already
published `updateProfile(false)` when it is deferred, but a later
`SetSessionOptions` request deliberately retains this executor and may set
`enable_profile=false` on the same context. `finalizeQuery()` then calls
`updateProfile(true)`, which rechecks that new value and returns, leaving the
old profile unfinished. With profiling still enabled, it also rebuilds the
final summary from later mutable catalog/database/backend-selection and related
state. Please snapshot the deferred query's profiling decision and query-scoped
summary inputs, or retain the running summary and update only terminal fields,
and test finalization after intervening option changes.
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSessionOptions.java:
##########
@@ -0,0 +1,295 @@
+// 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.arrowflight;
+
+import org.apache.doris.analysis.SetType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.util.Util;
+import org.apache.doris.nereids.util.SqlLiteralUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.QueryState.MysqlStateType;
+import org.apache.doris.qe.SqlModeHelper;
+import org.apache.doris.qe.VarAttrDef;
+import org.apache.doris.qe.VariableMgr;
+import org.apache.doris.qe.VariableMgr.VarContext;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.arrow.flight.SessionOptionValue;
+import org.apache.arrow.flight.SessionOptionValueFactory;
+import org.apache.arrow.flight.SessionOptionValueVisitor;
+import org.apache.arrow.flight.SetSessionOptionsResult;
+import org.apache.arrow.flight.SetSessionOptionsResult.ErrorValue;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The session options of an Arrow Flight SQL session, as the
SetSessionOptions and GetSessionOptions
+ * actions see them.
+ *
+ * <p>An option stands for the statement of the session that sets it: {@code
catalog} is the current
+ * catalog ({@code SWITCH}), {@code schema} the current database ({@code
USE}), and any other name is
+ * the session variable {@code SHOW VARIABLES} lists under that name ({@code
SET SESSION}). Those two
+ * names are what the ADBC Flight SQL driver puts on the wire for {@code
adbc.connection.catalog} and
+ * {@code adbc.connection.db_schema}, and what the Flight SQL JDBC driver
sends for its
+ * {@code catalog} property; every other option name the drivers pass through
as given. Setting an
+ * option runs its statement as a command of the session, so it is checked,
audited and takes effect
+ * exactly as if the client had sent the statement. Reading the options back
gives what
+ * {@code SHOW VARIABLES} shows, every value as a string: the one
representation {@code SET} accepts
+ * back, whatever the Java type of the variable behind it. The names an option
goes by are exactly
+ * the names that are read back, spelled the same: a client looks an option up
in the result of
+ * GetSessionOptions by the name it set it under.
+ *
+ * <p>The result of setting an option is one of the three {@link ErrorValue}s
per name and nothing
+ * else, so the reason a value was refused only reaches the frontend log.
+ */
+public final class FlightSessionOptions {
+ private static final Logger LOG =
LogManager.getLogger(FlightSessionOptions.class);
+
+ /** The current catalog; set with {@code SWITCH}. */
+ public static final String CATALOG = "catalog";
+ /** The current database; set with {@code USE}. */
+ public static final String SCHEMA = "schema";
+
+ private FlightSessionOptions() {
+ }
+
+ /**
+ * Sets the options of one request, each on its own, and returns the error
of every option that
+ * could not be set; an option absent from the result was set. {@code
catalog} goes first and
+ * {@code schema} second, since {@code USE} names a database of the
current catalog; the others
+ * follow in name order, so that a request applies the same way every
time. Runs as a command of
+ * the session, under its command lock.
+ */
+ public static Map<String, SetSessionOptionsResult.Error>
set(ConnectContext ctx,
+ Map<String, SessionOptionValue> options) {
+ List<String> names = new ArrayList<>(options.keySet());
+ Collections.sort(names);
+ names.remove(SCHEMA);
+ names.remove(CATALOG);
+ if (options.containsKey(SCHEMA)) {
+ names.add(0, SCHEMA);
+ }
+ if (options.containsKey(CATALOG)) {
+ names.add(0, CATALOG);
+ }
+ Map<String, SetSessionOptionsResult.Error> errors = new
LinkedHashMap<>();
+ for (String name : names) {
+ ErrorValue error = setOne(ctx, name, options.get(name));
+ if (error != null) {
+ errors.put(name, new SetSessionOptionsResult.Error(error));
+ }
+ }
+ return errors;
+ }
+
+ /** Sets one option and returns why it could not be, or null when it was
set. */
+ @VisibleForTesting
+ static ErrorValue setOne(ConnectContext ctx, String name,
SessionOptionValue value) {
+ if (CATALOG.equals(name)) {
+ return switchCatalog(ctx, value);
+ }
+ if (SCHEMA.equals(name)) {
+ return useDatabase(ctx, value);
+ }
+ return setVariable(ctx, name, value);
+ }
+
+ /**
+ * The options of the session: the current catalog, the current database
(an empty string when
+ * none has been chosen) and every session variable {@code SHOW VARIABLES}
would list, in its text.
+ */
+ public static Map<String, SessionOptionValue> get(ConnectContext ctx) {
+ Map<String, SessionOptionValue> options = new LinkedHashMap<>();
+ options.put(CATALOG,
SessionOptionValueFactory.makeSessionOptionValue(ctx.getDefaultCatalog()));
+ String database = ctx.getDatabase();
+ options.put(SCHEMA,
SessionOptionValueFactory.makeSessionOptionValue(database == null ? "" :
database));
+ for (List<String> row : VariableMgr.dump(SetType.SESSION,
ctx.getSessionVariable(), null)) {
+ // A row is name, value, default value, changed.
+ options.put(row.get(0),
SessionOptionValueFactory.makeSessionOptionValue(row.get(1)));
+ }
+ return options;
+ }
+
+ private static ErrorValue switchCatalog(ConnectContext ctx,
SessionOptionValue value) {
+ String catalog = value.acceptVisitor(STRING_VALUE);
+ if (catalog == null || catalog.isEmpty()) {
+ return ErrorValue.INVALID_VALUE;
+ }
+ // SWITCH checks the name's format before anything else and its error
code would not reach
+ // here (see runStatement); the check is one of the value alone, so it
is made here first.
+ try {
+ Util.checkCatalogAllRules(catalog);
+ } catch (AnalysisException e) {
+ LOG.warn("session option {} of Arrow Flight SQL connection {}
could not be set, catalog name {} "
+ + "is malformed: {}", CATALOG, ctx.getConnectionId(),
catalog, e.getMessage());
+ return ErrorValue.INVALID_VALUE;
+ }
+ return runStatement(ctx, CATALOG, "SWITCH " +
quoteIdentifier(catalog), ErrorCode.ERR_UNKNOWN_CATALOG);
+ }
+
+ private static ErrorValue useDatabase(ConnectContext ctx,
SessionOptionValue value) {
+ String database = value.acceptVisitor(STRING_VALUE);
+ if (database == null || database.isEmpty()) {
Review Comment:
[P2] Let clients clear the selected schema
After selecting a database, this API cannot restore Doris's valid initial
no-database state: Flight's valueless option is the conventional unset request,
and `GetSessionOptions` itself represents that state as `schema = ""`, but both
forms are rejected here. This also breaks the generic ADBC option-erase path
and prevents a GET value from being set back. Please handle schema clear by
restoring `currentDb` to the supported empty state, and cover both a valueless
erase and the empty-string round trip.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
