[spark] Fix spark desc command output with partition info

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/AbstractSparkTable.scala

@@ -17,11 +17,12 @@
 
 package org.apache.fluss.spark.catalog
 
-import org.apache.fluss.config.{Configuration => FlussConfiguration}
-import org.apache.fluss.metadata.{TableInfo, TablePath}
+import org.apache.fluss.metadata.TableInfo
 import org.apache.fluss.spark.SparkConversions
 
+import org.apache.spark.sql.CatalogV2UtilShim
 import org.apache.spark.sql.connector.catalog.{Table, TableCapability}
+import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.types.StructType
 
 import java.util
@@ -34,7 +35,7 @@ abstract class AbstractSparkTable(tableInfo: TableInfo) extends Table {
     SparkConversions.toSparkDataType(tableInfo.getSchema.getRowType)
 
   protected lazy val _partitionSchema = new StructType(
-    _schema.fields.filter(tableInfo.getPartitionKeys.contains))
+    _schema.fields.filter(e => tableInfo.getPartitionKeys.contains(e.name)))
 
   override def name(): String = tableInfo.toString
 
@@ -43,4 +44,8 @@ abstract class AbstractSparkTable(tableInfo: TableInfo) extends Table {
   override def capabilities(): util.Set[TableCapability] = {
     Set(TableCapability.BATCH_WRITE).asJava
   }
+
+  override def partitioning(): Array[Transform] = {
+    CatalogV2UtilShim.toSparkTransforms(_partitionSchema.fields.map(_.name))
+  }
 }

fluss-spark/fluss-spark-common/src/main/scala/org/apache/spark/sql/CatalogV2UtilShim.scala

@@ -0,0 +1,28 @@
+/*
+ * 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.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.connector.expressions.Transform
+
+/** Shim of org.apache.spark.sql.connector.catalog.CatalogV2Util. */
+object CatalogV2UtilShim {
+  def toSparkTransforms(colNames: Seq[String]): Array[Transform] = {
+    colNames.asTransforms
+  }
+}

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussCatalogTest.scala

@@ -21,6 +21,7 @@ import org.apache.fluss.metadata.{DatabaseDescriptor, Schema, TableDescriptor, T
 import org.apache.fluss.types.{DataTypes, RowType}
 
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.Identifier
 import org.assertj.core.api.Assertions.{assertThat, assertThatList}
 
 import scala.collection.JavaConverters._
@@ -144,12 +145,18 @@ class FlussCatalogTest extends FlussSparkTestBase {
   }
 
   test("Catalog: check namespace and table created by admin") {
+    val dbName = "db_by_fluss_admin"
+    val tblName = "tbl_by_fluss_admin"
+    val catalog = spark.sessionState.catalogManager.currentCatalog.asInstanceOf[SparkCatalog]
+
+    // check namespace
     val dbDesc = DatabaseDescriptor.builder().comment("created by admin").build()
-    admin.createDatabase("db_by_admin", dbDesc, true).get()
-    checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Row("db_by_admin") :: Nil)
+    admin.createDatabase(dbName, dbDesc, true).get()
+    assert(catalog.namespaceExists(Array(dbName)))
+    checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Row(dbName) :: Nil)
 
-    sql("USE db_by_admin")
-    val tablePath = TablePath.of("db_by_admin", "tbl_by_admin")
+    // check table
+    val tablePath = TablePath.of(dbName, tblName)
     val rt = RowType
       .builder()
       .field("id", DataTypes.INT())
@@ -162,15 +169,25 @@ class FlussCatalogTest extends FlussSparkTestBase {
       .partitionedBy("pt")
       .build()
     admin.createTable(tablePath, tableDesc, false).get()
-    checkAnswer(sql("SHOW TABLES"), Row("db_by_admin", "tbl_by_admin", false) :: Nil)
+    assert(
+      catalog.tableExists(Identifier.of(Array(tablePath.getDatabaseName), tablePath.getTableName)))
+    val expectDescTable = Seq(
+      Row("id", "int", null),
+      Row("name", "string", null),
+      Row("pt", "string", null),
+      Row("# Partition Information", "", ""),
+      Row("# col_name", "data_type", "comment"),
+      Row("pt", "string", null)
+    )
     checkAnswer(
-      sql("DESC tbl_by_admin"),
-      Row("id", "int", null) :: Row("name", "string", null) :: Row("pt", "string", null) :: Nil)
+      sql(s"DESC $dbName.$tblName"),
+      expectDescTable
+    )
 
     admin.dropTable(tablePath, true).get()
-    checkAnswer(sql("SHOW TABLES"), Nil)
+    checkAnswer(sql(s"SHOW TABLES IN $dbName"), Nil)
 
-    admin.dropDatabase("db_by_admin", true, true).get()
+    admin.dropDatabase(dbName, true, true).get()
     checkAnswer(sql("SHOW DATABASES"), Row(DEFAULT_DATABASE) :: Nil)
   }
 }

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala

@@ -30,8 +30,6 @@ import org.apache.spark.SparkConf
 import org.apache.spark.sql.QueryTest
 import org.apache.spark.sql.test.SharedSparkSession
 import org.junit.jupiter.api.extension.RegisterExtension
-import org.scalactic.source.Position
-import org.scalatest.Tag
 
 import java.time.Duration
 
@@ -41,16 +39,17 @@ class FlussSparkTestBase extends QueryTest with SharedSparkSession {
 
   import FlussSparkTestBase._
 
-  protected val DEFAULT_DATABASE = "fluss";
+  protected val DEFAULT_CATALOG = "fluss_catalog"
+  protected val DEFAULT_DATABASE = "fluss"
 
   protected var conn: Connection = _
   protected var admin: Admin = _
 
   override protected def sparkConf: SparkConf = {
     super.sparkConf
-      .set("spark.sql.catalog.fluss_catalog", classOf[SparkCatalog].getName)
-      .set("spark.sql.catalog.fluss_catalog.bootstrap.servers", bootstrapServers)
-      .set("spark.sql.defaultCatalog", "fluss_catalog")
+      .set(s"spark.sql.catalog.$DEFAULT_CATALOG", classOf[SparkCatalog].getName)
+      .set(s"spark.sql.catalog.$DEFAULT_CATALOG.bootstrap.servers", bootstrapServers)
+      .set("spark.sql.defaultCatalog", DEFAULT_CATALOG)
   }
 
   override protected def beforeAll(): Unit = {