update

YannByron · YannByron · commit 4f8e790231f6 · 2025-12-26T18:28:35.000+08:00
diff --git a/.scalafmt.conf b/.scalafmt.conf
@@ -47,8 +47,8 @@ rewrite.rules = [AvoidInfix, Imports, RedundantBraces, SortModifiers]
 # Imports
 rewrite.imports.sort = scalastyle
 rewrite.imports.groups = [
-    ["org.apache.paimon\\..*"],
-    ["org.apache.paimon.shade\\..*"],
+    ["org.apache.fluss\\..*"],
+    ["org.apache.fluss.shade\\..*"],
     [".*"],
     ["javax\\..*"],
     ["java\\..*"],
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/Configuration.java b/fluss-common/src/main/java/org/apache/fluss/config/Configuration.java
@@ -18,6 +18,7 @@
 package org.apache.fluss.config;
 
 import org.apache.fluss.annotation.PublicStable;
+import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.utils.CollectionUtils;
 
 import org.slf4j.Logger;
@@ -643,7 +644,8 @@ <T> void setValueInternal(String key, T value) {
         }
     }
 
-    private Optional<Object> getRawValue(String key) {
+    @VisibleForTesting
+    public Optional<Object> getRawValue(String key) {
         if (key == null) {
             throw new NullPointerException("Key must not be null.");
         }
diff --git a/fluss-spark/fluss-spark-3.4/pom.xml b/fluss-spark/fluss-spark-3.4/pom.xml
@@ -26,25 +26,23 @@
         <version>0.9-SNAPSHOT</version>
     </parent>
 
-    <artifactId>fluss-spark-3.4</artifactId>
+    <artifactId>fluss-spark-3.4_${scala.binary.version}</artifactId>
     <name>Fluss : Engine Spark : 3.4</name>
 
     <properties>
         <spark.version>3.4.3</spark.version>
-        <scala.version>${scala212.version}</scala.version>
-        <scala.binary.version>2.12</scala.binary.version>
     </properties>
 
     <dependencies>
         <dependency>
             <groupId>org.apache.fluss</groupId>
-            <artifactId>fluss-spark-common</artifactId>
+            <artifactId>fluss-spark-common_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.fluss</groupId>
-            <artifactId>fluss-spark-ut</artifactId>
+            <artifactId>fluss-spark-ut_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
             <classifier>tests</classifier>
             <scope>test</scope>
diff --git a/fluss-spark/fluss-spark-3.5/pom.xml b/fluss-spark/fluss-spark-3.5/pom.xml
@@ -26,25 +26,23 @@
         <version>0.9-SNAPSHOT</version>
     </parent>
 
-    <artifactId>fluss-spark-3.5</artifactId>
+    <artifactId>fluss-spark-3.5_${scala.binary.version}</artifactId>
     <name>Fluss : Engine Spark : 3.5</name>
 
     <properties>
         <spark.version>3.5.7</spark.version>
-        <scala.version>${scala212.version}</scala.version>
-        <scala.binary.version>2.12</scala.binary.version>
     </properties>
 
     <dependencies>
         <dependency>
             <groupId>org.apache.fluss</groupId>
-            <artifactId>fluss-spark-common</artifactId>
+            <artifactId>fluss-spark-common_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.fluss</groupId>
-            <artifactId>fluss-spark-ut</artifactId>
+            <artifactId>fluss-spark-ut_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
             <classifier>tests</classifier>
             <scope>test</scope>
diff --git a/fluss-spark/fluss-spark-common/pom.xml b/fluss-spark/fluss-spark-common/pom.xml
@@ -26,15 +26,9 @@
         <version>0.9-SNAPSHOT</version>
     </parent>
 
-    <artifactId>fluss-spark-common</artifactId>
+    <artifactId>fluss-spark-common_${scala.binary.version}</artifactId>
     <name>Fluss : Engine Spark : Common</name>
 
-    <properties>
-        <spark.version>3.5.7</spark.version>
-        <scala.version>${scala212.version}</scala.version>
-        <scala.binary.version>2.12</scala.binary.version>
-    </properties>
-
     <dependencies>
         <dependency>
             <groupId>org.apache.spark</groupId>
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkCatalog.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkCatalog.scala
@@ -20,6 +20,7 @@ package org.apache.fluss.spark
 import org.apache.fluss.exception.{DatabaseNotExistException, TableAlreadyExistException, TableNotExistException}
 import org.apache.fluss.metadata.TablePath
 import org.apache.fluss.spark.catalog.{SupportsFlussNamespaces, WithFlussAdmin}
+
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog, TableChange}
 import org.apache.spark.sql.connector.expressions.Transform
@@ -31,7 +32,7 @@ import java.util.concurrent.ExecutionException
 
 import scala.collection.JavaConverters._
 
-class FlussCatalog extends TableCatalog with SupportsFlussNamespaces with WithFlussAdmin {
+class SparkCatalog extends TableCatalog with SupportsFlussNamespaces with WithFlussAdmin {
 
   private var catalogName: String = "fluss"
 
@@ -48,7 +49,7 @@ class FlussCatalog extends TableCatalog with SupportsFlussNamespaces with WithFl
 
   override def loadTable(ident: Identifier): Table = {
     try {
-      FlussTable(admin.getTableInfo(toTablePath(ident)).get())
+      SparkTable(admin.getTableInfo(toTablePath(ident)).get())
     } catch {
       case e: ExecutionException if e.getCause.isInstanceOf[TableNotExistException] =>
         throw new NoSuchTableException(ident)
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkConversions.scala
@@ -20,8 +20,9 @@ package org.apache.fluss.spark
 import org.apache.fluss.config.FlussConfigUtils
 import org.apache.fluss.metadata.{Schema, TableDescriptor}
 import org.apache.fluss.spark.SparkConnectorOptions._
-import org.apache.fluss.spark.types.{FlussDataTypeToSparkDataType, SparkToFlussTypeVisitor}
+import org.apache.fluss.spark.types.{FlussToSparkTypeVisitor, SparkToFlussTypeVisitor}
 import org.apache.fluss.types.RowType
+
 import org.apache.spark.sql.FlussIdentityTransform
 import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.connector.expressions.Transform
@@ -38,7 +39,7 @@ object SparkConversions {
     SparkToFlussTypeVisitor.visit(schema).asInstanceOf[RowType]
 
   def toSparkDataType(rowType: RowType): StructType =
-    FlussDataTypeToSparkDataType.visit(rowType).asInstanceOf[StructType]
+    FlussToSparkTypeVisitor.visit(rowType).asInstanceOf[StructType]
 
   def toFlussTable(
       sparkSchema: StructType,
@@ -76,7 +77,7 @@ object SparkConversions {
 
     val (tableProps, customProps) =
       caseInsensitiveProps.filterNot(SPARK_TABLE_OPTIONS.contains).partition {
-        case (key, _) => FlussConfigUtils.TABLE_OPTIONS.containsKey(key)
+        case (key, _) => key.startsWith(FlussConfigUtils.TABLE_PREFIX)
       }
 
     tableDescriptorBuilder
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkTable.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/SparkTable.scala
@@ -18,8 +18,8 @@
 package org.apache.fluss.spark
 
 import org.apache.fluss.metadata.TableInfo
-import org.apache.fluss.spark.catalog.{FlussTableInfo, SupportsFlussPartitionManagement}
+import org.apache.fluss.spark.catalog.{AbstractSparkTable, SupportsFlussPartitionManagement}
 
-case class FlussTable(table: TableInfo)
-  extends FlussTableInfo(table)
+case class SparkTable(table: TableInfo)
+  extends AbstractSparkTable(table)
   with SupportsFlussPartitionManagement {}
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/AbstractSparkTable.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/AbstractSparkTable.scala
@@ -19,14 +19,15 @@ package org.apache.fluss.spark.catalog
 
 import org.apache.fluss.metadata.TableInfo
 import org.apache.fluss.spark.SparkConversions
+
 import org.apache.spark.sql.connector.catalog.{Table, TableCapability}
 import org.apache.spark.sql.types.StructType
 
 import java.util
 
 import scala.collection.JavaConverters._
 
-abstract class FlussTableInfo(tableInfo: TableInfo) extends Table {
+abstract class AbstractSparkTable(tableInfo: TableInfo) extends Table {
 
   protected lazy val _schema: StructType =
     SparkConversions.toSparkDataType(tableInfo.getSchema.getRowType)
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/SupportsFlussNamespaces.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/SupportsFlussNamespaces.scala
@@ -20,6 +20,7 @@ package org.apache.fluss.spark.catalog
 import org.apache.fluss.exception.DatabaseNotExistException
 import org.apache.fluss.metadata.DatabaseDescriptor
 import org.apache.fluss.utils.Preconditions
+
 import org.apache.spark.sql.catalyst.analysis.NoSuchNamespaceException
 import org.apache.spark.sql.connector.catalog.{NamespaceChange, SupportsNamespaces}
 
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/SupportsFlussPartitionManagement.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/SupportsFlussPartitionManagement.scala
@@ -23,7 +23,7 @@ import org.apache.spark.sql.types.StructType
 
 import java.util
 
-trait SupportsFlussPartitionManagement extends FlussTableInfo with SupportsPartitionManagement {
+trait SupportsFlussPartitionManagement extends AbstractSparkTable with SupportsPartitionManagement {
 
   override def partitionSchema(): StructType = _partitionSchema
 
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/WithFlussAdmin.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/catalog/WithFlussAdmin.scala
@@ -21,6 +21,7 @@ import org.apache.fluss.client.{Connection, ConnectionFactory}
 import org.apache.fluss.client.admin.Admin
 import org.apache.fluss.config.{Configuration => FlussConfiguration}
 import org.apache.fluss.utils.{IOUtils, Preconditions}
+
 import org.apache.spark.sql.connector.catalog.CatalogPlugin
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/types/FlussToSparkTypeVisitor.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/types/FlussToSparkTypeVisitor.scala
@@ -18,11 +18,12 @@
 package org.apache.fluss.spark.types
 
 import org.apache.fluss.types._
+
 import org.apache.spark.sql.types.{DataType => SparkDataType, DataTypes => SparkDataTypes}
 
 import scala.collection.JavaConverters._
 
-object FlussDataTypeToSparkDataType extends DataTypeVisitor[SparkDataType] {
+object FlussToSparkTypeVisitor extends DataTypeVisitor[SparkDataType] {
 
   override def visit(charType: CharType): SparkDataType = {
     org.apache.spark.sql.types.CharType(charType.getLength)
@@ -89,14 +90,16 @@ object FlussDataTypeToSparkDataType extends DataTypeVisitor[SparkDataType] {
   }
 
   override def visit(arrayType: ArrayType): SparkDataType = {
-    SparkDataTypes.createArrayType(arrayType.getElementType.accept(this), arrayType.isNullable)
+    SparkDataTypes.createArrayType(
+      arrayType.getElementType.accept(this),
+      arrayType.getElementType.isNullable)
   }
 
   override def visit(mapType: MapType): SparkDataType = {
     SparkDataTypes.createMapType(
       mapType.getKeyType.accept(this),
       mapType.getValueType.accept(this),
-      mapType.isNullable
+      mapType.getValueType.isNullable
     )
   }
 
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/types/SparkToFlussTypeVisitor.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/types/SparkToFlussTypeVisitor.scala
@@ -18,6 +18,7 @@
 package org.apache.fluss.spark.types
 
 import org.apache.fluss.types.{ArrayType => FlussArrayType, DataField => FlussDataField, DataType => FlussDataType, MapType => FlussMapType, _}
+
 import org.apache.spark.sql.types.{ArrayType => SparkArrayType, DataType => SparkDataType, MapType => SparkMapType, StructType, UserDefinedType}
 
 import scala.collection.JavaConverters._
@@ -53,7 +54,7 @@ object SparkToFlussTypeVisitor {
   }
 
   private def visitArrayType(at: SparkArrayType): FlussArrayType = {
-    new FlussArrayType(at.containsNull, visit(at.elementType))
+    new FlussArrayType(visit(at.elementType).copy(at.containsNull))
   }
 
   private def visitPrimitiveType(t: SparkDataType): FlussDataType = {
@@ -76,18 +77,20 @@ object SparkToFlussTypeVisitor {
         new DecimalType(dt.precision, dt.scale)
       case _: org.apache.spark.sql.types.BinaryType =>
         new BinaryType(BinaryType.MAX_LENGTH)
-      case vct: org.apache.spark.sql.types.VarcharType =>
-        new CharType(vct.length)
+      case _: org.apache.spark.sql.types.VarcharType =>
+        new StringType()
       case ct: org.apache.spark.sql.types.CharType =>
         new CharType(ct.length)
       case _: org.apache.spark.sql.types.StringType =>
         new StringType()
       case _: org.apache.spark.sql.types.DateType =>
         new DateType()
       case _: org.apache.spark.sql.types.TimestampType =>
-        new LocalZonedTimestampType()
+        // spark only support 6 digits of precision
+        new LocalZonedTimestampType(6)
       case _: org.apache.spark.sql.types.TimestampNTZType =>
-        new TimestampType()
+        // spark only support 6 digits of precision
+        new TimestampType(6)
       case _ =>
         throw new UnsupportedOperationException(s"Data type(${t.catalogString}) is not supported");
     }
diff --git a/fluss-spark/fluss-spark-ut/pom.xml b/fluss-spark/fluss-spark-ut/pom.xml
@@ -27,13 +27,13 @@
         <version>0.9-SNAPSHOT</version>
     </parent>
 
-    <artifactId>fluss-spark-ut</artifactId>
+    <artifactId>fluss-spark-ut_${scala.binary.version}</artifactId>
     <name>Fluss : Engine Spark : UT</name>
 
     <dependencies>
         <dependency>
             <groupId>org.apache.fluss</groupId>
-            <artifactId>fluss-spark-common</artifactId>
+            <artifactId>fluss-spark-common_${scala.binary.version}</artifactId>
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussCatalogTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussCatalogTest.scala
@@ -19,6 +19,7 @@ package org.apache.fluss.spark
 
 import org.apache.fluss.metadata.{DatabaseDescriptor, Schema, TableDescriptor, TablePath}
 import org.apache.fluss.types.{DataTypes, RowType}
+
 import org.apache.spark.sql.Row
 import org.assertj.core.api.Assertions.{assertThat, assertThatList}
 
@@ -72,6 +73,9 @@ class FlussCatalogTest extends FlussSparkTestBase {
         .build())
     assertThat(testPartitionedTable.getPartitionKeys.get(0)).isEqualTo("pt")
     assertThat(testPartitionedTable.getCustomProperties.containsKey("key")).isEqualTo(true)
+    assertThat(
+      testPartitionedTable.getCustomProperties.getRawValue("key").get().asInstanceOf[String])
+      .isEqualTo("value")
 
     sql("DROP TABLE test_tbl")
     sql("DROP TABLE test_pt_tbl")
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala
@@ -21,6 +21,7 @@ import org.apache.fluss.client.{Connection, ConnectionFactory}
 import org.apache.fluss.client.admin.Admin
 import org.apache.fluss.config.{ConfigOptions, Configuration}
 import org.apache.fluss.server.testutils.FlussClusterExtension
+
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.QueryTest
 import org.apache.spark.sql.test.SharedSparkSession
@@ -41,7 +42,7 @@ class FlussSparkTestBase extends QueryTest with SharedSparkSession {
 
   override protected def sparkConf: SparkConf = {
     super.sparkConf
-      .set("spark.sql.catalog.fluss_catalog", classOf[FlussCatalog].getName)
+      .set("spark.sql.catalog.fluss_catalog", classOf[SparkCatalog].getName)
       .set("spark.sql.catalog.fluss_catalog.bootstrap.servers", bootstrapServers)
       .set("spark.sql.defaultCatalog", "fluss_catalog")
   }
diff --git a/fluss-spark/pom.xml b/fluss-spark/pom.xml
@@ -31,14 +31,11 @@
     <name>Fluss : Engine Spark :</name>
     <packaging>pom</packaging>
 
-    <properties>
-        <scala212.version>2.12.18</scala212.version>
-        <scala213.version>2.13.16</scala213.version>
-    </properties>
-
     <modules>
         <module>fluss-spark-common</module>
         <module>fluss-spark-ut</module>
+        <module>fluss-spark-3.5</module>
+        <module>fluss-spark-3.4</module>
     </modules>
 
     <dependencies>
@@ -303,7 +300,7 @@
                         <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
                         <junitxml>.</junitxml>
                         <argLine>-ea -Xmx4g -Xss4m -XX:MaxMetaspaceSize=2g -XX:ReservedCodeCacheSize=128m ${extraJavaTestArgs} -Dio.netty.tryReflectionSetAccessible=true</argLine>
-                        <filereports>PaimonTestSuite.txt</filereports>
+                        <filereports>FlussTestSuite.txt</filereports>
                         <forkMode>once</forkMode>
                         <noScalaTestIgnore>true</noScalaTestIgnore>
                     </configuration>
diff --git a/pom.xml b/pom.xml
@@ -93,8 +93,10 @@
         <iceberg.version>1.10.0</iceberg.version>
 
         <!-- spark & scala -->
+        <scala212.version>2.12.18</scala212.version>
+        <scala213.version>2.13.16</scala213.version>
         <scala.binary.version>2.12</scala.binary.version>
-        <scala.version>2.12.18</scala.version>
+        <scala.version>${scala212.version}</scala.version>
         <spark.version>3.5.7</spark.version>
 
         <fluss.hadoop.version>3.4.0</fluss.hadoop.version>
@@ -464,20 +466,6 @@
             </build>
         </profile>
 
-        <profile>
-            <id>spark3</id>
-            <modules>
-                <module>fluss-spark/fluss-spark-3.5</module>
-                <module>fluss-spark/fluss-spark-3.4</module>
-            </modules>
-            <activation>
-                <activeByDefault>true</activeByDefault>
-                <property>
-                    <name>spark3</name>
-                </property>
-            </activation>
-        </profile>
-
         <profile>
             <id>test-coverage</id>
             <properties>

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`package org.apache.fluss.config;`
`19`	`19`
`20`	`20`	`import org.apache.fluss.annotation.PublicStable;`
	`21`	`+import org.apache.fluss.annotation.VisibleForTesting;`
`21`	`22`	`import org.apache.fluss.utils.CollectionUtils;`
`22`	`23`
`23`	`24`	`import org.slf4j.Logger;`
`@@ -643,7 +644,8 @@ <T> void setValueInternal(String key, T value) {`
`643`	`644`	`}`
`644`	`645`	`}`
`645`	`646`
`646`		`- private Optional<Object> getRawValue(String key) {`
	`647`	`+ @VisibleForTesting`
	`648`	`+ public Optional<Object> getRawValue(String key) {`
`647`	`649`	`if (key == null) {`
`648`	`650`	`throw new NullPointerException("Key must not be null.");`
`649`	`651`	`}`