You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "srielau (via GitHub)" <gi...@apache.org> on 2026/04/16 01:58:22 UTC
[PR] [SPARK-56501] SET PATH syntax [spark]
srielau opened a new pull request, #55364:
URL: https://github.com/apache/spark/pull/55364
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3101530475
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala:
##########
@@ -119,6 +119,11 @@ case class SetCommand(kv: Option[(String, Option[String])])
messageParameters = Map("variableName" -> toSQLId(varName)))
}
}
+ if (key == SQLConf.SESSION_PATH.key) {
Review Comment:
Agreed the guard was incomplete. Moved the session path entirely out of SQLConf into structured state on `CatalogManager` (new `_sessionPath: Option[Seq[Seq[String]]]` field). The `SESSION_PATH` config key no longer exists, so `spark.conf.set`/`RESET`/`RuntimeConfig` cannot touch it. The guard is removed. See second commit.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
Review Comment:
Done. Fixed to "current catalog".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ private[sql] def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
+ private[sql] def concreteSessionPathEntry(
+ entry: Seq[String],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[String] = {
+ if (isVirtualCurrentSchemaPathEntry(entry)) {
+ if (currentNamespace.isEmpty) Seq(currentCatalog)
+ else currentCatalog +: currentNamespace
+ } else {
+ entry
+ }
+ }
+
+ /** Expand markers in stored session path using the current catalog and namespace. */
+ private[sql] def expandSessionPathMarkers(
+ entries: Seq[Seq[String]],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] =
+ entries.map(concreteSessionPathEntry(_, currentCatalog, currentNamespace))
+
+ /**
+ * Separator between path entries in serialized session path
+ * (catalog.namespace,catalog.namespace).
+ */
+ private[sql] val SESSION_PATH_ENTRY_SEPARATOR = ","
+
+ /**
+ * Parses a session path string into a list of path entries.
+ * Format: "catalog1.namespace1,catalog2.namespace2"
+ * (comma-separated, each entry catalog.namespace).
+ */
+ private[sql] def parseSessionPath(pathStr: String): Seq[Seq[String]] = {
+ if (pathStr == null || pathStr.trim.isEmpty) return Seq.empty
+ pathStr.split(SESSION_PATH_ENTRY_SEPARATOR).map { entry =>
+ val trimmed = entry.trim
+ val dot = trimmed.indexOf('.')
+ if (dot <= 0 || dot == trimmed.length - 1) {
+ Seq(trimmed)
+ } else {
+ Seq(trimmed.substring(0, dot).trim, trimmed.substring(dot + 1).trim)
+ }
+ }.toSeq.filter(_.nonEmpty)
+ }
+
+ /**
+ * Formats path entries to session path string.
+ * Each entry is catalog.namespace; entries separated by comma.
+ */
+ private[sql] def formatSessionPath(pathEntries: Seq[Seq[String]]): String =
+ pathEntries.map(_.mkString(".")).mkString(SESSION_PATH_ENTRY_SEPARATOR)
Review Comment:
Fully addressed. Moved to structured state on `CatalogManager` — the path is stored as `Option[Seq[Seq[String]]]`, so there is no serialization at all. Multi-level namespaces and backtick-quoted identifiers with `.`/`,` are preserved structurally. The `parseSessionPath`/`formatSessionPath` methods and separator constants have been removed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ private[sql] def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
+ private[sql] def concreteSessionPathEntry(
+ entry: Seq[String],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[String] = {
+ if (isVirtualCurrentSchemaPathEntry(entry)) {
+ if (currentNamespace.isEmpty) Seq(currentCatalog)
+ else currentCatalog +: currentNamespace
+ } else {
+ entry
+ }
+ }
+
+ /** Expand markers in stored session path using the current catalog and namespace. */
+ private[sql] def expandSessionPathMarkers(
+ entries: Seq[Seq[String]],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] =
+ entries.map(concreteSessionPathEntry(_, currentCatalog, currentNamespace))
+
+ /**
+ * Separator between path entries in serialized session path
+ * (catalog.namespace,catalog.namespace).
+ */
+ private[sql] val SESSION_PATH_ENTRY_SEPARATOR = ","
+
+ /**
+ * Parses a session path string into a list of path entries.
+ * Format: "catalog1.namespace1,catalog2.namespace2"
+ * (comma-separated, each entry catalog.namespace).
+ */
+ private[sql] def parseSessionPath(pathStr: String): Seq[Seq[String]] = {
+ if (pathStr == null || pathStr.trim.isEmpty) return Seq.empty
+ pathStr.split(SESSION_PATH_ENTRY_SEPARATOR).map { entry =>
+ val trimmed = entry.trim
+ val dot = trimmed.indexOf('.')
+ if (dot <= 0 || dot == trimmed.length - 1) {
+ Seq(trimmed)
+ } else {
+ Seq(trimmed.substring(0, dot).trim, trimmed.substring(dot + 1).trim)
+ }
+ }.toSeq.filter(_.nonEmpty)
+ }
+
+ /**
+ * Formats path entries to session path string.
Review Comment:
Moot — the method was removed entirely (path is now structured state on CatalogManager, no serialization needed).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -953,10 +953,9 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
* Raises TABLE_OR_VIEW_NOT_FOUND with a formatted list of resolution path entries.
*
* @param quotedResolutionPathEntries each string is already a display-ready path entry (typically
- * `toSQLId` of each segment from `SQLConf.resolutionSearchPath`). They are joined with
- * ", " inside square brackets. This differs from [[noSuchTableError]](nameParts:
- * Seq[String]), which builds one dotted bracketed path from `nameParts.dropRight(1)` via
- * [[org.apache.spark.sql.catalyst.analysis.NoSuchTableException]].
+ * `toSQLId` of each path entry from `SQLConf.sqlResolutionPathEntries`). They are joined
+ * with ", " inside square brackets, same as
+ * [[org.apache.spark.sql.catalyst.analysis.NoSuchItemExceptionHelper.formatSearchPath]].
Review Comment:
Done. Restored the original doc that correctly describes the difference from `formatSearchPath`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3101527442
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
Review Comment:
Done. SchemaInPath now requires ≥2 parts (multi-level namespaces accepted). 1-part form removed — agreed it's a SET-time capture trap. `qualifySchemaParts` collapsed to a simple arity check per your suggestion.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
Review Comment:
Done. Replaced with `expanded.foreach { ... }` validation + `seen.add(key)`, then use `expanded` directly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = (normalize(concrete.head),
+ normalize(concrete.lift(1).getOrElse("")))
+ if (seen.contains(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ seen.add(key)
+ Some(entry)
+ }
+
+ if (deduped.isEmpty) {
+ conf.unsetConf(SQLConf.SESSION_PATH)
+ } else {
+ conf.setConfString(SQLConf.SESSION_PATH.key, SQLConf.formatSessionPath(deduped))
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = CatalogManager.BUILTIN_NAMESPACE
+ val session = CatalogManager.SESSION_NAMESPACE
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ // Default path = session order (first/second/last). Clear path; use at resolution time.
+ Seq.empty
Review Comment:
Done. `DEFAULT_PATH` is now restricted to the sole-element form — when combined with other elements it throws `UNSUPPORTED_FEATURE.DEFAULT_PATH_COMBINED`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = (normalize(concrete.head),
+ normalize(concrete.lift(1).getOrElse("")))
+ if (seen.contains(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ seen.add(key)
+ Some(entry)
+ }
+
+ if (deduped.isEmpty) {
+ conf.unsetConf(SQLConf.SESSION_PATH)
+ } else {
+ conf.setConfString(SQLConf.SESSION_PATH.key, SQLConf.formatSessionPath(deduped))
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = CatalogManager.BUILTIN_NAMESPACE
+ val session = CatalogManager.SESSION_NAMESPACE
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ // Default path = session order (first/second/last). Clear path; use at resolution time.
+ Seq.empty
+ case PathElement.SystemPath =>
+ Seq(Seq(systemCatalog, builtin), Seq(systemCatalog, session))
+ case PathElement.CurrentDatabase | PathElement.CurrentSchema =>
+ Seq(Seq(systemCatalog, SQLConf.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA))
+ case PathElement.PathRef =>
+ conf.sessionPath match {
+ case Some(s) => SQLConf.parseSessionPath(s)
+ case None => Seq.empty
+ }
+ case PathElement.SchemaInPath(parts) =>
+ qualifySchemaParts(parts, systemCatalog, currentCatalog)
+ }
+ }
+
+ /** Qualify schema parts at SET time: well-known -> SYSTEM; else current catalog + namespace. */
+ private def qualifySchemaParts(
+ parts: Seq[String],
+ systemCatalog: String,
+ currentCatalog: String): Seq[Seq[String]] = {
+ val wellKnown = Set(
+ CatalogManager.BUILTIN_NAMESPACE.toLowerCase(Locale.ROOT),
+ CatalogManager.SESSION_NAMESPACE.toLowerCase(Locale.ROOT))
+ if (parts.isEmpty) return Seq.empty
+ if (parts.length > 2) {
+ throw QueryCompilationErrors.invalidSqlPathSchemaReferenceError(parts.mkString("."))
+ }
+ if (parts.size == 1) {
+ val ns = parts.head
+ if (wellKnown.contains(ns.toLowerCase(Locale.ROOT))) {
+ Seq(Seq(systemCatalog, ns))
+ } else {
+ Seq(Seq(currentCatalog, ns))
+ }
+ } else {
+ Seq(Seq(parts.head, parts(1)))
+ }
+ }
Review Comment:
Done. Both concerns addressed:
- ≥3-part now accepted (multi-level namespaces like `iceberg_cat.db1.db2.schema`)
- 1-part disallowed — the method collapsed to exactly your suggestion:
```scala
private def qualifySchemaParts(parts: Seq[String]): Seq[Seq[String]] = {
if (parts.length < 2) {
throw QueryCompilationErrors.invalidSqlPathSchemaReferenceError(parts.mkString("."))
}
Seq(parts)
}
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -357,6 +357,19 @@ class SparkSqlAstBuilder extends AstBuilder {
* SET TIME ZONE INTERVAL 10 HOURS;
* }}}
*/
+ override def visitSetPath(ctx: SetPathContext): LogicalPlan = withOrigin(ctx) {
+ val elements = (0 until ctx.pathElement().size()).map { i =>
+ val pe = ctx.pathElement(i)
+ if (pe.DEFAULT_PATH() != null) PathElement.DefaultPath
+ else if (pe.SYSTEM_PATH() != null) PathElement.SystemPath
+ else if (pe.PATH() != null) PathElement.PathRef
+ else if (pe.CURRENT_DATABASE() != null) PathElement.CurrentDatabase
+ else if (pe.CURRENT_SCHEMA() != null) PathElement.CurrentSchema
+ else PathElement.SchemaInPath(visitMultipartIdentifier(pe.multipartIdentifier()))
+ }
Review Comment:
Done. Applied the suggested `ctx.pathElement().asScala.map { pe => ... }.toSeq` form.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -1429,7 +1442,9 @@ class SparkSqlAstBuilder extends AstBuilder {
val tableName = ctx.identifierReference.getText.split("\\.").lastOption.getOrElse("table")
throw QueryCompilationErrors.describeJsonNotExtendedError(tableName)
}
- val relation = createUnresolvedTableOrView(ctx.identifierReference, "DESCRIBE TABLE")
+ val relation = createUnresolvedTableOrView(
+ ctx.identifierReference,
+ "DESCRIBE TABLE")
Review Comment:
Done. Reverted.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on PR #55364:
URL: https://github.com/apache/spark/pull/55364#issuecomment-4269401660
@cloud-fan Thank you for the thorough review. All comments addressed in two commits:
**Commit 1** — direct fixes:
- `qualifySchemaParts` simplified: 1-part rejected, ≥2-part (multi-level namespaces) accepted as-is
- `deduped` → `foreach` validation; duplicate key now handles `Seq[String]`
- `DEFAULT_PATH` restricted to sole-element form
- `SYSTEM_PATH` respects `sessionFunctionResolutionOrder`
- Idiomatic `asScala.map`; reverted unrelated formatting
- Fixed `QueryCompilationErrors` doc, `SQLConf` doc ("session catalog" → "current catalog")
- Strengthened all test assertions (exact `===`), added cross-alias dup, multi-level ns, DEFAULT_PATH combined, backtick round-trip, 1-part rejection tests
**Commit 2** — structural redesign per your preferred approach:
- Moved session path from `SQLConf` string to structured `Option[Seq[Seq[String]]]` on `CatalogManager`
- Removed `SESSION_PATH` config, `parseSessionPath`/`formatSessionPath`, serialization separators, `SetCommand` guard, `SET_PATH_VIA_SET` error
- `SetPathCommand` writes `CatalogManager` directly; `ReplaceCurrentLike` reads `catalogManager.currentPathString`
- Session cloning wired via `BaseSessionStateBuilder` (`copySessionPathFrom`)
- Added test: programmatic `spark.conf.set` has no effect; cloned session inherits path
Regarding your summary concerns:
- **(a)** Storing path as structured `CatalogManager` state — done.
- Lossy serialization — eliminated entirely (no serialization).
- Guard bypass — structurally impossible (config key removed).
- **(d)** 1-part rejected, ≥3-part accepted, `qualifySchemaParts` collapsed to arity check.
- `DEFAULT_PATH` inconsistency — fixed (sole-element restriction).
- `SYSTEM_PATH` ordering — now respects `spark.sql.function.resolutionOrder`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3102678741
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala:
##########
@@ -152,17 +152,16 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
* Replaces the expression of CurrentDatabase, CurrentCatalog, CurrentPath, and CurrentUser
* with the current values.
*/
-case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
+case class ReplaceCurrentLike(
+ catalogManager: CatalogManager,
+ sqlConf: SQLConf) extends Rule[LogicalPlan] {
Review Comment:
`sqlConf: SQLConf` is never referenced in the body after `currentPathStr` was changed to call `catalogManager.currentPathString`. Remove the parameter (and the `conf` argument in `Optimizer.scala:338`).
```suggestion
case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
```
(collapse lines 155-157 into a single line.)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -126,6 +126,42 @@ class CatalogManager(
_currentNamespace = Some(namespace)
}
+ private var _sessionPath: Option[Seq[Seq[String]]] = None
+
+ /** Returns the raw stored session path entries, or None if no path is set. */
+ def sessionPathEntries: Option[Seq[Seq[String]]] = synchronized { _sessionPath }
+
+ def setSessionPath(entries: Seq[Seq[String]]): Unit = synchronized {
+ _sessionPath = Some(entries)
+ }
+
+ def clearSessionPath(): Unit = synchronized {
+ _sessionPath = None
+ }
+
+ private[sql] def copySessionPathFrom(other: CatalogManager): Unit = synchronized {
+ _sessionPath = other.sessionPathEntries
+ }
+
+ /**
+ * String form of the current resolution path for CURRENT_PATH().
+ * When PATH is enabled and a session path is stored, formats the effective path entries
+ * with markers expanded. Otherwise falls back to the legacy resolutionSearchPath.
+ */
+ def currentPathString: String = {
+ import CatalogV2Implicits._
+ val entries = if (conf.pathEnabled) _sessionPath else None
Review Comment:
All other accessors of `_sessionPath` (`sessionPathEntries`, `setSessionPath`, `clearSessionPath`, `copySessionPathFrom`) synchronize on `this`, but `currentPathString` reads `_sessionPath` directly. Under concurrent `SET PATH` from another thread this is a data race. Read via the synchronized accessor instead:
```suggestion
val entries = if (conf.pathEnabled) sessionPathEntries else None
```
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2061,6 +2061,12 @@
],
"sqlState" : "42711"
},
+ "DUPLICATE_SQL_PATH_ENTRY" : {
+ "message" : [
+ "Duplicate SQL path entry <pathEntry>. The session SQL path cannot contain the same catalog.schema more than once (including after expanding shortcuts like DEFAULT_PATH or SYSTEM_PATH)."
Review Comment:
Multi-level namespaces (e.g. `iceberg_cat.db1.db2`) are now explicitly accepted by `SchemaInPath`, but this message still says "the same catalog.schema". Reword so it stays accurate for 3+ parts.
```suggestion
"Duplicate SQL path entry <pathEntry>. The session SQL path cannot contain the same path entry more than once (including after expanding shortcuts like DEFAULT_PATH or SYSTEM_PATH)."
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala:
##########
@@ -152,17 +152,16 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
* Replaces the expression of CurrentDatabase, CurrentCatalog, CurrentPath, and CurrentUser
* with the current values.
*/
-case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
+case class ReplaceCurrentLike(
+ catalogManager: CatalogManager,
+ sqlConf: SQLConf) extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = {
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
lazy val currentNamespace = catalogManager.currentNamespace.quoted
lazy val currentNamespaceSeq = catalogManager.currentNamespace.toSeq
Review Comment:
`currentNamespaceSeq` was only used by the removed inline `resolutionSearchPath` call and now has no reader. Drop it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,40 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in the session path.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the current catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
Review Comment:
`SESSION_PATH_VIRTUAL_CURRENT_SCHEMA`, `isVirtualCurrentSchemaPathEntry`, `concreteSessionPathEntry`, and `expandSessionPathMarkers` are internal encoding helpers for session-path state — not configuration. The state they manipulate lives on `CatalogManager`, and both call sites (`CatalogManager.currentPathString`, `SetPathCommand`) reach across the module boundary to pull them back. Consider moving them to the `CatalogManager` companion (or a small dedicated `SessionPath` utility under `connector.catalog`) so the representation and its consumers sit together. `PATH_ENABLED` of course stays here — that one *is* configuration.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
Review Comment:
Two easy gaps worth closing in this suite:
- `PATH_ENABLED=true` with no `SET PATH` ever executed should fall back to the legacy `resolutionSearchPath`. The existing "PATH disabled" test only exercises the disabled branch; covering enabled-but-unset locks in the fallback.
- `SET PATH = DEFAULT_PATH, system.builtin` should raise `DUPLICATE_SQL_PATH_ENTRY` (since `DEFAULT_PATH` expands to include `system.builtin`). No test currently exercises shortcut-expansion + explicit clash.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
Review Comment:
`clearSessionPath()` is only reachable when `expanded.isEmpty`, which requires `SET PATH = PATH` against an already-unset path. There's no user-facing way to reset — no `RESET PATH`, and `SET PATH = DEFAULT_PATH` materializes the default entries rather than clearing. Is the dead path + missing reset intentional? If so, consider dropping the `clearSessionPath()` branch; otherwise please add an explicit reset (e.g. `RESET PATH`, or have `SET PATH = DEFAULT_PATH` clear to `None`).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3111676861
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -357,6 +357,18 @@ class SparkSqlAstBuilder extends AstBuilder {
* SET TIME ZONE INTERVAL 10 HOURS;
* }}}
*/
+ override def visitSetPath(ctx: SetPathContext): LogicalPlan = withOrigin(ctx) {
Review Comment:
Done. Moved visitSetPath below with its own Scaladoc; restored the SET TIME ZONE doc above visitSetTimeZone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -204,4 +241,39 @@ private[sql] object CatalogManager {
(nameParts.length == 2 && nameParts.head.equalsIgnoreCase(SESSION_NAMESPACE)) ||
isFullyQualifiedSystemSessionViewName(nameParts)
}
+
+ /** True if a SQL path entry denotes `system.session` (case-insensitive). */
+ def isSystemSessionPathEntry(parts: Seq[String]): Boolean =
Review Comment:
Removed from this PR — no call sites here. Will be added in the persist-path PR (SPARK-56520) which needs it for stripping system.session from stored view/function paths.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
Review Comment:
Done. Parts containing `." are now backtick-quoted in the error message so the rendered entry stays unambiguous.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
+ catalogManager.clearSessionPath()
+ } else {
+ catalogManager.setSessionPath(expanded)
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ catalogManager: CatalogManager,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = Seq(systemCatalog, CatalogManager.BUILTIN_NAMESPACE)
+ val session = Seq(systemCatalog, CatalogManager.SESSION_NAMESPACE)
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ val currentSchema = Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+ conf.sessionFunctionResolutionOrder match {
Review Comment:
Done. Extracted `SQLConf.defaultPathOrder(catalogEntries)` and `SQLConf.systemPathOrder` — single source of truth for the session/builtin ordering. `SetPathCommand.expandPathElements` now delegates to these, eliminating the duplicated ordering logic.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
Review Comment:
Done. Added two tests: DEFAULT_PATH with `sessionFunctionResolutionOrder = "first"` (asserts session comes first) and `"last"` (asserts session comes last).
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
Review Comment:
Investigated: `cloneSession()` shares the parent's `CatalogManager` instance (same as `USE CATALOG`/`USE SCHEMA`). Independence would require `cloneSession` to create a separate `CatalogManager`, which is a larger change. Added a comment documenting this behavior. The clone-inherits test still validates the snapshot-on-clone semantics.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3101531528
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
Review Comment:
All addressed:
- Tests now use exact `===` assertions on path entries instead of weak `.contains()`
- Added: cross-alias duplicate detection (`current_database, current_schema`)
- Added: `DEFAULT_PATH` combined form rejection test
- Added: multi-level namespace (3+ parts) acceptance test
- Added: backtick-quoted identifiers with dots round-trip test
- Added: unqualified (1-part) schema reference rejection test
- Added: programmatic `spark.conf.set` has no effect on `CURRENT_PATH()`
- Added: cloned session inherits parent path
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "dtenedor (via GitHub)" <gi...@apache.org>.
dtenedor commented on PR #55364:
URL: https://github.com/apache/spark/pull/55364#issuecomment-4291730211
LGTM, merging to master
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3119354139
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -126,6 +126,42 @@ class CatalogManager(
_currentNamespace = Some(namespace)
}
+ private var _sessionPath: Option[Seq[Seq[String]]] = None
Review Comment:
Done. Implemented the typed ADT:
```scala
sealed trait SessionPathEntry {
def resolve(currentCatalog: String, currentNamespace: Seq[String]): Seq[String]
}
case object CurrentSchemaEntry extends SessionPathEntry
case class LiteralPathEntry(parts: Seq[String]) extends SessionPathEntry
```
Session path type changed from `Option[Seq[Seq[String]]]` to `Option[Seq[SessionPathEntry]]`. Deleted `SESSION_PATH_VIRTUAL_CURRENT_SCHEMA`, `isVirtualCurrentSchemaPathEntry`, `concreteSessionPathEntry`, and `expandSessionPathMarkers`. All consumers now pattern-match. 27 tests pass.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3119322997
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -204,4 +241,33 @@ private[sql] object CatalogManager {
(nameParts.length == 2 && nameParts.head.equalsIgnoreCase(SESSION_NAMESPACE)) ||
isFullyQualifiedSystemSessionViewName(nameParts)
}
+
+ /** Virtual marker for CURRENT_SCHEMA / CURRENT_DATABASE in SET PATH. */
+ val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
Review Comment:
Scaladoc is narrower than the method's role — `concreteSessionPathEntry` is also the engine for `expandSessionPathMarkers` (line 271), which `currentPathString` uses at CURRENT_PATH() display time. Generalize so the next reader knows it's shared between SET-time dedup and display-time expansion.
```suggestion
/** Materialize virtual current-schema entry against the current catalog/namespace. */
```
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val cloned = spark.cloneSession()
+ val clonedPath = cloned.sql("SELECT current_path()").collect().head.getString(0)
+ val entries = pathEntries(clonedPath)
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Cloned session should inherit parent path; got: $entries")
+ }
+ }
+
+ test("PATH enabled: duplicate path entry raises error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, spark_catalog.default")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema appends to path") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_append_test")
+ try {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ sql("SET PATH = PATH, spark_catalog.path_append_test")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin",
+ "spark_catalog.path_append_test"),
+ s"PATH, schema should append; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_append_test")
+ }
+ }
+ }
+
+ test("PATH enabled: SET PATH = CURRENT_SCHEMA / CURRENT_DATABASE") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"current_schema should expand to current schema, got: $entries")
+ sql("SET PATH = current_database, system.builtin")
+ val entries2 = pathEntries(currentPath())
+ assert(entries2 === Seq("spark_catalog.default", "system.builtin"),
+ s"current_database should expand to current schema, got: $entries2")
+ }
+ }
+
+ test("PATH enabled: cross-alias duplicate detection (current_database, current_schema)") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_database, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: virtual CURRENT_SCHEMA expands to USE schema") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_virt_schema")
+ try {
+ sql("USE spark_catalog.path_virt_schema")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.path_virt_schema", "system.builtin"),
+ s"CURRENT_SCHEMA in PATH should reflect USE; got: $entries")
+ } finally {
+ sql("USE spark_catalog.default")
+ sql("DROP SCHEMA IF EXISTS path_virt_schema")
+ }
+ }
+ }
+
+ test("PATH enabled: duplicate after expanding CURRENT_SCHEMA") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when CURRENT_SCHEMA repeated") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_schema, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when SYSTEM_PATH listed twice") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = SYSTEM_PATH, SYSTEM_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = SYSTEM_PATH includes system.builtin and system.session") {
+ withPathEnabled {
+ sql("SET PATH = SYSTEM_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"SYSTEM_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"SYSTEM_PATH should include system.session; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH on unset session includes defaults") {
+ withPathEnabled {
+ sql("SET PATH = PATH, spark_catalog.extra")
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"PATH on unset session should include builtin defaults; got: $entries")
+ assert(entries.contains("spark_catalog.extra"),
+ s"PATH on unset session should include appended schema; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema after DEFAULT_PATH (empty session path)") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_from_empty")
+ try {
+ sql("SET PATH = DEFAULT_PATH")
+ sql("SET PATH = PATH, spark_catalog.path_from_empty")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.path_from_empty"),
+ s"PATH after cleared path should append schema; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_from_empty")
+ }
+ }
+ }
+
+ test("PATH enabled: unqualified (1-part) schema reference is rejected") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = myschema")
+ },
+ condition = "INVALID_SQL_PATH_SCHEMA_REFERENCE",
+ parameters = Map(
+ "qualifiedName" -> "myschema"))
+ }
+ }
+
+ test("PATH enabled: multi-level namespace (3+ parts) is accepted") {
+ withPathEnabled {
+ sql("SET PATH = iceberg_cat.db1.db2, spark_catalog.default")
+ val entries = pathEntries(currentPath())
+ assert(entries.head === "iceberg_cat.db1.db2",
+ s"Multi-level namespace should be accepted; got: $entries")
+ }
+ }
+
+ test("PATH enabled: backtick-quoted identifiers with dots round-trip correctly") {
+ withPathEnabled {
+ sql("SET PATH = `cat.a`.`sch.b`")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("`cat.a`.`sch.b`"),
+ s"Backtick-quoted identifiers should round-trip; got: $entries")
+ }
+ }
+
+ test("PATH enabled: stored path preserves typed case, resolution is case-insensitive") {
+ withPathEnabled {
+ sql("SET PATH = Spark_Catalog.Default, System.Builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("Spark_Catalog.Default", "System.Builtin"),
+ s"Stored path should preserve case; got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH respects sessionFunctionResolutionOrder = first") {
+ withSQLConf(
+ SQLConf.PATH_ENABLED.key -> "true",
+ SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "first") {
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.head.contains("session"),
+ s"With 'first' order, session should come first; got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH respects sessionFunctionResolutionOrder = last") {
+ withSQLConf(
+ SQLConf.PATH_ENABLED.key -> "true",
+ SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "last") {
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.last.contains("session"),
+ s"With 'last' order, session should come last; got: $entries")
+ }
+ }
+
+ // Note: cloneSession() shares the parent's CatalogManager, so SET PATH on either session
+ // affects both. This matches the behavior of USE CATALOG / USE SCHEMA on cloned sessions.
+ // Independence would require cloneSession to create a separate CatalogManager.
Review Comment:
This Note reads the code backward — `BaseSessionStateBuilder.catalogManager` constructs `new CatalogManager(v2SessionCatalog, catalog)` per forked session and only explicitly copies the session path via `copySessionPathFrom`. The `CatalogManager` instances are not shared.
That said, other `CatalogManager` state propagation on clone (current catalog/namespace, registered catalogs) is currently incidental rather than designed — worth sorting out separately. Rather than rewrite the note now, leave a TODO so a follow-up can pin down the intended semantics.
```suggestion
// TODO: cloneSession() constructs a new CatalogManager per forked session and
// explicitly copies only the stored session path via copySessionPathFrom.
// Other CatalogManager state propagation (current catalog/namespace, registered
// catalogs) on clone is currently incidental — audit and pin down the intended
// semantics in a follow-up.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val cloned = spark.cloneSession()
+ val clonedPath = cloned.sql("SELECT current_path()").collect().head.getString(0)
+ val entries = pathEntries(clonedPath)
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Cloned session should inherit parent path; got: $entries")
+ }
+ }
+
+ test("PATH enabled: duplicate path entry raises error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, spark_catalog.default")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema appends to path") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_append_test")
+ try {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ sql("SET PATH = PATH, spark_catalog.path_append_test")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin",
+ "spark_catalog.path_append_test"),
+ s"PATH, schema should append; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_append_test")
+ }
+ }
+ }
+
+ test("PATH enabled: SET PATH = CURRENT_SCHEMA / CURRENT_DATABASE") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"current_schema should expand to current schema, got: $entries")
+ sql("SET PATH = current_database, system.builtin")
+ val entries2 = pathEntries(currentPath())
+ assert(entries2 === Seq("spark_catalog.default", "system.builtin"),
+ s"current_database should expand to current schema, got: $entries2")
+ }
+ }
+
+ test("PATH enabled: cross-alias duplicate detection (current_database, current_schema)") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_database, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: virtual CURRENT_SCHEMA expands to USE schema") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_virt_schema")
+ try {
+ sql("USE spark_catalog.path_virt_schema")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.path_virt_schema", "system.builtin"),
+ s"CURRENT_SCHEMA in PATH should reflect USE; got: $entries")
+ } finally {
+ sql("USE spark_catalog.default")
+ sql("DROP SCHEMA IF EXISTS path_virt_schema")
+ }
+ }
+ }
+
+ test("PATH enabled: duplicate after expanding CURRENT_SCHEMA") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when CURRENT_SCHEMA repeated") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_schema, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when SYSTEM_PATH listed twice") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = SYSTEM_PATH, SYSTEM_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = SYSTEM_PATH includes system.builtin and system.session") {
+ withPathEnabled {
+ sql("SET PATH = SYSTEM_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"SYSTEM_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"SYSTEM_PATH should include system.session; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH on unset session includes defaults") {
+ withPathEnabled {
+ sql("SET PATH = PATH, spark_catalog.extra")
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"PATH on unset session should include builtin defaults; got: $entries")
+ assert(entries.contains("spark_catalog.extra"),
+ s"PATH on unset session should include appended schema; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema after DEFAULT_PATH (empty session path)") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_from_empty")
+ try {
+ sql("SET PATH = DEFAULT_PATH")
+ sql("SET PATH = PATH, spark_catalog.path_from_empty")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.path_from_empty"),
+ s"PATH after cleared path should append schema; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_from_empty")
+ }
+ }
+ }
+
+ test("PATH enabled: unqualified (1-part) schema reference is rejected") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = myschema")
+ },
+ condition = "INVALID_SQL_PATH_SCHEMA_REFERENCE",
+ parameters = Map(
+ "qualifiedName" -> "myschema"))
+ }
+ }
+
+ test("PATH enabled: multi-level namespace (3+ parts) is accepted") {
+ withPathEnabled {
+ sql("SET PATH = iceberg_cat.db1.db2, spark_catalog.default")
+ val entries = pathEntries(currentPath())
+ assert(entries.head === "iceberg_cat.db1.db2",
+ s"Multi-level namespace should be accepted; got: $entries")
+ }
+ }
+
+ test("PATH enabled: backtick-quoted identifiers with dots round-trip correctly") {
+ withPathEnabled {
+ sql("SET PATH = `cat.a`.`sch.b`")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("`cat.a`.`sch.b`"),
+ s"Backtick-quoted identifiers should round-trip; got: $entries")
+ }
+ }
+
+ test("PATH enabled: stored path preserves typed case, resolution is case-insensitive") {
Review Comment:
Two related gaps on this test:
- The name claims "resolution is case-insensitive" but the body only verifies case preservation on the stored/displayed path — resolution isn't wired yet, and even the case-insensitive **duplicate detection** that *is* implemented (`SetPathCommand.run:62-66`) is untested. Either trim the name to "stored path preserves typed case" or actually assert the case-insensitive claim.
- Under default `spark.sql.caseSensitive=false`, `SET PATH = spark_catalog.DEFAULT, spark_catalog.default` should raise `DUPLICATE_SQL_PATH_ENTRY`. Worth adding alongside the existing duplicate tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3103863892
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
Review Comment:
Done. Added both tests:
1. **Enabled-but-unset fallback**: `PATH_ENABLED=true` with no `SET PATH` verifies `CURRENT_PATH()` falls back to legacy `resolutionSearchPath` (contains builtin and current schema).
2. **DEFAULT_PATH + explicit builtin duplicate**: `SET PATH = DEFAULT_PATH, system.builtin` raises `DUPLICATE_SQL_PATH_ENTRY` since `DEFAULT_PATH` already expands to include `system.builtin`.
Suite now has 24 tests, all passing.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3119629930
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val cloned = spark.cloneSession()
+ val clonedPath = cloned.sql("SELECT current_path()").collect().head.getString(0)
+ val entries = pathEntries(clonedPath)
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Cloned session should inherit parent path; got: $entries")
+ }
+ }
+
+ test("PATH enabled: duplicate path entry raises error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, spark_catalog.default")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema appends to path") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_append_test")
+ try {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ sql("SET PATH = PATH, spark_catalog.path_append_test")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin",
+ "spark_catalog.path_append_test"),
+ s"PATH, schema should append; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_append_test")
+ }
+ }
+ }
+
+ test("PATH enabled: SET PATH = CURRENT_SCHEMA / CURRENT_DATABASE") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"current_schema should expand to current schema, got: $entries")
+ sql("SET PATH = current_database, system.builtin")
+ val entries2 = pathEntries(currentPath())
+ assert(entries2 === Seq("spark_catalog.default", "system.builtin"),
+ s"current_database should expand to current schema, got: $entries2")
+ }
+ }
+
+ test("PATH enabled: cross-alias duplicate detection (current_database, current_schema)") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_database, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: virtual CURRENT_SCHEMA expands to USE schema") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_virt_schema")
+ try {
+ sql("USE spark_catalog.path_virt_schema")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.path_virt_schema", "system.builtin"),
+ s"CURRENT_SCHEMA in PATH should reflect USE; got: $entries")
+ } finally {
+ sql("USE spark_catalog.default")
+ sql("DROP SCHEMA IF EXISTS path_virt_schema")
+ }
+ }
+ }
+
+ test("PATH enabled: duplicate after expanding CURRENT_SCHEMA") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when CURRENT_SCHEMA repeated") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_schema, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when SYSTEM_PATH listed twice") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = SYSTEM_PATH, SYSTEM_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = SYSTEM_PATH includes system.builtin and system.session") {
+ withPathEnabled {
+ sql("SET PATH = SYSTEM_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"SYSTEM_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"SYSTEM_PATH should include system.session; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH on unset session includes defaults") {
+ withPathEnabled {
+ sql("SET PATH = PATH, spark_catalog.extra")
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"PATH on unset session should include builtin defaults; got: $entries")
+ assert(entries.contains("spark_catalog.extra"),
+ s"PATH on unset session should include appended schema; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema after DEFAULT_PATH (empty session path)") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_from_empty")
+ try {
+ sql("SET PATH = DEFAULT_PATH")
+ sql("SET PATH = PATH, spark_catalog.path_from_empty")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.path_from_empty"),
+ s"PATH after cleared path should append schema; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_from_empty")
+ }
+ }
+ }
+
+ test("PATH enabled: unqualified (1-part) schema reference is rejected") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = myschema")
+ },
+ condition = "INVALID_SQL_PATH_SCHEMA_REFERENCE",
+ parameters = Map(
+ "qualifiedName" -> "myschema"))
+ }
+ }
+
+ test("PATH enabled: multi-level namespace (3+ parts) is accepted") {
+ withPathEnabled {
+ sql("SET PATH = iceberg_cat.db1.db2, spark_catalog.default")
+ val entries = pathEntries(currentPath())
+ assert(entries.head === "iceberg_cat.db1.db2",
+ s"Multi-level namespace should be accepted; got: $entries")
+ }
+ }
+
+ test("PATH enabled: backtick-quoted identifiers with dots round-trip correctly") {
+ withPathEnabled {
+ sql("SET PATH = `cat.a`.`sch.b`")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("`cat.a`.`sch.b`"),
+ s"Backtick-quoted identifiers should round-trip; got: $entries")
+ }
+ }
+
+ test("PATH enabled: stored path preserves typed case, resolution is case-insensitive") {
+ withPathEnabled {
+ sql("SET PATH = Spark_Catalog.Default, System.Builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("Spark_Catalog.Default", "System.Builtin"),
+ s"Stored path should preserve case; got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH respects sessionFunctionResolutionOrder = first") {
+ withSQLConf(
+ SQLConf.PATH_ENABLED.key -> "true",
+ SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "first") {
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.head.contains("session"),
+ s"With 'first' order, session should come first; got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH respects sessionFunctionResolutionOrder = last") {
+ withSQLConf(
+ SQLConf.PATH_ENABLED.key -> "true",
+ SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "last") {
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.last.contains("session"),
+ s"With 'last' order, session should come last; got: $entries")
+ }
+ }
+
+ // Note: cloneSession() shares the parent's CatalogManager, so SET PATH on either session
+ // affects both. This matches the behavior of USE CATALOG / USE SCHEMA on cloned sessions.
+ // Independence would require cloneSession to create a separate CatalogManager.
Review Comment:
Done. Replaced the note with the suggested TODO about auditing CatalogManager state propagation on clone.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -204,4 +241,33 @@ private[sql] object CatalogManager {
(nameParts.length == 2 && nameParts.head.equalsIgnoreCase(SESSION_NAMESPACE)) ||
isFullyQualifiedSystemSessionViewName(nameParts)
}
+
+ /** Virtual marker for CURRENT_SCHEMA / CURRENT_DATABASE in SET PATH. */
+ val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
Review Comment:
Moot — the `concreteSessionPathEntry` method was deleted entirely in the ADT refactor. `SessionPathEntry.resolve` now serves both duplicate checking and display-time expansion.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,373 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val cloned = spark.cloneSession()
+ val clonedPath = cloned.sql("SELECT current_path()").collect().head.getString(0)
+ val entries = pathEntries(clonedPath)
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Cloned session should inherit parent path; got: $entries")
+ }
+ }
+
+ test("PATH enabled: duplicate path entry raises error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, spark_catalog.default")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema appends to path") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_append_test")
+ try {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ sql("SET PATH = PATH, spark_catalog.path_append_test")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin",
+ "spark_catalog.path_append_test"),
+ s"PATH, schema should append; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_append_test")
+ }
+ }
+ }
+
+ test("PATH enabled: SET PATH = CURRENT_SCHEMA / CURRENT_DATABASE") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"current_schema should expand to current schema, got: $entries")
+ sql("SET PATH = current_database, system.builtin")
+ val entries2 = pathEntries(currentPath())
+ assert(entries2 === Seq("spark_catalog.default", "system.builtin"),
+ s"current_database should expand to current schema, got: $entries2")
+ }
+ }
+
+ test("PATH enabled: cross-alias duplicate detection (current_database, current_schema)") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_database, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: virtual CURRENT_SCHEMA expands to USE schema") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_virt_schema")
+ try {
+ sql("USE spark_catalog.path_virt_schema")
+ sql("SET PATH = current_schema, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.path_virt_schema", "system.builtin"),
+ s"CURRENT_SCHEMA in PATH should reflect USE; got: $entries")
+ } finally {
+ sql("USE spark_catalog.default")
+ sql("DROP SCHEMA IF EXISTS path_virt_schema")
+ }
+ }
+ }
+
+ test("PATH enabled: duplicate after expanding CURRENT_SCHEMA") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.default, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when CURRENT_SCHEMA repeated") {
+ withPathEnabled {
+ sql("USE spark_catalog.default")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = current_schema, current_schema")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "spark_catalog.default"))
+ }
+ }
+
+ test("PATH enabled: duplicate when SYSTEM_PATH listed twice") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = SYSTEM_PATH, SYSTEM_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH = SYSTEM_PATH includes system.builtin and system.session") {
+ withPathEnabled {
+ sql("SET PATH = SYSTEM_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"SYSTEM_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"SYSTEM_PATH should include system.session; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH on unset session includes defaults") {
+ withPathEnabled {
+ sql("SET PATH = PATH, spark_catalog.extra")
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"PATH on unset session should include builtin defaults; got: $entries")
+ assert(entries.contains("spark_catalog.extra"),
+ s"PATH on unset session should include appended schema; got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = PATH, schema after DEFAULT_PATH (empty session path)") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_from_empty")
+ try {
+ sql("SET PATH = DEFAULT_PATH")
+ sql("SET PATH = PATH, spark_catalog.path_from_empty")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.path_from_empty"),
+ s"PATH after cleared path should append schema; got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_from_empty")
+ }
+ }
+ }
+
+ test("PATH enabled: unqualified (1-part) schema reference is rejected") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = myschema")
+ },
+ condition = "INVALID_SQL_PATH_SCHEMA_REFERENCE",
+ parameters = Map(
+ "qualifiedName" -> "myschema"))
+ }
+ }
+
+ test("PATH enabled: multi-level namespace (3+ parts) is accepted") {
+ withPathEnabled {
+ sql("SET PATH = iceberg_cat.db1.db2, spark_catalog.default")
+ val entries = pathEntries(currentPath())
+ assert(entries.head === "iceberg_cat.db1.db2",
+ s"Multi-level namespace should be accepted; got: $entries")
+ }
+ }
+
+ test("PATH enabled: backtick-quoted identifiers with dots round-trip correctly") {
+ withPathEnabled {
+ sql("SET PATH = `cat.a`.`sch.b`")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("`cat.a`.`sch.b`"),
+ s"Backtick-quoted identifiers should round-trip; got: $entries")
+ }
+ }
+
+ test("PATH enabled: stored path preserves typed case, resolution is case-insensitive") {
Review Comment:
Done. Renamed test to "stored path preserves typed case" (removed unsubstantiated "resolution is case-insensitive" claim). Added new test: `SET PATH = spark_catalog.DEFAULT, spark_catalog.default` raises `DUPLICATE_SQL_PATH_ENTRY` under default `caseSensitive=false`. 28 tests pass.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "dtenedor (via GitHub)" <gi...@apache.org>.
dtenedor closed pull request #55364: [SPARK-56501] SET PATH syntax
URL: https://github.com/apache/spark/pull/55364
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3117433053
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8352,29 +8362,44 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
*/
def prioritizeSystemCatalog: Boolean = !getConf(SQLConf.PERSISTENT_CATALOG_FIRST)
+ def pathEnabled: Boolean = getConf(SQLConf.PATH_ENABLED)
+
/**
* Returns the resolution search path for error messages and resolution order.
* This is the single source of truth for the search path used for functions, tables, and views.
* Uses [[sessionFunctionResolutionOrder]]: "first" (session first), "second" (session second),
* "last" (session last). When catalogPath is empty, returns only system namespaces.
*/
- def resolutionSearchPath(catalogPath: Seq[String]): Seq[Seq[String]] = {
+ def resolutionSearchPath(catalogPath: Seq[String]): Seq[Seq[String]] =
+ defaultPathOrder(Seq(catalogPath))
Review Comment:
After the refactor, `resolutionSearchPath(Seq.empty)` goes through `defaultPathOrder(Seq(Seq.empty))` → `catalogEntries.isEmpty` is false → else branch → returns `Seq(builtin, session) ++ Seq(Seq.empty)`, i.e. a trailing empty-`Seq` entry instead of the 2-entry system-only path the doc at line 8371 promises. No current caller passes an empty `catalogPath` (they all build `currentCatalog.name +: currentNamespace`), so this is latent — but either the doc should be updated or the implementation guarded, e.g.:
```suggestion
def resolutionSearchPath(catalogPath: Seq[String]): Seq[Seq[String]] =
defaultPathOrder(if (catalogPath.isEmpty) Seq.empty else Seq(catalogPath))
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -126,6 +126,42 @@ class CatalogManager(
_currentNamespace = Some(namespace)
}
+ private var _sessionPath: Option[Seq[Seq[String]]] = None
Review Comment:
Storing `Option[Seq[Seq[String]]]` + a `"system.current_schema"` sentinel string forces every consumer (here `currentPathString`, and the future resolution consumer) to call `isVirtualCurrentSchemaPathEntry` and disambiguate. Since parse time already has a typed `PathElement` ADT, consider keeping structured storage — e.g.:
```scala
sealed trait SessionPathEntry
case object CurrentSchemaMarker extends SessionPathEntry
case class LiteralEntry(parts: Seq[String]) extends SessionPathEntry
private var _sessionPath: Option[Seq[SessionPathEntry]] = None
```
That deletes `SESSION_PATH_VIRTUAL_CURRENT_SCHEMA`, `isVirtualCurrentSchemaPathEntry`, `concreteSessionPathEntry`, and `expandSessionPathMarkers` in one shot, and the follow-up resolution PR gets an exhaustive match instead of a string check. Happy to leave the current form if you'd prefer to keep this change focused — just wanted to raise it before more consumers land.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" ->
+ concrete.map(p => if (p.contains(".")) s"`$p`" else p).mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
+ catalogManager.clearSessionPath()
+ } else {
+ catalogManager.setSessionPath(expanded)
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ catalogManager: CatalogManager,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ val currentSchema = Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+ conf.defaultPathOrder(Seq(currentSchema))
+ case PathElement.SystemPath =>
+ conf.systemPathOrder
+ case PathElement.CurrentDatabase | PathElement.CurrentSchema =>
+ Seq(Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA))
+ case PathElement.PathRef =>
+ catalogManager.sessionPathEntries.getOrElse(Seq.empty)
Review Comment:
`PathRef` falling back to `Seq.empty` when no path is stored is a silent foot-gun: `SET PATH = PATH, spark_catalog.foo` on an enabled-but-unset session clobbers the defaults and yields just `[spark_catalog.foo]`. Once resolution is wired in the follow-up, builtin functions would stop resolving. Consider either (a) raising an error ("PATH shortcut used but no session path is set"), or (b) expanding to the `DEFAULT_PATH` entries so the composition semantics match the read-time fallback in `currentPathString`. A companion test would lock in the chosen behavior.
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -448,6 +448,7 @@ setResetStatement
| SET TIME ZONE interval #setTimeZone
| SET TIME ZONE timezone #setTimeZone
| SET TIME ZONE .*? #setTimeZone
+ | SET PATH EQ pathElement (COMMA pathElement)* #setPath
Review Comment:
Stray double space before `#setPath` misaligns the label column relative to neighbors.
```suggestion
| SET PATH EQ pathElement (COMMA pathElement)* #setPath
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2439,6 +2439,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PATH_ENABLED =
+ buildConf("spark.sql.path.enabled")
+ .version("4.2.0")
+ .doc("When true, enables the SQL Standard PATH feature: SET PATH, path-based routine " +
+ "resolution, and CURRENT_PATH(). When false, SET PATH has no effect and resolution uses " +
+ "the default path only.")
Review Comment:
Doc says "SET PATH has no effect" when disabled, but the command throws `UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED` (as the `PATH disabled: SET PATH is rejected` test confirms).
```suggestion
.doc("When true, enables the SQL Standard PATH feature: SET PATH, path-based routine " +
"resolution, and CURRENT_PATH(). When false, SET PATH is rejected and resolution uses " +
"the default path only.")
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3098413153
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
Review Comment:
Doc says "1 = unqualified namespace, 2+ = catalog.namespace..." but `qualifySchemaParts` rejects ≥3 parts. Also "Qualified at run." is missing "time". (Note: if the ≥3-part rejection is lifted per my comment below on `qualifySchemaParts`, this doc should be rewritten to reflect multi-part namespaces.)
```suggestion
/** Schema name parts (1 = unqualified namespace, 2 = catalog.namespace). Qualified at run time. */
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
Review Comment:
`deduped` is misleading — this `flatMap` never filters (it either returns `Some(entry)` or throws). It's validation with a pass-through. Consider `expanded.foreach { … check … }` and then use `expanded` directly, or rename to `validated`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -1429,7 +1442,9 @@ class SparkSqlAstBuilder extends AstBuilder {
val tableName = ctx.identifierReference.getText.split("\\.").lastOption.getOrElse("table")
throw QueryCompilationErrors.describeJsonNotExtendedError(tableName)
}
- val relation = createUnresolvedTableOrView(ctx.identifierReference, "DESCRIBE TABLE")
+ val relation = createUnresolvedTableOrView(
+ ctx.identifierReference,
+ "DESCRIBE TABLE")
Review Comment:
This formatting change in `visitDescribeTable` is unrelated to SET PATH; please revert to keep the diff focused.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = (normalize(concrete.head),
+ normalize(concrete.lift(1).getOrElse("")))
+ if (seen.contains(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ seen.add(key)
+ Some(entry)
+ }
+
+ if (deduped.isEmpty) {
+ conf.unsetConf(SQLConf.SESSION_PATH)
+ } else {
+ conf.setConfString(SQLConf.SESSION_PATH.key, SQLConf.formatSessionPath(deduped))
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = CatalogManager.BUILTIN_NAMESPACE
+ val session = CatalogManager.SESSION_NAMESPACE
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ // Default path = session order (first/second/last). Clear path; use at resolution time.
+ Seq.empty
+ case PathElement.SystemPath =>
+ Seq(Seq(systemCatalog, builtin), Seq(systemCatalog, session))
+ case PathElement.CurrentDatabase | PathElement.CurrentSchema =>
+ Seq(Seq(systemCatalog, SQLConf.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA))
+ case PathElement.PathRef =>
+ conf.sessionPath match {
+ case Some(s) => SQLConf.parseSessionPath(s)
+ case None => Seq.empty
+ }
+ case PathElement.SchemaInPath(parts) =>
+ qualifySchemaParts(parts, systemCatalog, currentCatalog)
+ }
+ }
+
+ /** Qualify schema parts at SET time: well-known -> SYSTEM; else current catalog + namespace. */
+ private def qualifySchemaParts(
+ parts: Seq[String],
+ systemCatalog: String,
+ currentCatalog: String): Seq[Seq[String]] = {
+ val wellKnown = Set(
+ CatalogManager.BUILTIN_NAMESPACE.toLowerCase(Locale.ROOT),
+ CatalogManager.SESSION_NAMESPACE.toLowerCase(Locale.ROOT))
+ if (parts.isEmpty) return Seq.empty
+ if (parts.length > 2) {
+ throw QueryCompilationErrors.invalidSqlPathSchemaReferenceError(parts.mkString("."))
+ }
+ if (parts.size == 1) {
+ val ns = parts.head
+ if (wellKnown.contains(ns.toLowerCase(Locale.ROOT))) {
+ Seq(Seq(systemCatalog, ns))
+ } else {
+ Seq(Seq(currentCatalog, ns))
+ }
+ } else {
+ Seq(Seq(parts.head, parts(1)))
+ }
+ }
Review Comment:
Two concerns:
- The ≥3-part rejection is too strict. Spark supports multi-level namespaces (e.g., `iceberg_cat.db1.db2.schema`); a `SET PATH` entry should accept any arity ≥ 2.
- The 1-part form is a SET-time capture trap: `USE CATALOG A; SET PATH = myschema; USE CATALOG B` silently freezes the entry as `A.myschema`, invisible in the stored state. `CURRENT_SCHEMA`/`CURRENT_DATABASE` already cover the dynamic case and `SYSTEM_PATH` covers the SYSTEM shortcut, so 1-part adds a footgun without functional gain.
If 1-part is disallowed, the whole method collapses to an arity check — `parts.head` is the catalog and `parts.tail` is the namespace by convention, no inline qualification or `builtin`/`session` → SYSTEM remap needed:
```scala
private def qualifySchemaParts(parts: Seq[String]): Seq[Seq[String]] = {
if (parts.length < 2) {
throw QueryCompilationErrors.invalidSqlPathSchemaReferenceError(parts.mkString("."))
}
Seq(parts)
}
```
Could you confirm whether 1-part is intentional, and what the intended semantics are against later `USE CATALOG`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a schema (optionally qualified).
+ * For SchemaInPath(parts), qualification with current catalog or SYSTEM is done at run time.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Schema name parts (1 = unqualified namespace, 2+ = catalog.namespace...). Qualified at run. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, currentCatalog, currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[(String, String)]
+ val deduped = expanded.flatMap { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = (normalize(concrete.head),
+ normalize(concrete.lift(1).getOrElse("")))
+ if (seen.contains(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ seen.add(key)
+ Some(entry)
+ }
+
+ if (deduped.isEmpty) {
+ conf.unsetConf(SQLConf.SESSION_PATH)
+ } else {
+ conf.setConfString(SQLConf.SESSION_PATH.key, SQLConf.formatSessionPath(deduped))
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = CatalogManager.BUILTIN_NAMESPACE
+ val session = CatalogManager.SESSION_NAMESPACE
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ // Default path = session order (first/second/last). Clear path; use at resolution time.
+ Seq.empty
Review Comment:
`DEFAULT_PATH` expanding to `Seq.empty` means it silently disappears in combined forms: `SET PATH = spark_catalog.foo, DEFAULT_PATH` stores only `spark_catalog.foo`, and `SET PATH = DEFAULT_PATH, DEFAULT_PATH` succeeds without the duplicate error that `SYSTEM_PATH, SYSTEM_PATH` raises. Either restrict `DEFAULT_PATH` to the sole-element form (reject otherwise) or expand it into its concrete entries so it participates in the rest of the pipeline like `SYSTEM_PATH`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
Review Comment:
"session catalog" is a specific Spark term (`spark_catalog` / `SESSION_CATALOG_NAME`); here the expansion uses the **current** catalog.
```suggestion
* It is stored literally and expanded to the current catalog + namespace when building routine
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ private[sql] def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
+ private[sql] def concreteSessionPathEntry(
+ entry: Seq[String],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[String] = {
+ if (isVirtualCurrentSchemaPathEntry(entry)) {
+ if (currentNamespace.isEmpty) Seq(currentCatalog)
+ else currentCatalog +: currentNamespace
+ } else {
+ entry
+ }
+ }
+
+ /** Expand markers in stored session path using the current catalog and namespace. */
+ private[sql] def expandSessionPathMarkers(
+ entries: Seq[Seq[String]],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] =
+ entries.map(concreteSessionPathEntry(_, currentCatalog, currentNamespace))
+
+ /**
+ * Separator between path entries in serialized session path
+ * (catalog.namespace,catalog.namespace).
+ */
+ private[sql] val SESSION_PATH_ENTRY_SEPARATOR = ","
+
+ /**
+ * Parses a session path string into a list of path entries.
+ * Format: "catalog1.namespace1,catalog2.namespace2"
+ * (comma-separated, each entry catalog.namespace).
+ */
+ private[sql] def parseSessionPath(pathStr: String): Seq[Seq[String]] = {
+ if (pathStr == null || pathStr.trim.isEmpty) return Seq.empty
+ pathStr.split(SESSION_PATH_ENTRY_SEPARATOR).map { entry =>
+ val trimmed = entry.trim
+ val dot = trimmed.indexOf('.')
+ if (dot <= 0 || dot == trimmed.length - 1) {
+ Seq(trimmed)
+ } else {
+ Seq(trimmed.substring(0, dot).trim, trimmed.substring(dot + 1).trim)
+ }
+ }.toSeq.filter(_.nonEmpty)
+ }
+
+ /**
+ * Formats path entries to session path string.
Review Comment:
Minor grammar.
```suggestion
* Formats path entries into a session path string.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -357,6 +357,19 @@ class SparkSqlAstBuilder extends AstBuilder {
* SET TIME ZONE INTERVAL 10 HOURS;
* }}}
*/
+ override def visitSetPath(ctx: SetPathContext): LogicalPlan = withOrigin(ctx) {
+ val elements = (0 until ctx.pathElement().size()).map { i =>
+ val pe = ctx.pathElement(i)
+ if (pe.DEFAULT_PATH() != null) PathElement.DefaultPath
+ else if (pe.SYSTEM_PATH() != null) PathElement.SystemPath
+ else if (pe.PATH() != null) PathElement.PathRef
+ else if (pe.CURRENT_DATABASE() != null) PathElement.CurrentDatabase
+ else if (pe.CURRENT_SCHEMA() != null) PathElement.CurrentSchema
+ else PathElement.SchemaInPath(visitMultipartIdentifier(pe.multipartIdentifier()))
+ }
Review Comment:
Idiomatic Scala for ANTLR list access:
```suggestion
val elements = ctx.pathElement().asScala.map { pe =>
if (pe.DEFAULT_PATH() != null) PathElement.DefaultPath
else if (pe.SYSTEM_PATH() != null) PathElement.SystemPath
else if (pe.PATH() != null) PathElement.PathRef
else if (pe.CURRENT_DATABASE() != null) PathElement.CurrentDatabase
else if (pe.CURRENT_SCHEMA() != null) PathElement.CurrentSchema
else PathElement.SchemaInPath(visitMultipartIdentifier(pe.multipartIdentifier()))
}.toSeq
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -953,10 +953,9 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
* Raises TABLE_OR_VIEW_NOT_FOUND with a formatted list of resolution path entries.
*
* @param quotedResolutionPathEntries each string is already a display-ready path entry (typically
- * `toSQLId` of each segment from `SQLConf.resolutionSearchPath`). They are joined with
- * ", " inside square brackets. This differs from [[noSuchTableError]](nameParts:
- * Seq[String]), which builds one dotted bracketed path from `nameParts.dropRight(1)` via
- * [[org.apache.spark.sql.catalyst.analysis.NoSuchTableException]].
+ * `toSQLId` of each path entry from `SQLConf.sqlResolutionPathEntries`). They are joined
+ * with ", " inside square brackets, same as
+ * [[org.apache.spark.sql.catalyst.analysis.NoSuchItemExceptionHelper.formatSearchPath]].
Review Comment:
Two problems: `SQLConf.sqlResolutionPathEntries` does not exist (the method is `resolutionSearchPath`), and the "same as `NoSuchItemExceptionHelper.formatSearchPath`" claim is wrong — `formatSearchPath` joins parts with `.` to produce a single bracketed dotted path (`` [`cat`.`ns`] ``), while this method joins a list of already-dotted entries with `", "` (`` [`cat`.`ns1`, `cat`.`ns2`] ``). The old doc correctly described this as a difference.
```suggestion
* @param quotedResolutionPathEntries each string is already a display-ready path entry (typically
* `toSQLId` of each segment from `SQLConf.resolutionSearchPath`). They are joined with
* ", " inside square brackets. This differs from
* [[org.apache.spark.sql.catalyst.analysis.NoSuchItemExceptionHelper.formatSearchPath]],
* which builds one dotted bracketed path from a single search path.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala:
##########
@@ -119,6 +119,11 @@ case class SetCommand(kv: Option[(String, Option[String])])
messageParameters = Map("variableName" -> toSQLId(varName)))
}
}
+ if (key == SQLConf.SESSION_PATH.key) {
Review Comment:
This guard only covers the SQL `SET` statement. `RESET spark.sql.session.path`, bare `RESET`, and programmatic `spark.conf.set/unset("spark.sql.session.path", …)` (via `RuntimeConfig`) all bypass it — the invariant "only settable via `SET PATH`" is not actually enforced. More broadly, could you describe the rationale for storing the session path as a `SQLConf` string rather than as structured state on `CatalogManager` (alongside `currentCatalog`/`currentNamespace`)? Structured storage would remove the need for this guard entirely, and also fix the serialization format concern on `parseSessionPath`/`formatSessionPath`.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
Review Comment:
A few gaps in this suite:
- Most tests assert via `.contains(substring)`, which is too weak — e.g., the primary "SET PATH and CURRENT_PATH()" test doesn't verify `system.builtin` is in the path. Prefer checking the exact stored/displayed path (split by `,`, compare entries).
- Missing test for cross-alias duplicate detection: `SET PATH = current_database, current_schema` should raise `DUPLICATE_SQL_PATH_ENTRY` since both map to the same virtual entry.
- Missing test for `SET PATH = spark_catalog.foo, DEFAULT_PATH` (documents intended semantics — see the `DEFAULT_PATH` comment).
- Missing test for multi-level namespaces (`SET PATH = iceberg_cat.db1.db2.schema`) — currently rejected, would expose the `qualifySchemaParts` comment.
- Missing test for backtick-quoted identifiers containing `.` or `,` (would expose the `parseSessionPath`/`formatSessionPath` comment).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,71 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in `spark.sql.session.path`.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the session catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
+
+ /** True if this path entry is the virtual current-schema slot (`system.current_schema`). */
+ private[sql] def isVirtualCurrentSchemaPathEntry(entry: Seq[String]): Boolean =
+ entry.length == 2 &&
+ entry.head.equalsIgnoreCase(SYSTEM_CATALOG_NAME) &&
+ entry(1).equalsIgnoreCase(SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+
+ /** Materialize virtual current-schema entry for duplicate checks at SET PATH time. */
+ private[sql] def concreteSessionPathEntry(
+ entry: Seq[String],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[String] = {
+ if (isVirtualCurrentSchemaPathEntry(entry)) {
+ if (currentNamespace.isEmpty) Seq(currentCatalog)
+ else currentCatalog +: currentNamespace
+ } else {
+ entry
+ }
+ }
+
+ /** Expand markers in stored session path using the current catalog and namespace. */
+ private[sql] def expandSessionPathMarkers(
+ entries: Seq[Seq[String]],
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] =
+ entries.map(concreteSessionPathEntry(_, currentCatalog, currentNamespace))
+
+ /**
+ * Separator between path entries in serialized session path
+ * (catalog.namespace,catalog.namespace).
+ */
+ private[sql] val SESSION_PATH_ENTRY_SEPARATOR = ","
+
+ /**
+ * Parses a session path string into a list of path entries.
+ * Format: "catalog1.namespace1,catalog2.namespace2"
+ * (comma-separated, each entry catalog.namespace).
+ */
+ private[sql] def parseSessionPath(pathStr: String): Seq[Seq[String]] = {
+ if (pathStr == null || pathStr.trim.isEmpty) return Seq.empty
+ pathStr.split(SESSION_PATH_ENTRY_SEPARATOR).map { entry =>
+ val trimmed = entry.trim
+ val dot = trimmed.indexOf('.')
+ if (dot <= 0 || dot == trimmed.length - 1) {
+ Seq(trimmed)
+ } else {
+ Seq(trimmed.substring(0, dot).trim, trimmed.substring(dot + 1).trim)
+ }
+ }.toSeq.filter(_.nonEmpty)
+ }
+
+ /**
+ * Formats path entries to session path string.
+ * Each entry is catalog.namespace; entries separated by comma.
+ */
+ private[sql] def formatSessionPath(pathEntries: Seq[Seq[String]]): String =
+ pathEntries.map(_.mkString(".")).mkString(SESSION_PATH_ENTRY_SEPARATOR)
Review Comment:
The string serialization is lossy on two dimensions:
- Multi-level namespaces cannot round-trip: `formatSessionPath(Seq(Seq("cat", "db1", "db2")))` produces `"cat.db1.db2"`, which `parseSessionPath` splits at the *first* `.` back to `Seq("cat", "db1.db2")`. Given Spark supports multi-level namespaces (e.g., Iceberg), this isn't an edge case.
- Backtick-quoted identifiers containing `.` or `,` also round-trip incorrectly (``SET PATH = `cat.a`.`sch` ``, or any entry containing `,`).
Both break subsequent duplicate checks too. Either escape `.`/`,` in the serialized form, or store the path as structured session state rather than a string config (preferred — it also addresses the `SET` guard concern in `SetCommand.scala`).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3108498301
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala:
##########
@@ -357,6 +357,18 @@ class SparkSqlAstBuilder extends AstBuilder {
* SET TIME ZONE INTERVAL 10 HOURS;
* }}}
*/
+ override def visitSetPath(ctx: SetPathContext): LogicalPlan = withOrigin(ctx) {
Review Comment:
The Scaladoc directly above (lines 351-358) documents `SET TIME ZONE` — the examples reference `SESSION_LOCAL_TIMEZONE` and three TIME ZONE forms. Inserting `visitSetPath` here detached the block from `visitSetTimeZone` (line 372, now undocumented) and implicitly reattached it to `visitSetPath`. Please move the block back above `visitSetTimeZone` and add a new Scaladoc for `visitSetPath`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
+ catalogManager.clearSessionPath()
+ } else {
+ catalogManager.setSessionPath(expanded)
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ catalogManager: CatalogManager,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+ val builtin = Seq(systemCatalog, CatalogManager.BUILTIN_NAMESPACE)
+ val session = Seq(systemCatalog, CatalogManager.SESSION_NAMESPACE)
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ val currentSchema = Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+ conf.sessionFunctionResolutionOrder match {
Review Comment:
This branch duplicates the ordering logic in `SQLConf.resolutionSearchPath` (SQLConf.scala:8373) — both must stay in sync for `CURRENT_PATH()`'s stored-vs-fallback paths to agree on `spark.sql.function.resolutionOrder`. `SYSTEM_PATH` below (line 109) has the same issue. Consider extracting a helper that returns the legacy default ordering as `Seq[Seq[String]]`, with the current-schema slot supplied by the caller (virtual marker here; actual `catalogPath` for `resolutionSearchPath`). That removes the drift risk.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
Review Comment:
All DEFAULT_PATH/SYSTEM_PATH tests run with the default `sessionFunctionResolutionOrder = "second"`. The `"first"` and `"last"` branches in `expandPathElements` are uncovered — adding two parametric tests (e.g. via `withSQLConf(SESSION_FUNCTION_RESOLUTION_ORDER.key -> "first")`) that assert the expected entry order would lock them in before the resolution engine starts consuming the stored path.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Tests for SET PATH command and session path management.
+ * Covers the feature flag, SET PATH syntax, CURRENT_PATH() reflecting stored path,
+ * DEFAULT_PATH, SYSTEM_PATH, CURRENT_SCHEMA/CURRENT_DATABASE expansion,
+ * PATH (append), duplicate detection, and error conditions.
+ *
+ * Resolution-level tests (tables/functions resolving via the stored path)
+ * belong in a separate suite once the resolution engine is wired.
+ */
+class SetPathSuite extends QueryTest with SharedSparkSession {
+
+ private def withPathEnabled(f: => Unit): Unit = {
+ withSQLConf(SQLConf.PATH_ENABLED.key -> "true")(f)
+ }
+
+ private def currentPath(): String =
+ sql("SELECT current_path()").collect().head.getString(0)
+
+ private def pathEntries(pathStr: String): Seq[String] =
+ pathStr.split(",").map(_.trim).toSeq
+
+ test("PATH disabled: CURRENT_PATH() returns default path") {
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"Expected default path to contain spark_catalog.default, got: $entries")
+ assert(entries.exists(_.contains("builtin")),
+ s"Expected default path to contain builtin, got: $entries")
+ }
+
+ test("PATH disabled: SET PATH is rejected") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = spark_catalog.other")
+ },
+ condition = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ sqlState = "0A000",
+ parameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+
+ test("PATH enabled but no SET PATH: falls back to legacy resolutionSearchPath") {
+ withPathEnabled {
+ val entries = pathEntries(currentPath())
+ assert(entries.exists(_.contains("builtin")),
+ s"Enabled-but-unset should fall back to legacy path with builtin, got: $entries")
+ assert(entries.exists(_.contains("default")),
+ s"Enabled-but-unset should include current schema, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH + explicit builtin raises duplicate") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, system.builtin")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("PATH enabled: SET PATH and CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Expected exact path entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: SET PATH = DEFAULT_PATH restores default") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default")
+ sql("SET PATH = DEFAULT_PATH")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("spark_catalog.default"),
+ s"After SET PATH = DEFAULT_PATH expected current schema in path, got: $entries")
+ assert(entries.contains("system.builtin"),
+ s"After SET PATH = DEFAULT_PATH expected system.builtin, got: $entries")
+ assert(entries.contains("system.session"),
+ s"After SET PATH = DEFAULT_PATH expected system.session, got: $entries")
+ assert(entries.length === 3,
+ s"DEFAULT_PATH should expand to 3 entries, got: $entries")
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH composes with other path elements") {
+ withPathEnabled {
+ sql("CREATE SCHEMA IF NOT EXISTS path_extra_test")
+ try {
+ sql("SET PATH = DEFAULT_PATH, spark_catalog.path_extra_test")
+ val entries = pathEntries(currentPath())
+ assert(entries.contains("system.builtin"),
+ s"DEFAULT_PATH should include system.builtin; got: $entries")
+ assert(entries.contains("system.session"),
+ s"DEFAULT_PATH should include system.session; got: $entries")
+ assert(entries.last === "spark_catalog.path_extra_test",
+ s"Extra schema should be appended after DEFAULT_PATH; got: $entries")
+ assert(entries.length === 4,
+ s"DEFAULT_PATH + 1 extra should be 4 entries, got: $entries")
+ } finally {
+ sql("DROP SCHEMA IF EXISTS path_extra_test")
+ }
+ }
+ }
+
+ test("PATH enabled: DEFAULT_PATH, DEFAULT_PATH raises duplicate error") {
+ withPathEnabled {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SET PATH = DEFAULT_PATH, DEFAULT_PATH")
+ },
+ condition = "DUPLICATE_SQL_PATH_ENTRY",
+ sqlState = Some("42732"),
+ parameters = Map("pathEntry" -> "system.builtin"))
+ }
+ }
+
+ test("programmatic SET of spark.sql.session.path has no effect on CURRENT_PATH()") {
+ withPathEnabled {
+ sql("SET PATH = spark_catalog.default, system.builtin")
+ spark.conf.set("spark.sql.session.path", "garbage")
+ val entries = pathEntries(currentPath())
+ assert(entries === Seq("spark_catalog.default", "system.builtin"),
+ s"Programmatic SET should not affect path; got: $entries")
+ spark.conf.unset("spark.sql.session.path")
+ }
+ }
+
+ test("PATH enabled: cloned session inherits parent path") {
Review Comment:
The clone test verifies that a clone inherits the parent's path. Consider also asserting the inverse — after cloning, a `SET PATH` on the parent must not propagate to the child, and vice versa. Documents the intended one-shot-snapshot semantics of `copySessionPathFrom`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -204,4 +241,39 @@ private[sql] object CatalogManager {
(nameParts.length == 2 && nameParts.head.equalsIgnoreCase(SESSION_NAMESPACE)) ||
isFullyQualifiedSystemSessionViewName(nameParts)
}
+
+ /** True if a SQL path entry denotes `system.session` (case-insensitive). */
+ def isSystemSessionPathEntry(parts: Seq[String]): Boolean =
Review Comment:
`isSystemSessionPathEntry` is added here but has no call sites in the PR — a grep finds only the definition. If it's meant as a future hook, a Scaladoc note would help; otherwise drop it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
Review Comment:
`concrete.mkString(".")` loses backtick quoting — if a user writes ``SET PATH = `cat.a`.`sch.b`, `cat.a`.`sch.b` ``, the duplicate-entry error reports `pathEntry = "cat.a.sch.b"` instead of the quoted form they typed, which is ambiguous with a 3-part namespace. Consider `toSQLId`-style quoting (or backticking each part when it contains `.`) so the rendered entry matches input. Low priority — diagnostic quality only.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3119153305
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ CatalogManager.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" ->
+ concrete.map(p => if (p.contains(".")) s"`$p`" else p).mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
+ catalogManager.clearSessionPath()
+ } else {
+ catalogManager.setSessionPath(expanded)
+ }
+ Seq.empty
+ }
+
+ private def expandPathElements(
+ elements: Seq[PathElement],
+ conf: SQLConf,
+ catalogManager: CatalogManager,
+ currentCatalog: String,
+ currentNamespace: Seq[String]): Seq[Seq[String]] = {
+ val systemCatalog = CatalogManager.SYSTEM_CATALOG_NAME
+
+ elements.flatMap {
+ case PathElement.DefaultPath =>
+ val currentSchema = Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA)
+ conf.defaultPathOrder(Seq(currentSchema))
+ case PathElement.SystemPath =>
+ conf.systemPathOrder
+ case PathElement.CurrentDatabase | PathElement.CurrentSchema =>
+ Seq(Seq(systemCatalog, CatalogManager.SESSION_PATH_VIRTUAL_CURRENT_SCHEMA))
+ case PathElement.PathRef =>
+ catalogManager.sessionPathEntries.getOrElse(Seq.empty)
Review Comment:
Done. `PathRef` on an unset session now expands to `DEFAULT_PATH` entries (via `conf.defaultPathOrder`) so builtin/session are preserved when composing. Added test: `SET PATH = PATH, spark_catalog.extra` on unset session verifies builtin defaults are included.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8352,29 +8362,44 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
*/
def prioritizeSystemCatalog: Boolean = !getConf(SQLConf.PERSISTENT_CATALOG_FIRST)
+ def pathEnabled: Boolean = getConf(SQLConf.PATH_ENABLED)
+
/**
* Returns the resolution search path for error messages and resolution order.
* This is the single source of truth for the search path used for functions, tables, and views.
* Uses [[sessionFunctionResolutionOrder]]: "first" (session first), "second" (session second),
* "last" (session last). When catalogPath is empty, returns only system namespaces.
*/
- def resolutionSearchPath(catalogPath: Seq[String]): Seq[Seq[String]] = {
+ def resolutionSearchPath(catalogPath: Seq[String]): Seq[Seq[String]] =
+ defaultPathOrder(Seq(catalogPath))
Review Comment:
Done. `resolutionSearchPath(Seq.empty)` now guards with `if (catalogPath.isEmpty) Seq.empty else Seq(catalogPath)` before passing to `defaultPathOrder`, so empty catalog path correctly returns the system-only 2-entry path.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2439,6 +2439,16 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val PATH_ENABLED =
+ buildConf("spark.sql.path.enabled")
+ .version("4.2.0")
+ .doc("When true, enables the SQL Standard PATH feature: SET PATH, path-based routine " +
+ "resolution, and CURRENT_PATH(). When false, SET PATH has no effect and resolution uses " +
+ "the default path only.")
Review Comment:
Done. Fixed: "SET PATH is rejected" instead of "has no effect".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -126,6 +126,42 @@ class CatalogManager(
_currentNamespace = Some(namespace)
}
+ private var _sessionPath: Option[Seq[Seq[String]]] = None
Review Comment:
Good suggestion. Leaving the current string-sentinel form for now to keep this PR focused — but agree the typed ADT (`CurrentSchemaMarker` / `LiteralEntry`) is cleaner and will consider it when more consumers land.
##########
sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4:
##########
@@ -448,6 +448,7 @@ setResetStatement
| SET TIME ZONE interval #setTimeZone
| SET TIME ZONE timezone #setTimeZone
| SET TIME ZONE .*? #setTimeZone
+ | SET PATH EQ pathElement (COMMA pathElement)* #setPath
Review Comment:
Done. Fixed alignment.
##########
sql/core/src/test/scala/org/apache/spark/sql/SetPathSuite.scala:
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.spark.sql
+
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
Review Comment:
Done in the prior commit. Tests for `"first"` and `"last"` sessionFunctionResolutionOrder are in place.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56501] SET PATH syntax [spark]
Posted by "srielau (via GitHub)" <gi...@apache.org>.
srielau commented on code in PR #55364:
URL: https://github.com/apache/spark/pull/55364#discussion_r3103576612
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala:
##########
@@ -152,17 +152,16 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
* Replaces the expression of CurrentDatabase, CurrentCatalog, CurrentPath, and CurrentUser
* with the current values.
*/
-case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
+case class ReplaceCurrentLike(
+ catalogManager: CatalogManager,
+ sqlConf: SQLConf) extends Rule[LogicalPlan] {
Review Comment:
Done. Removed `sqlConf` parameter from `ReplaceCurrentLike` — it now takes only `catalogManager`. Also removed the `conf` argument in `Optimizer.scala`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala:
##########
@@ -152,17 +152,16 @@ object ComputeCurrentTime extends Rule[LogicalPlan] {
* Replaces the expression of CurrentDatabase, CurrentCatalog, CurrentPath, and CurrentUser
* with the current values.
*/
-case class ReplaceCurrentLike(catalogManager: CatalogManager) extends Rule[LogicalPlan] {
+case class ReplaceCurrentLike(
+ catalogManager: CatalogManager,
+ sqlConf: SQLConf) extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = {
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
lazy val currentNamespace = catalogManager.currentNamespace.quoted
lazy val currentNamespaceSeq = catalogManager.currentNamespace.toSeq
Review Comment:
Done. Removed `currentNamespaceSeq` — no longer needed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -126,6 +126,42 @@ class CatalogManager(
_currentNamespace = Some(namespace)
}
+ private var _sessionPath: Option[Seq[Seq[String]]] = None
+
+ /** Returns the raw stored session path entries, or None if no path is set. */
+ def sessionPathEntries: Option[Seq[Seq[String]]] = synchronized { _sessionPath }
+
+ def setSessionPath(entries: Seq[Seq[String]]): Unit = synchronized {
+ _sessionPath = Some(entries)
+ }
+
+ def clearSessionPath(): Unit = synchronized {
+ _sessionPath = None
+ }
+
+ private[sql] def copySessionPathFrom(other: CatalogManager): Unit = synchronized {
+ _sessionPath = other.sessionPathEntries
+ }
+
+ /**
+ * String form of the current resolution path for CURRENT_PATH().
+ * When PATH is enabled and a session path is stored, formats the effective path entries
+ * with markers expanded. Otherwise falls back to the legacy resolutionSearchPath.
+ */
+ def currentPathString: String = {
+ import CatalogV2Implicits._
+ val entries = if (conf.pathEnabled) _sessionPath else None
Review Comment:
Done. `currentPathString` now synchronizes on `this`, consistent with all other `_sessionPath` accessors.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2061,6 +2061,12 @@
],
"sqlState" : "42711"
},
+ "DUPLICATE_SQL_PATH_ENTRY" : {
+ "message" : [
+ "Duplicate SQL path entry <pathEntry>. The session SQL path cannot contain the same catalog.schema more than once (including after expanding shortcuts like DEFAULT_PATH or SYSTEM_PATH)."
Review Comment:
Done. Message now says "the same namespace" instead of "the same catalog.schema".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -140,6 +143,40 @@ object SQLConf {
}
}
+ /**
+ * Second segment of the virtual path entry `system.current_schema` in the session path.
+ * CURRENT_SCHEMA and CURRENT_DATABASE in SET PATH both normalize to this sentinel (SQL aliases).
+ * It is stored literally and expanded to the current catalog + namespace when building routine
+ * resolution candidates and CURRENT_PATH().
+ */
+ private[sql] val SESSION_PATH_VIRTUAL_CURRENT_SCHEMA: String = "current_schema"
Review Comment:
Done. Moved `SESSION_PATH_VIRTUAL_CURRENT_SCHEMA`, `isVirtualCurrentSchemaPathEntry`, `concreteSessionPathEntry`, and `expandSessionPathMarkers` from `SQLConf` companion to `CatalogManager` companion. All callers updated.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetPathCommand.scala:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.spark.sql.execution.command
+
+import java.util.Locale
+
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Path element for SET PATH: either a well-known shortcut or a fully qualified schema reference.
+ * SchemaInPath requires at least 2 parts (catalog.namespace); multi-level namespaces are allowed.
+ */
+sealed trait PathElement
+
+object PathElement {
+ case object DefaultPath extends PathElement
+ case object SystemPath extends PathElement
+ case object PathRef extends PathElement
+ /**
+ * Current database/schema (SQL aliases). Stored as system.current_schema; expands when
+ * building resolution candidates so later USE SCHEMA is reflected.
+ */
+ case object CurrentDatabase extends PathElement
+ case object CurrentSchema extends PathElement
+ /** Fully qualified schema reference (catalog.namespace...). Must have at least 2 parts. */
+ case class SchemaInPath(parts: Seq[String]) extends PathElement
+}
+
+/**
+ * Command for SET PATH = pathElement (, pathElement)*
+ * Expands shortcuts at run time, validates no duplicates, and sets the internal session path.
+ */
+case class SetPathCommand(elements: Seq[PathElement]) extends LeafRunnableCommand {
+
+ override def output: Seq[Attribute] = Seq.empty
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ if (!sparkSession.sessionState.conf.pathEnabled) {
+ throw new AnalysisException(
+ errorClass = "UNSUPPORTED_FEATURE.SET_PATH_WHEN_DISABLED",
+ messageParameters = Map("config" -> SQLConf.PATH_ENABLED.key))
+ }
+ val conf = sparkSession.sessionState.conf
+ val catalogManager = sparkSession.sessionState.catalogManager
+ val currentCatalog = catalogManager.currentCatalog.name
+ val currentNamespace = catalogManager.currentNamespace.toSeq
+ val caseSensitive = conf.caseSensitiveAnalysis
+
+ val expanded = expandPathElements(elements, conf, catalogManager, currentCatalog,
+ currentNamespace)
+ val seen = new scala.collection.mutable.HashSet[Seq[String]]
+ expanded.foreach { entry =>
+ val concrete =
+ SQLConf.concreteSessionPathEntry(entry, currentCatalog, currentNamespace)
+ def normalize(s: String): String = if (caseSensitive) s else s.toLowerCase(Locale.ROOT)
+ val key = concrete.map(normalize)
+ if (!seen.add(key)) {
+ throw new AnalysisException(
+ errorClass = "DUPLICATE_SQL_PATH_ENTRY",
+ messageParameters = Map("pathEntry" -> concrete.mkString(".")))
+ }
+ }
+
+ if (expanded.isEmpty) {
Review Comment:
`clearSessionPath()` is indeed only reachable via `SET PATH = PATH` when no path is stored (which is a no-op). Leaving as-is for now — it's harmless and keeps the code symmetric with `setSessionPath`. A `RESET PATH` command could be added later if needed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org