This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 028760d8e5 [spark] Reject recursive Paimon view definitions (#10231)
028760d8e5 is described below
commit 028760d8e52ff179f42d568f64fafabf0e79740d
Author: Zouxxyy <[email protected]>
AuthorDate: Sat Oct 3 17:05:43 2026 +0800
[spark] Reject recursive Paimon view definitions (#10231)
---
.../analysis/PaimonViewDependencyChecker.scala | 146 +++++++++++++
.../catalyst/analysis/PaimonViewResolver.scala | 41 ++--
.../paimon/spark/execution/PaimonStrategy.scala | 1 +
.../paimon/spark/execution/PaimonViewExec.scala | 7 +-
.../paimon/spark/sql/PaimonViewTestBase.scala | 225 ++++++++++++++++++++-
5 files changed, 401 insertions(+), 19 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewDependencyChecker.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewDependencyChecker.scala
new file mode 100644
index 0000000000..04a1866e46
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewDependencyChecker.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.paimon.spark.catalyst.analysis
+
+import org.apache.paimon.catalog.Catalog.ViewNotExistException
+import org.apache.paimon.spark.catalog.SupportView
+
+import org.apache.spark.sql.{PaimonUtils, SparkSession}
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.connector.catalog.{Identifier, PaimonLookupCatalog}
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
+
+import java.util.Locale
+
+import scala.collection.mutable
+
+/** Checks Paimon view dependencies for cycles and excessive nesting. */
+private[spark] class PaimonViewDependencyChecker(spark: SparkSession) extends
PaimonLookupCatalog {
+
+ protected lazy val catalogManager = spark.sessionState.catalogManager
+
+ // Reuse only within one resolution traversal, never across queries or
metadata changes.
+ private val completedDepths = mutable.Map[ViewReference, Int]()
+ private val maxNestedViewDepth = spark.sessionState.conf.maxNestedViewDepth
+
+ def validate(catalog: SupportView, ident: Identifier, queryText: String):
Unit = {
+ val active = mutable.ArrayBuffer[ViewReference]()
+
+ def visit(
+ currentCatalog: SupportView,
+ currentIdent: Identifier,
+ currentQueryText: => String): Int = {
+ val reference = viewReference(currentCatalog, currentIdent)
+ val cycleStart = active.indexOf(reference)
+ if (cycleStart >= 0) {
+ val cycle = (active.slice(cycleStart, active.size) :+ reference)
+ .map(_.displayName)
+ .mkString(" -> ")
+ throw new IllegalArgumentException(s"Recursive Paimon view reference
detected: $cycle")
+ }
+
+ def checkDepth(depth: Int): Unit = {
+ if (active.size + depth > maxNestedViewDepth) {
+ val path = (active.toSeq :+ reference).map(_.displayName).mkString("
-> ")
+ throw new IllegalArgumentException(
+ s"Paimon view depth exceeds spark.sql.view.maxNestedViewDepth " +
+ s"($maxNestedViewDepth): $path")
+ }
+ }
+
+ completedDepths.get(reference) match {
+ case Some(depth) =>
+ // A shared dependency may be reached along a deeper path after its
first validation.
+ checkDepth(depth)
+ depth
+ case None =>
+ // Load first so that a table at the depth limit is not counted as
another view.
+ val query = currentQueryText
+ checkDepth(1)
+ active += reference
+ try {
+ val childDepth = viewDependencies(query).foldLeft(0) {
+ case (depth, (dependencyCatalog, dependencyIdent)) =>
+ val dependencyDepth =
+ try {
+ visit(
+ dependencyCatalog,
+ dependencyIdent,
+
dependencyCatalog.loadView(dependencyIdent).query(SupportView.DIALECT))
+ } catch {
+ case _: ViewNotExistException => 0
+ }
+ math.max(depth, dependencyDepth)
+ }
+ val depth = childDepth + 1
+ completedDepths(reference) = depth
+ depth
+ } finally {
+ active.remove(active.size - 1)
+ }
+ }
+ }
+
+ visit(catalog, ident, queryText)
+ }
+
+ private def viewDependencies(queryText: String): Seq[(SupportView,
Identifier)] = {
+ val parsedPlan =
PaimonUtils.parseQueryCompat(spark.sessionState.sqlParser, queryText)
+ val earlyRules = SparkShimLoader.shim.earlyBatchRules()
+ val rewritten = earlyRules.foldLeft(parsedPlan)((plan, rule) =>
rule.apply(plan))
+ val dependencies = mutable.ArrayBuffer[(SupportView, Identifier)]()
+
+ def collectDependencies(plan: LogicalPlan): Unit = {
+ plan.foreach {
+ node =>
+ node match {
+ case UnresolvedRelation(
+ CatalogAndIdentifier(dependencyCatalog: SupportView,
dependencyIdent),
+ _,
+ _) =>
+ dependencies += ((dependencyCatalog, dependencyIdent))
+ case _ =>
+ }
+ node.expressions.foreach {
+ _.collect { case subquery: SubqueryExpression => subquery.plan }
+ .foreach(collectDependencies)
+ }
+ }
+ }
+
+ collectDependencies(rewritten)
+ dependencies.toSeq
+ }
+
+ private def viewReference(catalog: SupportView, ident: Identifier):
ViewReference = {
+ def normalize(name: String): String = {
+ if (catalog.paimonCatalog().caseSensitive()) name else
name.toLowerCase(Locale.ROOT)
+ }
+ ViewReference(
+ catalog.paimonCatalogName(),
+ ident.namespace().toSeq.map(normalize),
+ normalize(ident.name()))
+ }
+
+ private case class ViewReference(catalogName: String, namespace:
Seq[String], name: String) {
+ def displayName: String = (catalogName +: namespace :+ name).mkString(".")
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewResolver.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewResolver.scala
index d60e9ab8f3..a85466caa9 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewResolver.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonViewResolver.scala
@@ -40,24 +40,31 @@ case class PaimonViewResolver(spark: SparkSession)
protected lazy val catalogManager = spark.sessionState.catalogManager
- override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperators {
- case u @ UnresolvedRelation(parts @ CatalogAndIdentifier(catalog:
SupportView, ident), _, _) =>
- try {
- val view = catalog.loadView(ident)
- createViewRelation(parts, view)
- } catch {
- case _: ViewNotExistException =>
- u
- }
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ val dependencyChecker = new PaimonViewDependencyChecker(spark)
+ plan.resolveOperatorsDown {
+ case u @ UnresolvedRelation(
+ parts @ CatalogAndIdentifier(catalog: SupportView, ident),
+ _,
+ _) =>
+ try {
+ val view = catalog.loadView(ident)
+ dependencyChecker.validate(catalog, ident,
view.query(SupportView.DIALECT))
+ createViewRelation(parts, view)
+ } catch {
+ case _: ViewNotExistException =>
+ u
+ }
- case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog: SupportView,
ident), _, _) =>
- try {
- catalog.loadView(ident)
- ResolvedPaimonView(catalog, ident)
- } catch {
- case _: ViewNotExistException =>
- u
- }
+ case u @ UnresolvedTableOrView(CatalogAndIdentifier(catalog:
SupportView, ident), _, _) =>
+ try {
+ catalog.loadView(ident)
+ ResolvedPaimonView(catalog, ident)
+ } catch {
+ case _: ViewNotExistException =>
+ u
+ }
+ }
}
private def createViewRelation(nameParts: Seq[String], view: View):
LogicalPlan = {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 770b9eec66..893d7a2734 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
@@ -173,6 +173,7 @@ case class PaimonStrategy(spark: SparkSession)
replace
) =>
CreatePaimonViewExec(
+ spark,
viewCatalog,
ident,
queryText,
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonViewExec.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonViewExec.scala
index e7ddaf463d..f0ae4c7206 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonViewExec.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonViewExec.scala
@@ -19,9 +19,11 @@
package org.apache.paimon.spark.execution
import org.apache.paimon.spark.catalog.SupportView
+import org.apache.paimon.spark.catalyst.analysis.PaimonViewDependencyChecker
import org.apache.paimon.spark.leafnode.PaimonLeafV2CommandExec
import org.apache.paimon.view.View
+import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute,
GenericInternalRow}
import org.apache.spark.sql.catalyst.util.{escapeSingleQuotedString,
quoteIfNeeded, StringUtils}
@@ -33,6 +35,7 @@ import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
case class CreatePaimonViewExec(
+ spark: SparkSession,
catalog: SupportView,
ident: Identifier,
queryText: String,
@@ -62,8 +65,10 @@ case class CreatePaimonViewExec(
// Apply column aliases and comments to the view schema
val finalSchema = applyColumnAliasesAndComments(viewSchema, columnAliases,
columnComments)
- // Note: for replace just drop then create ,this operation is non-atomic.
+ // Note: for replace just drop then create, this operation is non-atomic.
if (replace) {
+ // The analyzed plan has already expanded referenced views, so validate
the original SQL text.
+ new PaimonViewDependencyChecker(spark).validate(catalog, ident,
queryText)
catalog.dropView(ident, true)
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala
index f4e48f318f..57226aa868 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonViewTestBase.scala
@@ -18,9 +18,15 @@
package org.apache.paimon.spark.sql
-import org.apache.paimon.spark.PaimonHiveTestBase
+import org.apache.paimon.spark.{PaimonHiveTestBase, SparkCatalog}
+import org.apache.paimon.spark.catalog.SupportView
+import org.apache.paimon.spark.catalyst.analysis.PaimonViewResolver
+import org.apache.paimon.view.View
import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.Identifier
+
+import java.util.Collections
abstract class PaimonViewTestBase extends PaimonHiveTestBase {
@@ -64,6 +70,214 @@ abstract class PaimonViewTestBase extends
PaimonHiveTestBase {
}
}
+ test("Paimon View: reject recursive replacement") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withView("v1", "v2") {
+ sql("CREATE VIEW v1 AS SELECT 1 AS id")
+ val selfReference = intercept[IllegalArgumentException] {
+ sql("CREATE OR REPLACE VIEW v1 AS SELECT * FROM v1")
+ }
+ assert(selfReference.getMessage.contains("Recursive Paimon view
reference detected"))
+ assert(
+ selfReference.getMessage.contains("paimon_hive.test_db.v1 ->
paimon_hive.test_db.v1"))
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
+
+ sql("CREATE VIEW v2 AS SELECT * FROM v1")
+ val indirectReference = intercept[IllegalArgumentException] {
+ sql("CREATE OR REPLACE VIEW v1 AS SELECT * FROM v2")
+ }
+ assert(
+ indirectReference.getMessage.contains(
+ "paimon_hive.test_db.v1 -> paimon_hive.test_db.v2 ->
paimon_hive.test_db.v1"))
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
+ }
+ }
+ }
+
+ test("Paimon View: reject persisted recursive views") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withView("v1", "v2") {
+ val viewCatalog =
+
spark.sessionState.catalogManager.currentCatalog.asInstanceOf[SupportView]
+ val v1 = Identifier.of(Array("test_db"), "v1")
+ val schema = sql("SELECT 1 AS id").schema
+ sql("CREATE VIEW v1 AS SELECT 1 AS id")
+ sql("CREATE VIEW v2 AS SELECT * FROM v1")
+ Seq(
+ "SELECT * FROM v1",
+ "SELECT * FROM v2",
+ "SELECT 1 AS id WHERE EXISTS (SELECT * FROM v1)",
+ "SELECT (SELECT max(id) FROM v1) AS id"
+ ).foreach {
+ query =>
+ // Simulate a recursive view persisted by an older Paimon version.
+ viewCatalog.dropView(v1, true)
+ viewCatalog.createView(
+ v1,
+ schema,
+ query,
+ null,
+ Collections.emptyMap[String, String](),
+ false)
+ val error = intercept[IllegalArgumentException] {
+ sql("SELECT * FROM v1")
+ }
+ assert(error.getMessage.contains("Recursive Paimon view reference
detected"))
+ assert(error.getMessage.contains("paimon_hive.test_db.v1"))
+ }
+ // An invalid legacy definition can still be replaced with a valid one.
+ sql("CREATE OR REPLACE VIEW v1 AS SELECT 2 AS id")
+ checkAnswer(sql("SELECT * FROM v2"), Seq(Row(2)))
+ }
+ }
+ }
+
+ test("Paimon View: reject recursive replacement in subqueries") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withTable("t") {
+ withView("v1") {
+ sql("CREATE TABLE t (id INT) USING paimon")
+ sql("INSERT INTO t VALUES (1)")
+ sql("CREATE VIEW v1 AS SELECT * FROM t")
+
+ Seq(
+ "SELECT * FROM t WHERE EXISTS (SELECT * FROM v1)",
+ "SELECT * FROM t WHERE id IN (SELECT id FROM v1)",
+ "SELECT (SELECT max(id) FROM v1) AS id FROM t",
+ "WITH c AS (SELECT * FROM v1) SELECT * FROM c"
+ ).foreach {
+ query =>
+ val error = intercept[IllegalArgumentException] {
+ sql(s"CREATE OR REPLACE VIEW v1 AS $query")
+ }
+ assert(error.getMessage.contains("paimon_hive.test_db.v1 ->
paimon_hive.test_db.v1"))
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
+ }
+ }
+ }
+ }
+ }
+
+ test("Paimon View: reject recursive replacement with mixed case
identifiers") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withView("v1") {
+ sql("CREATE VIEW v1 AS SELECT 1 AS id")
+ val error = intercept[IllegalArgumentException] {
+ sql("CREATE OR REPLACE VIEW v1 AS SELECT * FROM TEST_DB.V1")
+ }
+ assert(error.getMessage.contains("paimon_hive.test_db.v1 ->
paimon_hive.test_db.v1"))
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
+ }
+ }
+ }
+
+ test("Paimon View: allow CTE shadowing and shared view dependencies") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withView("v1", "v2", "v3", "v4") {
+ sql("CREATE VIEW v1 AS SELECT 1 AS id")
+ sql("CREATE OR REPLACE VIEW v1 AS WITH v1 AS (SELECT 2 AS id) SELECT *
FROM v1")
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(2)))
+ sql("CREATE VIEW v2 AS SELECT * FROM v1")
+ sql("CREATE VIEW v3 AS SELECT * FROM v1")
+ sql("CREATE VIEW v4 AS SELECT * FROM v2 UNION ALL SELECT * FROM v3")
+ checkAnswer(sql("SELECT * FROM v4"), Seq(Row(2), Row(2)))
+ }
+ }
+ }
+
+ test("Paimon View: reuse validation within a traversal but not across
queries") {
+ val catalogName = "paimon_counting_views"
+ withSQLConf(
+ s"spark.sql.catalog.$catalogName" ->
classOf[CountingViewSparkCatalog].getName,
+ s"spark.sql.catalog.$catalogName.metastore" -> "hive",
+ s"spark.sql.catalog.$catalogName.uri" -> PaimonHiveTestBase.hiveUri,
+ s"spark.sql.catalog.$catalogName.warehouse" ->
tempHiveDBDir.getCanonicalPath
+ ) {
+ sql(s"USE $catalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ val names = (0 until 8).map(i => s"v$i")
+ withView(names: _*) {
+ sql("CREATE VIEW v0 AS SELECT 1 AS id")
+ names.indices.drop(1).foreach {
+ i => sql(s"CREATE VIEW ${names(i)} AS SELECT * FROM ${names(i -
1)}")
+ }
+ val catalog = spark.sessionState.catalogManager.currentCatalog
+ .asInstanceOf[CountingViewSparkCatalog]
+ val resolver = PaimonViewResolver(spark)
+ def query = spark.sessionState.sqlParser.parsePlan(s"SELECT * FROM
${names.last}")
+
+ catalog.viewLoads = 0
+ resolver(query)
+ // A chain must not trigger another full dependency scan for every
expanded view.
+ assert(catalog.viewLoads <= names.size * 2, s"View loads:
${catalog.viewLoads}")
+
+ // Reusing the rule for a new query must detect a newly persisted
cycle.
+ val first = Identifier.of(Array("test_db"), names.head)
+ catalog.dropView(first, true)
+ catalog.createView(
+ first,
+ sql("SELECT 1 AS id").schema,
+ s"SELECT * FROM ${names.last}",
+ null,
+ Collections.emptyMap[String, String](),
+ false)
+ val error = intercept[IllegalArgumentException] {
+ resolver(query)
+ }
+ assert(error.getMessage.contains("Recursive Paimon view reference
detected"))
+ }
+ }
+ }
+ }
+
+ test("Paimon View: enforce nested depth for shared dependencies") {
+ sql(s"USE $paimonHiveCatalogName")
+ withDatabase("test_db") {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ withTable("t") {
+ withView("v1", "v2", "v3", "v4") {
+ sql("CREATE TABLE t (id INT) USING paimon")
+ sql("INSERT INTO t VALUES (1)")
+ sql("CREATE VIEW v1 AS SELECT * FROM t")
+ sql("CREATE VIEW v2 AS SELECT * FROM v1")
+ sql("CREATE VIEW v3 AS SELECT * FROM v2")
+ sql("CREATE VIEW v4 AS SELECT * FROM v1 UNION ALL SELECT * FROM v3")
+ withSQLConf("spark.sql.view.maxNestedViewDepth" -> "1") {
+ checkAnswer(sql("SELECT * FROM v1"), Seq(Row(1)))
+ }
+ withSQLConf("spark.sql.view.maxNestedViewDepth" -> "3") {
+ checkAnswer(sql("SELECT * FROM v3"), Seq(Row(1)))
+ val error = intercept[IllegalArgumentException] {
+ sql("SELECT * FROM v4")
+ }
+
assert(error.getMessage.contains("spark.sql.view.maxNestedViewDepth (3)"))
+ }
+ withSQLConf("spark.sql.view.maxNestedViewDepth" -> "4") {
+ checkAnswer(sql("SELECT * FROM v4"), Seq(Row(1), Row(1)))
+ }
+ }
+ }
+ }
+ }
+
test("Paimon View: show views") {
Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
catalogName =>
@@ -395,3 +609,12 @@ abstract class PaimonViewTestBase extends
PaimonHiveTestBase {
}
}
}
+
+private[sql] class CountingViewSparkCatalog extends SparkCatalog {
+ var viewLoads: Int = 0
+
+ override def loadView(ident: Identifier): View = {
+ viewLoads += 1
+ super.loadView(ident)
+ }
+}
