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 a88b509a9615a..4b0e676d720cc 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,6 +19,7 @@ package org.apache.spark.sql.catalyst.catalog import java.net.URI +import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path import org.apache.hadoop.util.Shell @@ -258,6 +259,24 @@ object CatalogUtils { new Path(str).toUri } + def makeQualifiedNamespacePath( + locationUri: URI, + warehousePath: String, + hadoopConf: Configuration): URI = { + if (locationUri.isAbsolute) { + locationUri + } else { + val fullPath = new Path(warehousePath, CatalogUtils.URIToString(locationUri)) + makeQualifiedPath(fullPath.toUri, hadoopConf) + } + } + + def makeQualifiedPath(path: URI, hadoopConf: Configuration): URI = { + val hadoopPath = new Path(path) + val fs = hadoopPath.getFileSystem(hadoopConf) + fs.makeQualified(hadoopPath).toUri + } + private def normalizeColumnName( tableName: String, tableCols: Seq[String], diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala index f529b13ff5adc..610a683e86246 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala @@ -210,9 +210,7 @@ class SessionCatalog( * FileSystem is changed. */ private def makeQualifiedPath(path: URI): URI = { - val hadoopPath = new Path(path) - val fs = hadoopPath.getFileSystem(hadoopConf) - fs.makeQualified(hadoopPath).toUri + CatalogUtils.makeQualifiedPath(path, hadoopConf) } private def requireDbExists(db: String): Unit = { @@ -254,12 +252,7 @@ class SessionCatalog( } private def makeQualifiedDBPath(locationUri: URI): URI = { - if (locationUri.isAbsolute) { - locationUri - } else { - val fullPath = new Path(conf.warehousePath, CatalogUtils.URIToString(locationUri)) - makeQualifiedPath(fullPath.toUri) - } + CatalogUtils.makeQualifiedNamespacePath(locationUri, conf.warehousePath, hadoopConf) } def dropDatabase(db: String, ignoreIfNotExists: Boolean, cascade: Boolean): Unit = { 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 cbfeaa4f5d651..9bf273ce16c5b 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 @@ -22,6 +22,7 @@ import scala.collection.mutable import org.apache.spark.sql.{SparkSession, Strategy} import org.apache.spark.sql.catalyst.analysis.{ResolvedDBObjectName, ResolvedNamespace, ResolvedPartitionSpec, ResolvedTable} +import org.apache.spark.sql.catalyst.catalog.CatalogUtils import org.apache.spark.sql.catalyst.expressions import org.apache.spark.sql.catalyst.expressions.{And, Attribute, DynamicPruning, EmptyRow, Expression, Literal, NamedExpression, PredicateHelper, SubqueryExpression} import org.apache.spark.sql.catalyst.planning.PhysicalOperation @@ -38,6 +39,7 @@ import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors import org.apache.spark.sql.execution.{FilterExec, LeafExecNode, LocalTableScanExec, ProjectExec, RowDataSourceScanExec, SparkPlan} import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, PushableColumn, PushableColumnBase} import org.apache.spark.sql.execution.streaming.continuous.{WriteToContinuousDataSource, WriteToContinuousDataSourceExec} +import org.apache.spark.sql.internal.StaticSQLConf.WAREHOUSE_PATH import org.apache.spark.sql.sources.{BaseRelation, TableScan} import org.apache.spark.sql.types.{BooleanType, StringType} import org.apache.spark.sql.util.CaseInsensitiveStringMap @@ -311,10 +313,13 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat AlterNamespaceSetPropertiesExec(catalog.asNamespaceCatalog, ns, properties) :: Nil case SetNamespaceLocation(ResolvedNamespace(catalog, ns), location) => + val warehousePath = session.sharedState.conf.get(WAREHOUSE_PATH) + val nsPath = CatalogUtils.makeQualifiedNamespacePath( + CatalogUtils.stringToURI(location), warehousePath, session.sharedState.hadoopConf) AlterNamespaceSetPropertiesExec( catalog.asNamespaceCatalog, ns, - Map(SupportsNamespaces.PROP_LOCATION -> location)) :: Nil + Map(SupportsNamespaces.PROP_LOCATION -> CatalogUtils.URIToString(nsPath))) :: Nil case CommentOnNamespace(ResolvedNamespace(catalog, ns), comment) => AlterNamespaceSetPropertiesExec( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index 6cbbf680bc9ca..c6c05f109c299 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -1294,13 +1294,25 @@ class DataSourceV2SQLSuite assert(descriptionDf.collect() === Seq( Row("Namespace Name", "ns2"), Row(SupportsNamespaces.PROP_COMMENT.capitalize, "test namespace"), - Row(SupportsNamespaces.PROP_LOCATION.capitalize, "/tmp/ns_test_2"), + Row(SupportsNamespaces.PROP_LOCATION.capitalize, "file:/tmp/ns_test_2"), Row(SupportsNamespaces.PROP_OWNER.capitalize, defaultUser), Row("Properties", "")) ) } } + test("SPARK-37444: ALTER NAMESPACE .. SET LOCATION using v2 catalog with empty location") { + val ns = "testcat.ns1.ns2" + withNamespace(ns) { + sql(s"CREATE NAMESPACE IF NOT EXISTS $ns COMMENT " + + "'test namespace' LOCATION '/tmp/ns_test_1'") + val e = intercept[IllegalArgumentException] { + sql(s"ALTER DATABASE $ns SET LOCATION ''") + } + assert(e.getMessage.contains("Can not create a Path from an empty string")) + } + } + private def testShowNamespaces( sqlText: String, expected: Seq[String]): Unit = {