You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2024/11/19 13:40:33 UTC
(spark) branch master updated: Revert "[SPARK-49002][SQL] Consistently handle invalid locations in WAREHOUSE/SCHEMA/TABLE/PARTITION/DIRECTORY"
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 19509d07983d Revert "[SPARK-49002][SQL] Consistently handle invalid locations in WAREHOUSE/SCHEMA/TABLE/PARTITION/DIRECTORY"
19509d07983d is described below
commit 19509d07983d050d0234c03760433bb67c823055
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Tue Nov 19 21:40:07 2024 +0800
Revert "[SPARK-49002][SQL] Consistently handle invalid locations in WAREHOUSE/SCHEMA/TABLE/PARTITION/DIRECTORY"
This reverts commit b0f92a9261a34defa016361b1321f634f0516347.
---
.../src/main/resources/error/error-conditions.json | 6 --
.../catalyst/catalog/ExternalCatalogUtils.scala | 13 +---
.../spark/sql/errors/QueryExecutionErrors.scala | 10 +--
.../catalyst/analysis/ResolveSessionCatalog.scala | 10 ++-
.../spark/sql/execution/datasources/rules.scala | 14 ++--
.../datasources/v2/DataSourceV2Strategy.scala | 8 ++
.../analyzer-results/sql-on-files.sql.out | 80 --------------------
.../resources/sql-tests/inputs/sql-on-files.sql | 5 --
.../sql-tests/results/sql-on-files.sql.out | 88 ----------------------
.../AlterNamespaceSetLocationSuiteBase.scala | 15 ----
10 files changed, 29 insertions(+), 220 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index e8dbac21ab76..58718cb99f55 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -2774,12 +2774,6 @@
},
"sqlState" : "42K0E"
},
- "INVALID_LOCATION" : {
- "message" : [
- "The location name cannot be an invalid URI, but `<location>` was given."
- ],
- "sqlState" : "42K05"
- },
"INVALID_NON_DETERMINISTIC_EXPRESSIONS" : {
"message" : [
"The operator expects a deterministic expression, but the actual expression is <sqlExprs>."
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala
index ec01fe3ddc37..8960c7345521 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogUtils.scala
@@ -19,7 +19,6 @@ package org.apache.spark.sql.catalyst.catalog
import java.net.URI
-import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.util.Shell
@@ -29,7 +28,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, BasePredicate, BoundReference, Expression, Predicate}
import org.apache.spark.sql.catalyst.expressions.Hex.unhexDigits
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
-import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
+import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
@@ -281,15 +280,7 @@ object CatalogUtils {
* @return the URI of the path
*/
def stringToURI(str: String): URI = {
- if (StringUtils.isEmpty(str)) {
- throw QueryExecutionErrors.invalidLocationError(str, "INVALID_EMPTY_LOCATION")
- }
- try {
- new Path(str).toUri
- } catch {
- case e: IllegalArgumentException =>
- throw QueryExecutionErrors.invalidLocationError(str, "INVALID_LOCATION", e)
- }
+ new Path(str).toUri
}
def makeQualifiedDBObjectPath(
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index ba48000f2aec..ef4fdeeaddfe 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -2538,14 +2538,10 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
)
}
- def invalidLocationError(
- location: String,
- errorClass: String,
- cause: Throwable = null): SparkIllegalArgumentException = {
+ def invalidEmptyLocationError(location: String): SparkIllegalArgumentException = {
new SparkIllegalArgumentException(
- errorClass = errorClass,
- messageParameters = Map("location" -> location),
- cause = cause)
+ errorClass = "INVALID_EMPTY_LOCATION",
+ messageParameters = Map("location" -> location))
}
def malformedProtobufMessageDetectedInMessageParsingError(e: Throwable): Throwable = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
index 884c870e8eed..92c74f7bede1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.catalyst.analysis
+import org.apache.commons.lang3.StringUtils
+
import org.apache.spark.SparkException
import org.apache.spark.internal.LogKeys.CONFIG
import org.apache.spark.internal.MDC
@@ -30,7 +32,7 @@ import org.apache.spark.sql.catalyst.util.{quoteIfNeeded, toPrettySQL, CharVarch
import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
import org.apache.spark.sql.connector.catalog.{CatalogExtension, CatalogManager, CatalogPlugin, CatalogV2Util, LookupCatalog, SupportsNamespaces, V1Table}
import org.apache.spark.sql.connector.expressions.Transform
-import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.datasources.{CreateTable => CreateTableV1}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils
@@ -137,6 +139,9 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
AlterDatabasePropertiesCommand(db, properties)
case SetNamespaceLocation(ResolvedV1Database(db), location) if conf.useV1Command =>
+ if (StringUtils.isEmpty(location)) {
+ throw QueryExecutionErrors.invalidEmptyLocationError(location)
+ }
AlterDatabaseSetLocationCommand(db, location)
case RenameTable(ResolvedV1TableOrViewIdentifier(oldIdent), newName, isView) =>
@@ -240,6 +245,9 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
val comment = c.properties.get(SupportsNamespaces.PROP_COMMENT)
val location = c.properties.get(SupportsNamespaces.PROP_LOCATION)
val newProperties = c.properties -- CatalogV2Util.NAMESPACE_RESERVED_PROPERTIES
+ if (location.isDefined && location.get.isEmpty) {
+ throw QueryExecutionErrors.invalidEmptyLocationError(location.get)
+ }
CreateDatabaseCommand(name, c.ifNotExists, location, comment, newProperties)
case d @ DropNamespace(ResolvedV1Database(db), _, _) if conf.useV1Command =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
index e1b397aaeb36..eaad99ab05f8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala
@@ -22,7 +22,6 @@ import java.util.Locale
import scala.collection.mutable.{HashMap, HashSet}
import scala.jdk.CollectionConverters._
-import org.apache.spark.SparkIllegalArgumentException
import org.apache.spark.sql.{AnalysisException, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.catalog._
@@ -55,7 +54,7 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
val ident = unresolved.multipartIdentifier
val dataSource = DataSource(
sparkSession,
- paths = Seq(CatalogUtils.stringToURI(ident.last).toString),
+ paths = Seq(ident.last),
className = ident.head,
options = unresolved.options.asScala.toMap)
// `dataSource.providingClass` may throw ClassNotFoundException, the caller side will try-catch
@@ -67,6 +66,12 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
errorClass = "UNSUPPORTED_DATASOURCE_FOR_DIRECT_QUERY",
messageParameters = Map("dataSourceType" -> ident.head))
}
+ if (isFileFormat && ident.last.isEmpty) {
+ unresolved.failAnalysis(
+ errorClass = "INVALID_EMPTY_LOCATION",
+ messageParameters = Map("location" -> ident.last))
+ }
+
dataSource
}
@@ -89,11 +94,6 @@ class ResolveSQLOnFile(sparkSession: SparkSession) extends Rule[LogicalPlan] {
LogicalRelation(ds.resolveRelation())
} catch {
case _: ClassNotFoundException => u
- case e: SparkIllegalArgumentException if e.getCondition != null =>
- u.failAnalysis(
- errorClass = e.getCondition,
- messageParameters = e.getMessageParameters.asScala.toMap,
- cause = e)
case e: Exception if !e.isInstanceOf[AnalysisException] =>
// the provider is valid, but failed to create a logical plan
u.failAnalysis(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 1c265650e02a..499721fbae4e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources.v2
import scala.collection.mutable
+import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.fs.Path
import org.apache.spark.SparkException
@@ -380,6 +381,9 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
AlterNamespaceSetPropertiesExec(catalog.asNamespaceCatalog, ns, properties) :: Nil
case SetNamespaceLocation(ResolvedNamespace(catalog, ns, _), location) =>
+ if (StringUtils.isEmpty(location)) {
+ throw QueryExecutionErrors.invalidEmptyLocationError(location)
+ }
AlterNamespaceSetPropertiesExec(
catalog.asNamespaceCatalog,
ns,
@@ -392,6 +396,10 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
Map(SupportsNamespaces.PROP_COMMENT -> comment)) :: Nil
case CreateNamespace(ResolvedNamespace(catalog, ns, _), ifNotExists, properties) =>
+ val location = properties.get(SupportsNamespaces.PROP_LOCATION)
+ if (location.isDefined && location.get.isEmpty) {
+ throw QueryExecutionErrors.invalidEmptyLocationError(location.get)
+ }
val finalProperties = properties.get(SupportsNamespaces.PROP_LOCATION).map { loc =>
properties + (SupportsNamespaces.PROP_LOCATION -> makeQualifiedDBObjectPath(loc))
}.getOrElse(properties)
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/sql-on-files.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/sql-on-files.sql.out
index 2899b6c1b0be..b098a9758fe4 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/sql-on-files.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/sql-on-files.sql.out
@@ -34,26 +34,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM parquet.`file:tmp`
--- !query analysis
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 32,
- "fragment" : "parquet.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM parquet.`/file/not/found`
-- !query analysis
@@ -109,26 +89,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM orc.`file:tmp`
--- !query analysis
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 28,
- "fragment" : "orc.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM orc.`/file/not/found`
-- !query analysis
@@ -184,26 +144,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM csv.`file:tmp`
--- !query analysis
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 28,
- "fragment" : "csv.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM csv.`/file/not/found`
-- !query analysis
@@ -259,26 +199,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM json.`file:tmp`
--- !query analysis
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 29,
- "fragment" : "json.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM json.`/file/not/found`
-- !query analysis
diff --git a/sql/core/src/test/resources/sql-tests/inputs/sql-on-files.sql b/sql/core/src/test/resources/sql-tests/inputs/sql-on-files.sql
index 2b45a5060c99..8a00e4400e6b 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/sql-on-files.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/sql-on-files.sql
@@ -2,8 +2,6 @@ CREATE DATABASE IF NOT EXISTS sql_on_files;
-- Parquet
CREATE TABLE sql_on_files.test_parquet USING PARQUET AS SELECT 1;
SELECT * FROM parquet.``;
-SELECT * FROM parquet.`file:tmp`;
-
SELECT * FROM parquet.`/file/not/found`;
SELECT * FROM parquet.`${spark.sql.warehouse.dir}/sql_on_files.db/test_parquet`;
DROP TABLE sql_on_files.test_parquet;
@@ -11,7 +9,6 @@ DROP TABLE sql_on_files.test_parquet;
-- ORC
CREATE TABLE sql_on_files.test_orc USING ORC AS SELECT 1;
SELECT * FROM orc.``;
-SELECT * FROM orc.`file:tmp`;
SELECT * FROM orc.`/file/not/found`;
SELECT * FROM orc.`${spark.sql.warehouse.dir}/sql_on_files.db/test_orc`;
DROP TABLE sql_on_files.test_orc;
@@ -19,7 +16,6 @@ DROP TABLE sql_on_files.test_orc;
-- CSV
CREATE TABLE sql_on_files.test_csv USING CSV AS SELECT 1;
SELECT * FROM csv.``;
-SELECT * FROM csv.`file:tmp`;
SELECT * FROM csv.`/file/not/found`;
SELECT * FROM csv.`${spark.sql.warehouse.dir}/sql_on_files.db/test_csv`;
DROP TABLE sql_on_files.test_csv;
@@ -27,7 +23,6 @@ DROP TABLE sql_on_files.test_csv;
-- JSON
CREATE TABLE sql_on_files.test_json USING JSON AS SELECT 1;
SELECT * FROM json.``;
-SELECT * FROM json.`file:tmp`;
SELECT * FROM json.`/file/not/found`;
SELECT * FROM json.`${spark.sql.warehouse.dir}/sql_on_files.db/test_json`;
DROP TABLE sql_on_files.test_json;
diff --git a/sql/core/src/test/resources/sql-tests/results/sql-on-files.sql.out b/sql/core/src/test/resources/sql-tests/results/sql-on-files.sql.out
index 73eef8fe74f2..fc8f44bc22fe 100644
--- a/sql/core/src/test/resources/sql-tests/results/sql-on-files.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/sql-on-files.sql.out
@@ -37,28 +37,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM parquet.`file:tmp`
--- !query schema
-struct<>
--- !query output
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 32,
- "fragment" : "parquet.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM parquet.`/file/not/found`
-- !query schema
@@ -120,28 +98,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM orc.`file:tmp`
--- !query schema
-struct<>
--- !query output
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 28,
- "fragment" : "orc.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM orc.`/file/not/found`
-- !query schema
@@ -203,28 +159,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM csv.`file:tmp`
--- !query schema
-struct<>
--- !query output
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 28,
- "fragment" : "csv.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM csv.`/file/not/found`
-- !query schema
@@ -286,28 +220,6 @@ org.apache.spark.sql.AnalysisException
}
--- !query
-SELECT * FROM json.`file:tmp`
--- !query schema
-struct<>
--- !query output
-org.apache.spark.sql.AnalysisException
-{
- "errorClass" : "INVALID_LOCATION",
- "sqlState" : "42K05",
- "messageParameters" : {
- "location" : "file:tmp"
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 15,
- "stopIndex" : 29,
- "fragment" : "json.`file:tmp`"
- } ]
-}
-
-
-- !query
SELECT * FROM json.`/file/not/found`
-- !query schema
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterNamespaceSetLocationSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterNamespaceSetLocationSuiteBase.scala
index 64491f9ad974..00484c2efc83 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterNamespaceSetLocationSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterNamespaceSetLocationSuiteBase.scala
@@ -56,21 +56,6 @@ trait AlterNamespaceSetLocationSuiteBase extends QueryTest with DDLCommandTestUt
}
}
- test("Invalid location string") {
- val ns = s"$catalog.$namespace"
- withNamespace(ns) {
- sql(s"CREATE NAMESPACE $ns")
- val sqlText = s"ALTER NAMESPACE $ns SET LOCATION 'file:tmp'"
- val e = intercept[SparkIllegalArgumentException] {
- sql(sqlText)
- }
- checkError(
- exception = e,
- condition = "INVALID_LOCATION",
- parameters = Map("location" -> "file:tmp"))
- }
- }
-
test("Namespace does not exist") {
val ns = "not_exist"
val e = intercept[AnalysisException] {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org