morningman commented on code in PR #67966:
URL: https://github.com/apache/doris/pull/67966#discussion_r4013704584
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/protocol/FlightProtocolAdapter.java:
##########
@@ -58,6 +61,25 @@
* arrive on its own thread, and nothing in the transport serializes them.
{@link #runCommand}
* does, so that a session's {@link ConnectContext}, which is not thread-safe,
is only ever used
* by one command at a time.
+ *
+ * <p>Nor does a query end where the frontend's command does. The client pulls
the results from the
+ * backends by itself (DoGet), and the frontend hears nothing of it -- not
when it starts, not when
+ * it is done -- so a query may still be running on the backends while the
session runs its next
+ * command here. Two things of a request therefore outlive the command that
made them, and the
+ * contract for both is that the command stream owns them and nothing else
does: the result a
+ * statement materialized on this frontend, cached on the channel until the
client's DoGet takes it
+ * or the next request drops it; and the executors of queries whose
coordinator must stay alive
+ * until the backends are done with it, the {@link #deferredExecutors}. A
deferred executor is a
+ * closed object from the moment it is deferred: it carries what finalizing it
needs and reads
+ * nothing of the session's live state afterwards, since the session moves on
without it (a session
+ * option's SET, a metadata request, the next request) and since it may be
finalized from a thread
+ * that runs no command of the session (the timeout checker, a token expiry).
It is finalized
+ * exactly once, by whoever takes it out of the list under the list's lock --
the next request
Review Comment:
Fixed in 674e491f9a9. Teardown now tombstones the adapter:
`FlightProtocolAdapter.tearDown` (reached from `unregisterConnection` for
CloseSession, token expiry, KILL and the timeout checker) takes what the list
holds as before and closes the list under the same monitor. An executor
deferred after that is finalized on the spot by the command that deferred it
(`addDeferredExecutor`), and a command that gets its turn on the lock after
teardown fails with `UNAUTHENTICATED` (`callCommand`) instead of running on a
context no pool or checker sees any more. Latch-based test through the real
KILL path, with the command still holding the lock:
`FlightProtocolAdapterTest.testATornDownSessionFinalizesWhatIsDeferredToItOnTheSpot`.
Two corrections to the framing, for the record. The window was never the
publication instant: teardown landing anywhere between the request's start and
`deferForArrowFlight()` orphaned the executor the same way, on master as here,
because teardown has never serialized with a running command (which is what the
`runCommand` javadoc says since #67900); the tombstone covers all of it. And
teardown *after* publication finalizing the coordinator while the request is
still fetching its Arrow schema is the intended effect of closing a session
with a request in flight -- the request fails, as it should. Serializing
teardown with the command lock is not an option: CloseSession would block
behind a running query for up to the execution timeout, and the token cache's
removal listener cannot block inside a cache operation.
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSessionOptions.java:
##########
@@ -0,0 +1,312 @@
+// 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. And what is read back can be
set back: the empty
+ * value, Flight's way of unsetting an option, puts a variable back to its
default and the session
+ * back into no database, as it started -- the state GetSessionOptions reports
as an empty
+ * {@code schema}, which sets it too. A session is always in some catalog, so
{@code catalog} has no
+ * empty value.
+ *
+ * <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
+ * the session is in none, the value that puts it back there) 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)));
Review Comment:
Not changing this: it is a trade-off, made and documented (class javadoc,
and the "GetSessionOptions" paragraph of the PR description, whose manual-test
log shows the `get_option_int` refusal as the expected consequence), and the
proposed direction breaks the other half of the round trip.
Every getter of the ADBC Flight SQL driver is exact on the variant, the
string one included: `getSessionOption` in `flightsql_connection.go` does
`rawValue.(T)` for `GetOption` ("a string") as much as for `GetOptionInt` ("an
integer"), `GetOptionDouble` and the `optionbool.` prefix. So whichever type
the server returns, one client path fails. With strings, as here:
`SetOptionInt(query_timeout, 77)` then `GetOptionInt` -> `NOT_FOUND ... is not
an integer value`. With native types, as proposed: `SetOption(query_timeout,
"77")` then `GetOption` -> `NOT_FOUND ... is not a string value` -- and
`GetOption` is the default read path of every binding (`get_option` in Python,
`GetOption` in C++ and R). The reference Flight SQL server satisfies both only
because its session options are a key-value echo; Doris's options are typed
variables with one canonical representation, the text `SHOW VARIABLES` shows
and `SET` accepts back, and that is the contract this PR states. The ADBC
documentation
says no more than "get or set a string/numeric session option" for the prefix
and pins no server-side type. The two consumers that read options back today --
`adbc.connection.catalog` / `db_schema` (a string is required) and the
`adbc.flight.sql.session.options` JSON blob -- work either way.
If a typed contract is wanted later (int64 for int/long variables, bool for
boolean ones, double for double, text for the rest), it is a decision to
document and a separate change with its own compatibility note, not a defect of
this PR.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1076,7 +1121,13 @@ public void finalizeQuery() {
// received after unregisterQuery(), causing the instance profile to
be lost, so we should wait
// for the profile before unregisterQuery().
updateProfile(true);
- QeProcessorImpl.INSTANCE.unregisterQuery(context.queryId());
+ QeProcessorImpl.INSTANCE.unregisterQuery(queryId());
+ }
+
+ // The id this query runs under: the context's, until the query is
deferred for Arrow Flight and
+ // the context may move on to another statement before the query is
finalized.
+ private TUniqueId queryId() {
Review Comment:
Two backend-result statements in one request is the case of the thread on
`finishStatement`, and it is refused there now (674e491f9a9): a query that is
not the last statement stops the request the way a frontend-side result always
did, before the next statement runs, so the first executor is finalized in
`close()` while `context.queryId()` is still its own -- no second query id is
generated in that request. `SELECT; SET` is refused the same way, so the SET no
longer replaces the query's id either.
What remains is the fallback to `context.queryId()` for an executor
finalized within its own statement, which is what `finalizeQuery` has always
done for every protocol; the id this PR captures covers the one case the PR
itself opened, a query finalized after the session ran another statement.
Capturing a registration id for every executor is a `StmtExecutor` refactor
beyond this PR.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1297,8 +1366,22 @@ private void forwardToMaster() throws Exception {
}
}
+ /**
+ * Whether this statement's profile is reported, decided the first time it
is asked and the
+ * same ever after (see {@link #profileEnabled}): the update that
publishes the profile as
+ * RUNNING and the one that finishes it must agree, whatever the session
variable says by then.
+ */
+ private boolean reportsProfile() {
+ Boolean decided = profileEnabled;
+ if (decided == null) {
+ decided = context.getSessionVariable().enableProfile() &&
isProfileSafeStmt();
Review Comment:
The failure is real, but it is neither introduced nor reachable through
anything this PR changed, and the freeze is not its cause.
`Profile` is constructed in the `StmtExecutor` constructor from
`enableProfile()` as it stands at construction time (both `StmtExecutor`
constructors). In a multi-statement request the parse applies every statement's
SET_VAR up front (`LogicalPlanBuilder`, `setVarOnceInSql`), the first
executor's `execute()` finally reverts it, and the second executor is
constructed with `Profile(false)`. That is the same on a MySQL connection --
the loop is `ConnectProcessor.executeQuery`, shared -- and master ends the same
way: master's `updateProfile(false)` sees `enableProfile()` true once the hint
is re-applied and then hits the same early return in `Profile.updateSummary`
(`isQueryFinished` starts true for a disabled profile). `reportsProfile()` only
pins the decision so that the RUNNING and the final update agree; what it fixes
is a master bug of deferred Flight queries -- a `SET_VAR(enable_profile=true)`
query published as RUNNING and never finished, because the hint is reverted
before the
query is finalized -- and it can neither cause nor cure the construction-time
one.
Constructing `Profile` from the effective per-statement decision changes the
profile lifecycle of every protocol; that is its own PR, not this one.
--
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]
