[spark][flink] Fix vector-field column conversion in catalogs

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.types.VectorType;
+import org.apache.paimon.utils.StringUtils;
 
 import org.apache.flink.table.types.logical.BinaryType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -107,21 +108,21 @@ public static VectorType toVectorType(
         String dimKey = String.format("field.%s.vector-dim", fieldName);
         checkArgument(
                 options.containsKey(dimKey),
-                "When setting '"
-                        + CoreOptions.VECTOR_FIELD.key()
-                        + "', you must also set 'field.%s.vector-dim',"
-                        + " where %s is the name of the vector field.");
+                "When setting '%s', you must also set '%s'.",
+                CoreOptions.VECTOR_FIELD.key(),
+                dimKey);
         String vectorDim = options.get(dimKey);
         checkArgument(
-                !vectorDim.trim().isEmpty(),
-                "Expected an integer for vector-dim, but got empty value.");
+                !StringUtils.isNullOrWhitespaceOnly(vectorDim),
+                "Expected an integer for '%s', but got empty value.",
+                dimKey);
 
         try {
-            int dim = Integer.parseInt(vectorDim);
+            int dim = Integer.parseInt(vectorDim.trim());
             return DataTypes.VECTOR(dim, toDataType(elementType));
         } catch (NumberFormatException e) {
             throw new IllegalArgumentException(
-                    "Expected an integer for vector-dim, but got: " + vectorDim);
+                    String.format("Expected an integer for '%s', but got: %s.", dimKey, vectorDim));
         }
     }
 

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java

@@ -44,9 +44,9 @@
 import org.apache.paimon.types.BlobType;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.VectorType;
 import org.apache.paimon.utils.ExceptionUtils;
-import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.StringUtils;
 
 import org.apache.spark.sql.PaimonSparkSession$;
 import org.apache.spark.sql.SparkSession;
@@ -611,22 +611,8 @@ private Schema toInitialSchema(
                 type = toBlobType(field, false);
             } else if (blobFields.contains(name)) {
                 type = toBlobType(field, true);
-            } else if (vectorFields.contains(field.name())) {
-                Preconditions.checkArgument(
-                        field.dataType() instanceof ArrayType,
-                        "The type of blob field must be array");
-                ArrayType arrayType = (ArrayType) field.dataType();
-                String dimKey = String.format("field.%s.vector-dim", field.name());
-                Preconditions.checkArgument(
-                        properties.containsKey(dimKey),
-                        "When setting '"
-                                + CoreOptions.VECTOR_FIELD.key()
-                                + "', you must also set 'field.%s.vector-dim',"
-                                + " where %s is the name of the vector field.");
-                type =
-                        DataTypes.VECTOR(
-                                Integer.parseInt(properties.get(dimKey)),
-                                toPaimonType(arrayType.elementType()));
+            } else if (vectorFields.contains(name)) {
+                type = toVectorType(field, properties);
             } else {
                 type = toPaimonType(field.dataType()).copy(field.nullable());
             }
@@ -642,6 +628,36 @@ private Schema toInitialSchema(
         return schemaBuilder.build();
     }
 
