i7xh commented on a change in pull request #1085:
URL: https://github.com/apache/incubator-kyuubi/pull/1085#discussion_r710190711
##########
File path:
dev/kyuubi-extension-spark-3-1/src/main/scala/org/apache/kyuubi/sql/KyuubiSQLConf.scala
##########
@@ -94,4 +94,13 @@ object KyuubiSQLConf {
.version("1.4.0")
.intConf
.createOptional
+
+ val WATCHDOG_FORCED_MAXOUTPUTROWS =
+ buildConf("spark.sql.watchdog.forcedMaxOutputRows")
+ .doc("Add MaxOutputRows rule for output rows limitation " +
+ "to avoid huge output rows of non-limit query unexpectedly, " +
+ "it's a optional, it's optional that works with defined")
Review comment:
Could you give me some suggestion?😂
##########
File path:
dev/kyuubi-extension-spark-3-1/src/main/scala/org/apache/kyuubi/sql/watchdog/ForcedMaxOutputRowsRule.scala
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.sql.watchdog
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.{GlobalLimit, LogicalPlan,
Project}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.execution.command.InsertIntoDataSourceDirCommand
+
+import org.apache.kyuubi.sql.KyuubiSQLConf
+
+/*
+* Add ForcedMaxOutputRows rule for output rows limitation to avoid huge output
rows of non_limit query unexpectedly
+* mainly applied to cases as below:
+* case 1:
+* {{{
+* INSERT OVERWRITE DIRECTORY (path=STRING)?
+* USING format OPTIONS ([option1_name "option1_value", option2_name
"option2_value", ...])
+* SELECT ...
+* }}}
+*
+* case 2:
+* {{{
+* SELECT [c1, c2, ...]
+* }}}
+*
+* case 3:
+* {{{
+* WITH CTE AS (
+* ...)
+* SELECT [c1, c2, ...] FROM CTE ...
+* }}}
+*
+* The Logical Rule add a GlobalLimit node before root project
+* */
+case class ForcedMaxOutputRowsRule(session: SparkSession) extends
Rule[LogicalPlan] {
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!plan.resolved) {
+ plan
+ } else {
+ conf.getConf(KyuubiSQLConf.WATCHDOG_FORCED_MAXOUTPUTROWS) match {
+ case Some(forcedMaxOutputRows) => plan match {
+ case insert@InsertIntoDataSourceDirCommand(_, _, Project(_, _), _) =>
+ insert.copy(query = GlobalLimit(forcedMaxOutputRows, insert.query))
+ case project@Project(_, _) => project.origin match {
+ case Origin(None, None) => GlobalLimit(forcedMaxOutputRows,
project)
+ case Origin(Some(_), Some(0)) => GlobalLimit(forcedMaxOutputRows,
project)
Review comment:
> Seems a little hack, the `Origin` is used to distinguish if the plan
is subquery ?
The `Origin` provides a location for TreeNodes in the context of their
origin, it define as below(reference:
https://github.com/apache/spark/blob/branch-3.1/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreeNode.scala#L49
):
```scala
case class Origin(
line: Option[Int] = None,
startPosition: Option[Int] = None)
The chief output node as root with init position(None or Some(0)), it's not
used to distinguish subquery, we can use subquery class to identify
> is there any issue if we also add limit for suquery ? I think we don't
need care subquery as following:
>
> * scalar subquery; it return at most one row or exception
> * in subquery; it's ok to add limit so we can protect engine's driver (OOM)
> * exist subquery; it will be rewrite to scalar subquery, so it's also safe.
Add limit for subquery may make difference with origin query，my implement
will not limit node to subquery
Do you tend to add subquery with above three reasons or not？
--
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]
