fix catalog

apache · Dec 2, 2021 · 729cc22 · 729cc22
1 parent d9fa8a6
commit 729cc22
Showing 1 changed file with 3 additions and 6 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -323,11 +323,8 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
           provider match {
             case supportsExtract: SupportsCatalogOptions =>
               val ident = supportsExtract.extractIdentifier(dsOptions)
-              val catalog = if (ident.namespace().isEmpty) {
-                Array(dsOptions.get("catalog"))
-              } else {
-                ident.namespace()
-              }
+              val catalog = CatalogV2Util.getTableProviderCatalog(
+                supportsExtract, catalogManager, dsOptions)
 
               val location = Option(dsOptions.get("path")).map(TableCatalog.PROP_LOCATION -> _)
               val tableSpec = TableSpec(
@@ -342,7 +339,7 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
               runCommand(df.sparkSession) {
                 CreateTableAsSelect(
                   UnresolvedDBObjectName(
-                    catalog.toSeq :+ ident.name,
+                    catalog.name :: ident.name :: Nil,
                     isNamespace = false
                   ),
                   partitioningAsV2,