You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@kyuubi.apache.org by "wangzhigang1999 (via GitHub)" <gi...@apache.org> on 2026/06/29 08:13:20 UTC
[PR] [WIP][KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
wangzhigang1999 opened a new pull request, #7530:
URL: https://github.com/apache/kyuubi/pull/7530
### Why are the changes needed?
First half of #7458 (the statement hook; the result-set hook is a follow-up PR). Kyuubi lacks a gateway-level, per-statement hook before a statement is routed to the engine (Spark AuthZ is Spark-specific). This adds a server-side `StatementInterceptor` SPI, invoked for each interactive statement before operation creation and routing, that can **inspect / rewrite / reject** — one stable hook for authorization, risky-SQL interception, and rewriting.
- Lives in the dependency-free `kyuubi-server-plugin` module, JDK types only; does not expose session conf/credentials.
- `kyuubi.operation.statement.interceptors` is `serverOnly` (no session bypass); fail-closed on throw/`null`; eager init at startup.
- Context exposes `statementId()` (the client's operation handle id) to correlate a statement with its later operation/result set.
Interactive path only.
### How was this patch tested?
Unit (`StatementInterceptionSuite`, `PluginLoaderSuite`) and end-to-end JDBC→server→engine (`StatementInterceptorSuite`) covering proceed/rewrite/reject, the executeOnServer path, fail-closed, concurrency, and `statementId`. `spotless`/`scalastyle`/`AllKyuubiConfiguration` pass.
### Was this patch authored or co-authored using generative AI tooling?
Assisted-by: Claude Opus 4.8
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "wangzhigang1999 (via GitHub)" <gi...@apache.org>.
wangzhigang1999 commented on code in PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#discussion_r3892347961
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala:
##########
@@ -83,14 +85,20 @@ class ExecuteStatement(
protected def executeStatement(): Unit =
try {
withLocalProperties {
- setState(OperationState.RUNNING)
- info(diagnostics)
- Thread.currentThread().setContextClassLoader(spark.sharedState.jarClassLoader)
- addOperationListener()
- result = spark.sql(statement)
- iter = collectAsIterator(result)
- setCompiledStateIfNeeded()
- setState(OperationState.FINISHED)
+ val operationConf = spark.sessionState.conf.clone()
+ confOverlay.foreach { case (key, value) =>
+ operationConf.setConfString(key, value)
+ }
+ SQLConf.withExistingConf(operationConf) {
Review Comment:
Addressed with an empty-overlay fast path: the execution path now reuses the session SQLConf and only clones it when per-statement overrides are present.
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/plugin/StatementInterception.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.kyuubi.plugin
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.kyuubi.KyuubiSQLException
+
+/** A concrete, immutable [[StatementInterceptContext]] built per interceptor invocation. */
+private[kyuubi] class StatementInterceptContextImpl(
+ override val sessionId: String,
+ override val statementId: String,
+ override val user: String,
+ override val realUser: String,
+ override val ipAddress: String,
+ override val statement: String,
+ override val confOverlay: java.util.Map[String, String],
+ override val runAsync: Boolean,
+ override val queryTimeout: Long,
+ override val engineType: String) extends StatementInterceptContext
+
+private[kyuubi] case class InterceptedStatement(
+ statement: String,
+ confOverlay: Map[String, String])
+
+private[kyuubi] object StatementInterception {
+
+ /**
+ * Initialize interceptors in order. If an interceptor's initialize throws, every interceptor
+ * whose initialization was attempted is closed in reverse order before the error is rethrown, so
+ * that a failed server startup does not leak resources (threads, connections, external clients)
+ * allocated before the failure. Errors raised while closing are attached as suppressed
+ * exceptions.
+ */
+ def initialize(
+ interceptors: Seq[StatementInterceptor],
+ conf: java.util.Map[String, String]): Unit = {
+ val initialized = new scala.collection.mutable.ArrayBuffer[StatementInterceptor]()
+ try {
+ interceptors.foreach { interceptor =>
+ initialized += interceptor
+ interceptor.initialize(conf)
+ }
+ } catch {
+ case initError: Throwable =>
+ initialized.reverseIterator.foreach { interceptor =>
+ try {
+ interceptor.close()
+ } catch {
+ case NonFatal(closeError) => initError.addSuppressed(closeError)
+ }
+ }
+ throw initError
+ }
+ }
+
+ /**
+ * Run the interceptor chain over a statement. Each interceptor sees the statement and config
+ * overlay produced by the previous one; a reject, an exception, or a null result fails the
+ * statement.
+ *
+ * @param interceptors the interceptors in execution order
+ * @param initialStatement the original statement
+ * @param initialConfOverlay the client-provided per-statement config overlay
+ * @param contextFor builds a context for the current statement and config overlay
+ * @return the final statement and config overlay to route to the engine
+ */
+ def run(
+ interceptors: Seq[StatementInterceptor],
+ initialStatement: String,
+ initialConfOverlay: Map[String, String],
+ contextFor: (
+ String,
+ java.util.Map[String, String]) => StatementInterceptContext): InterceptedStatement = {
+ var currentStatement = initialStatement
+ var currentConfOverlay = initialConfOverlay
+ interceptors.foreach { interceptor =>
+ val interceptorName = interceptor.getClass.getName
+ val result =
+ try {
+ interceptor.beforeExecuteStatement(
+ contextFor(currentStatement, currentConfOverlay.asJava))
+ } catch {
+ case e: KyuubiSQLException => throw e
+ case NonFatal(e) =>
+ throw KyuubiSQLException(s"Statement interceptor $interceptorName failed", e)
+ }
+ if (result == null) {
+ throw KyuubiSQLException(s"Statement interceptor $interceptorName returned null")
+ }
+ result.action() match {
+ case StatementInterceptResult.Action.PROCEED => // keep the current statement
+ case StatementInterceptResult.Action.REWRITE => currentStatement = result.statement()
+ case StatementInterceptResult.Action.REJECT =>
+ // SQLState 42501 = "Insufficient Privilege". Use the specific access-rule subclass so
+ // clients can distinguish a policy rejection from the generic syntax-error class 42000.
+ throw KyuubiSQLException(result.message(), sqlState = "42501")
+ }
+ currentConfOverlay ++= result.confOverlay().asScala
Review Comment:
Addressed by assigning the merged immutable map back explicitly. This keeps the accumulated overlay clear and avoids relying on Scala assignment rewriting.
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/PlanOnlyStatement.scala:
##########
@@ -77,7 +78,11 @@ class PlanOnlyStatement(
override protected def runInternal(): Unit =
try {
withLocalProperties {
- SQLConf.withExistingConf(spark.sessionState.conf) {
+ val operationConf = spark.sessionState.conf.clone()
+ confOverlay.foreach { case (key, value) =>
+ operationConf.setConfString(key, value)
+ }
+ SQLConf.withExistingConf(operationConf) {
val parsed = spark.sessionState.sqlParser.parsePlan(statement)
Review Comment:
Addressed with the same empty-overlay fast path in plan-only mode, so the base SQLConf is reused unless an interceptor supplied overrides.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "wangzhigang1999 (via GitHub)" <gi...@apache.org>.
wangzhigang1999 commented on PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#issuecomment-4965129119
> Before taking a detailed look, another common use case, in addition to REWRITE/REJECT, is to tune the config.
Good point. The current SPI exposes the statement-level conf overlay as read-only, so an interceptor cannot tune it.
I am considering letting StatementInterceptResult carry a config-overlay delta. The interceptor chain would accumulate those updates in order, each following interceptor would see the updated overlay, and later values would win. The final overlay would then be passed to the engine operation together with the intercepted statement.
I prefer an explicit result delta over making StatementInterceptContext mutable, since config tuning can also be combined with REWRITE and the chain semantics remain clear.
Does that match the use case you have in mind?
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#discussion_r3892425641
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala:
##########
@@ -35,7 +35,15 @@ import org.apache.kyuubi.shaded.thrift.TException
import org.apache.kyuubi.shaded.thrift.transport.TTransportException
import org.apache.kyuubi.util.ThriftUtils
-abstract class KyuubiOperation(session: Session) extends AbstractOperation(session) {
+// The handle is injected as a constructor paramaccessor `override protected val handle` (mirroring
Review Comment:
Typo in the class comment: “paramaccessor” should be “parameter accessor” for clarity.
##########
extensions/server/kyuubi-server-plugin/src/main/java/org/apache/kyuubi/plugin/StatementInterceptResult.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.kyuubi.plugin;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The decision returned by a {@link StatementInterceptor} for a single statement: proceed with the
+ * current statement, rewrite it for the next interceptor and the engine, or reject it.
+ */
+public final class StatementInterceptResult {
+
+ public enum Action {
+ PROCEED,
+ REWRITE,
+ REJECT
+ }
+
+ private final Action action;
+ private final String statement; // the rewritten statement when action is REWRITE
+ private final String message; // the rejection reason when action is REJECT
+ private final Map<String, String> confOverlay;
+
+ private StatementInterceptResult(
+ Action action, String statement, String message, Map<String, String> confOverlay) {
+ this.action = action;
+ this.statement = statement;
+ this.message = message;
+ this.confOverlay = immutableConfOverlay(confOverlay);
+ }
+
+ /** Keep the current statement and pass it to the next interceptor. */
+ public static StatementInterceptResult proceed() {
+ return proceed(Collections.emptyMap());
+ }
+
+ /**
+ * Keep the current statement and apply {@code confOverlay} to this statement only. The entries
+ * are passed to subsequent interceptors and ultimately to the engine operation.
+ */
+ public static StatementInterceptResult proceed(Map<String, String> confOverlay) {
+ return new StatementInterceptResult(Action.PROCEED, null, null, confOverlay);
+ }
+
+ /**
+ * Replace the current statement with {@code statement}. It is passed to the next interceptor and
+ * ultimately to the engine. {@code statement} must not be null or blank.
+ */
+ public static StatementInterceptResult rewrite(String statement) {
+ return rewrite(statement, Collections.emptyMap());
+ }
+
+ /**
+ * Replace the current statement with {@code statement} and apply {@code confOverlay} to this
+ * statement only. Both are passed to subsequent interceptors and ultimately to the engine
+ * operation. {@code statement} must not be null or blank.
+ */
+ public static StatementInterceptResult rewrite(
+ String statement, Map<String, String> confOverlay) {
+ return new StatementInterceptResult(
+ Action.REWRITE, requireNonBlank(statement, "statement"), null, confOverlay);
+ }
+
+ /**
+ * Reject the statement immediately; the chain stops and the client receives an error. {@code
+ * message} must not be null or blank.
+ */
+ public static StatementInterceptResult reject(String message) {
+ return new StatementInterceptResult(
+ Action.REJECT, null, requireNonBlank(message, "message"), Collections.emptyMap());
+ }
+
+ public Action action() {
+ return action;
+ }
+
+ public String statement() {
+ return statement;
+ }
+
+ public String message() {
+ return message;
+ }
+
+ /** The immutable per-statement configuration updates returned by this interceptor. */
+ public Map<String, String> confOverlay() {
+ return confOverlay;
+ }
+
+ private static String requireNonBlank(String value, String name) {
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException(name + " must not be null or blank");
+ }
+ return value;
+ }
+
+ private static Map<String, String> immutableConfOverlay(Map<String, String> confOverlay) {
+ if (confOverlay == null) {
+ throw new IllegalArgumentException("confOverlay must not be null");
+ }
+ Map<String, String> copy = new HashMap<>();
+ confOverlay.forEach(
+ (key, value) -> {
+ if (key == null || value == null) {
+ throw new IllegalArgumentException("confOverlay must not contain null keys or values");
+ }
+ copy.put(key, value);
+ });
+ return Collections.unmodifiableMap(copy);
+ }
Review Comment:
`immutableConfOverlay` always allocates a new `HashMap` (and wraps it) even when the input overlay is empty. Since `proceed()`/`rewrite()` commonly pass `Collections.emptyMap()`, this adds avoidable per-statement allocation overhead. Add an empty fast-path and pre-size the map for non-empty overlays.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "pan3793 (via GitHub)" <gi...@apache.org>.
pan3793 commented on PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#issuecomment-4965040597
Before taking a detailed look, another common use case, in addition to REWRITE/REJECT, is to tune the config.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#discussion_r3889680479
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/PlanOnlyStatement.scala:
##########
@@ -77,7 +78,11 @@ class PlanOnlyStatement(
override protected def runInternal(): Unit =
try {
withLocalProperties {
- SQLConf.withExistingConf(spark.sessionState.conf) {
+ val operationConf = spark.sessionState.conf.clone()
+ confOverlay.foreach { case (key, value) =>
+ operationConf.setConfString(key, value)
+ }
+ SQLConf.withExistingConf(operationConf) {
val parsed = spark.sessionState.sqlParser.parsePlan(statement)
Review Comment:
This clones `spark.sessionState.conf` on every invocation even when `confOverlay` is empty. Since plan-only mode can be used frequently, avoid the extra clone by reusing the base `SQLConf` when there are no per-statement overrides.
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/plugin/StatementInterception.scala:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.kyuubi.plugin
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import org.apache.kyuubi.KyuubiSQLException
+
+/** A concrete, immutable [[StatementInterceptContext]] built per interceptor invocation. */
+private[kyuubi] class StatementInterceptContextImpl(
+ override val sessionId: String,
+ override val statementId: String,
+ override val user: String,
+ override val realUser: String,
+ override val ipAddress: String,
+ override val statement: String,
+ override val confOverlay: java.util.Map[String, String],
+ override val runAsync: Boolean,
+ override val queryTimeout: Long,
+ override val engineType: String) extends StatementInterceptContext
+
+private[kyuubi] case class InterceptedStatement(
+ statement: String,
+ confOverlay: Map[String, String])
+
+private[kyuubi] object StatementInterception {
+
+ /**
+ * Initialize interceptors in order. If an interceptor's initialize throws, every interceptor
+ * whose initialization was attempted is closed in reverse order before the error is rethrown, so
+ * that a failed server startup does not leak resources (threads, connections, external clients)
+ * allocated before the failure. Errors raised while closing are attached as suppressed
+ * exceptions.
+ */
+ def initialize(
+ interceptors: Seq[StatementInterceptor],
+ conf: java.util.Map[String, String]): Unit = {
+ val initialized = new scala.collection.mutable.ArrayBuffer[StatementInterceptor]()
+ try {
+ interceptors.foreach { interceptor =>
+ initialized += interceptor
+ interceptor.initialize(conf)
+ }
+ } catch {
+ case initError: Throwable =>
+ initialized.reverseIterator.foreach { interceptor =>
+ try {
+ interceptor.close()
+ } catch {
+ case NonFatal(closeError) => initError.addSuppressed(closeError)
+ }
+ }
+ throw initError
+ }
+ }
+
+ /**
+ * Run the interceptor chain over a statement. Each interceptor sees the statement and config
+ * overlay produced by the previous one; a reject, an exception, or a null result fails the
+ * statement.
+ *
+ * @param interceptors the interceptors in execution order
+ * @param initialStatement the original statement
+ * @param initialConfOverlay the client-provided per-statement config overlay
+ * @param contextFor builds a context for the current statement and config overlay
+ * @return the final statement and config overlay to route to the engine
+ */
+ def run(
+ interceptors: Seq[StatementInterceptor],
+ initialStatement: String,
+ initialConfOverlay: Map[String, String],
+ contextFor: (
+ String,
+ java.util.Map[String, String]) => StatementInterceptContext): InterceptedStatement = {
+ var currentStatement = initialStatement
+ var currentConfOverlay = initialConfOverlay
+ interceptors.foreach { interceptor =>
+ val interceptorName = interceptor.getClass.getName
+ val result =
+ try {
+ interceptor.beforeExecuteStatement(
+ contextFor(currentStatement, currentConfOverlay.asJava))
+ } catch {
+ case e: KyuubiSQLException => throw e
+ case NonFatal(e) =>
+ throw KyuubiSQLException(s"Statement interceptor $interceptorName failed", e)
+ }
+ if (result == null) {
+ throw KyuubiSQLException(s"Statement interceptor $interceptorName returned null")
+ }
+ result.action() match {
+ case StatementInterceptResult.Action.PROCEED => // keep the current statement
+ case StatementInterceptResult.Action.REWRITE => currentStatement = result.statement()
+ case StatementInterceptResult.Action.REJECT =>
+ // SQLState 42501 = "Insufficient Privilege". Use the specific access-rule subclass so
+ // clients can distinguish a policy rejection from the generic syntax-error class 42000.
+ throw KyuubiSQLException(result.message(), sqlState = "42501")
+ }
+ currentConfOverlay ++= result.confOverlay().asScala
Review Comment:
`currentConfOverlay` is typed as an immutable `Map[String, String]`, so using `++=` here is not valid and will not compile (the immutable collections API provides `++`, not `++=`). Assign the merged map back to `currentConfOverlay` instead so interceptor config deltas are accumulated correctly.
##########
externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala:
##########
@@ -83,14 +85,20 @@ class ExecuteStatement(
protected def executeStatement(): Unit =
try {
withLocalProperties {
- setState(OperationState.RUNNING)
- info(diagnostics)
- Thread.currentThread().setContextClassLoader(spark.sharedState.jarClassLoader)
- addOperationListener()
- result = spark.sql(statement)
- iter = collectAsIterator(result)
- setCompiledStateIfNeeded()
- setState(OperationState.FINISHED)
+ val operationConf = spark.sessionState.conf.clone()
+ confOverlay.foreach { case (key, value) =>
+ operationConf.setConfString(key, value)
+ }
+ SQLConf.withExistingConf(operationConf) {
Review Comment:
This clones `spark.sessionState.conf` for every statement, even when `confOverlay` is empty (the common case). Cloning `SQLConf` on every execution can add avoidable overhead; consider only cloning when there are actual overrides, and otherwise reuse `spark.sessionState.conf`.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "wangzhigang1999 (via GitHub)" <gi...@apache.org>.
wangzhigang1999 commented on PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#issuecomment-4840614008
**Design note — `statementId`**
`StatementInterceptContext#statementId()` is the operation-handle UUID, **pre-allocated before interception** and reused for the operation (instead of being self-generated afterwards). So one stable id spans intercept time → operation → result set → the handle the client gets. That's the correlation/join key for end-to-end audit and lineage, and for the planned result-set transform hook (second half of #7458). It's threaded down via a defaulted ctor param through `AbstractOperation`/`KyuubiOperation`/`ExecuteStatement`/`ExecutedCommandExec`, so no engine ripple.
Also manually verified on a Kubernetes deployment: REWRITE/REJECT over Beeline (reject surfaces SQLState `42000`), `statementId` matches the operation handle, and neither batch nor a session-level override bypasses the `serverOnly` interceptor.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org
Re: [PR] [KYUUBI #7458] Add server-side statement interceptor SPI [kyuubi]
Posted by "wangzhigang1999 (via GitHub)" <gi...@apache.org>.
wangzhigang1999 commented on code in PR #7530:
URL: https://github.com/apache/kyuubi/pull/7530#discussion_r3892473077
##########
kyuubi-server/src/main/scala/org/apache/kyuubi/operation/KyuubiOperation.scala:
##########
@@ -35,7 +35,15 @@ import org.apache.kyuubi.shaded.thrift.TException
import org.apache.kyuubi.shaded.thrift.transport.TTransportException
import org.apache.kyuubi.util.ThriftUtils
-abstract class KyuubiOperation(session: Session) extends AbstractOperation(session) {
+// The handle is injected as a constructor paramaccessor `override protected val handle` (mirroring
Review Comment:
Fixed the wording to use “parameter accessor” and reflowed the comment for clarity.
##########
extensions/server/kyuubi-server-plugin/src/main/java/org/apache/kyuubi/plugin/StatementInterceptResult.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.kyuubi.plugin;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The decision returned by a {@link StatementInterceptor} for a single statement: proceed with the
+ * current statement, rewrite it for the next interceptor and the engine, or reject it.
+ */
+public final class StatementInterceptResult {
+
+ public enum Action {
+ PROCEED,
+ REWRITE,
+ REJECT
+ }
+
+ private final Action action;
+ private final String statement; // the rewritten statement when action is REWRITE
+ private final String message; // the rejection reason when action is REJECT
+ private final Map<String, String> confOverlay;
+
+ private StatementInterceptResult(
+ Action action, String statement, String message, Map<String, String> confOverlay) {
+ this.action = action;
+ this.statement = statement;
+ this.message = message;
+ this.confOverlay = immutableConfOverlay(confOverlay);
+ }
+
+ /** Keep the current statement and pass it to the next interceptor. */
+ public static StatementInterceptResult proceed() {
+ return proceed(Collections.emptyMap());
+ }
+
+ /**
+ * Keep the current statement and apply {@code confOverlay} to this statement only. The entries
+ * are passed to subsequent interceptors and ultimately to the engine operation.
+ */
+ public static StatementInterceptResult proceed(Map<String, String> confOverlay) {
+ return new StatementInterceptResult(Action.PROCEED, null, null, confOverlay);
+ }
+
+ /**
+ * Replace the current statement with {@code statement}. It is passed to the next interceptor and
+ * ultimately to the engine. {@code statement} must not be null or blank.
+ */
+ public static StatementInterceptResult rewrite(String statement) {
+ return rewrite(statement, Collections.emptyMap());
+ }
+
+ /**
+ * Replace the current statement with {@code statement} and apply {@code confOverlay} to this
+ * statement only. Both are passed to subsequent interceptors and ultimately to the engine
+ * operation. {@code statement} must not be null or blank.
+ */
+ public static StatementInterceptResult rewrite(
+ String statement, Map<String, String> confOverlay) {
+ return new StatementInterceptResult(
+ Action.REWRITE, requireNonBlank(statement, "statement"), null, confOverlay);
+ }
+
+ /**
+ * Reject the statement immediately; the chain stops and the client receives an error. {@code
+ * message} must not be null or blank.
+ */
+ public static StatementInterceptResult reject(String message) {
+ return new StatementInterceptResult(
+ Action.REJECT, null, requireNonBlank(message, "message"), Collections.emptyMap());
+ }
+
+ public Action action() {
+ return action;
+ }
+
+ public String statement() {
+ return statement;
+ }
+
+ public String message() {
+ return message;
+ }
+
+ /** The immutable per-statement configuration updates returned by this interceptor. */
+ public Map<String, String> confOverlay() {
+ return confOverlay;
+ }
+
+ private static String requireNonBlank(String value, String name) {
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException(name + " must not be null or blank");
+ }
+ return value;
+ }
+
+ private static Map<String, String> immutableConfOverlay(Map<String, String> confOverlay) {
+ if (confOverlay == null) {
+ throw new IllegalArgumentException("confOverlay must not be null");
+ }
+ Map<String, String> copy = new HashMap<>();
+ confOverlay.forEach(
+ (key, value) -> {
+ if (key == null || value == null) {
+ throw new IllegalArgumentException("confOverlay must not contain null keys or values");
+ }
+ copy.put(key, value);
+ });
+ return Collections.unmodifiableMap(copy);
+ }
Review Comment:
Addressed with an empty fast path returning `Collections.emptyMap()`. Non-empty overlays now use the `HashMap(Map)` copy constructor before null validation and the unmodifiable wrapper, preserving the defensive immutable-copy contract while avoiding empty-map allocations.
--
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: notifications-unsubscribe@kyuubi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@kyuubi.apache.org
For additional commands, e-mail: notifications-help@kyuubi.apache.org