+    private static DataType toVectorType(StructField field, Map<String, String> properties) {
+        checkArgument(
+                field.dataType() instanceof ArrayType,
+                "The type of vector field '%s' must be array, but is %s.",
+                field.name(),
+                field.dataType().catalogString());
+        ArrayType arrayType = (ArrayType) field.dataType();
+
+        String dimKey = String.format("field.%s.vector-dim", field.name());
+        checkArgument(
+                properties.containsKey(dimKey),
+                "When setting '%s', you must also set '%s'.",
+                CoreOptions.VECTOR_FIELD.key(),
+                dimKey);
+        String vectorDim = properties.get(dimKey);
+        checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(vectorDim),
+                "Expected an integer for '%s', but got empty value.",
+                dimKey);
+
+        int dim;
+        try {
+            dim = Integer.parseInt(vectorDim.trim());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    String.format("Expected an integer for '%s', but got: %s.", dimKey, vectorDim));
+        }
+        return new VectorType(field.nullable(), dim, toPaimonType(arrayType.elementType()));
+    }
+
     private static DataType toBlobType(StructField field, boolean allowNested) {
         org.apache.spark.sql.types.DataType sparkType = field.dataType();
         if (sparkType instanceof BinaryType) {

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala

@@ -21,7 +21,7 @@ package org.apache.paimon.spark.sql
 import org.apache.paimon.catalog.Identifier
 import org.apache.paimon.schema.Schema
 import org.apache.paimon.spark.PaimonSparkTestBase
-import org.apache.paimon.types.DataTypes
+import org.apache.paimon.types.{DataTypes, VectorType}
 
 import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionsException
@@ -910,6 +910,78 @@ abstract class DDLTestBase extends PaimonSparkTestBase {
     assert(error.contains("Unsupported partition transform"))
   }
 
+  test("Paimon DDL: create table with vector-field") {
+    withTable("T") {
+      sql("""
+            |CREATE TABLE T (id BIGINT, embed ARRAY<FLOAT> NOT NULL)
+            |TBLPROPERTIES (
+            |  'vector-field' = 'embed',
+            |  'field.embed.vector-dim' = '3',
+            |  'row-tracking.enabled' = 'true',
+            |  'data-evolution.enabled' = 'true')
+            |""".stripMargin)
+
+      val rowType = loadTable("T").rowType()
+      val embedType = rowType.getTypeAt(rowType.getFieldIndex("embed"))
+      assert(embedType.isInstanceOf[VectorType])
+      val vectorType = embedType.asInstanceOf[VectorType]
+      assert(vectorType.getLength == 3)
+      assert(vectorType.getElementType == DataTypes.FLOAT())
+      // NOT NULL declared on the Spark column must survive the conversion.
+      assert(!vectorType.isNullable)
+    }
+  }
+
+  test("Paimon DDL: create table with invalid vector-field") {
+    def createVectorTable(column: String, options: String): Unit = {
+      sql(s"""
+             |CREATE TABLE T (id BIGINT, $column)
+             |TBLPROPERTIES (
+             |  'vector-field' = 'embed',
+             |  'row-tracking.enabled' = 'true',
+             |  'data-evolution.enabled' = 'true'
+             |  $options)
+             |""".stripMargin)
+    }
+
+    // A vector column must be declared as an array.
+    withTable("T") {
+      val error = intercept[Exception] {
+        createVectorTable("embed FLOAT", ", 'field.embed.vector-dim' = '3'")
+      }
+      assert(
+        error.getMessage.contains("The type of vector field 'embed' must be array, but is float"))
+    }
+
+    // The dimension option is required, and the message must name the real option key.
+    withTable("T") {
+      val error = intercept[Exception] {
+        createVectorTable("embed ARRAY<FLOAT>", "")
+      }
+      assert(
+        error.getMessage.contains(
+          "When setting 'vector-field', you must also set 'field.embed.vector-dim'."))
+    }
+
+    // An empty or non-integer dimension must fail with a readable message.
+    withTable("T") {
+      val error = intercept[Exception] {
+        createVectorTable("embed ARRAY<FLOAT>", ", 'field.embed.vector-dim' = '  '")
+      }
+      assert(
+        error.getMessage.contains(
+          "Expected an integer for 'field.embed.vector-dim', but got empty value."))
+    }
+    withTable("T") {
+      val error = intercept[Exception] {
+        createVectorTable("embed ARRAY<FLOAT>", ", 'field.embed.vector-dim' = 'abc'")
+      }
+      assert(
+        error.getMessage.contains(
+          "Expected an integer for 'field.embed.vector-dim', but got: abc."))
+    }
+  }
+
   test("Fix partition column generate wrong partition spec") {
     Seq(true, false).foreach {
       legacyPartName =>