This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 72d9b2cf52 [spark][flink] Fix vector-field column conversion in
catalogs (#9014)
72d9b2cf52 is described below
commit 72d9b2cf52f8d434fea4f3e764f606eb7d7c814e
Author: jackylee <[email protected]>
AuthorDate: Fri Aug 7 13:20:46 2026 +0800
[spark][flink] Fix vector-field column conversion in catalogs (#9014)
---
.../apache/paimon/flink/LogicalTypeConversion.java | 17 ++---
.../java/org/apache/paimon/spark/SparkCatalog.java | 52 +++++++++------
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 74 +++++++++++++++++++++-
3 files changed, 116 insertions(+), 27 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java
index 0ba8dc3954..04e9acb5ce 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java
@@ -24,6 +24,7 @@ import org.apache.paimon.types.DataType;
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 class LogicalTypeConversion {
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));
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 4e18c2318a..16fc1806d1 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -44,9 +44,9 @@ import org.apache.paimon.table.object.ObjectTable;
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 @@ public class SparkCatalog extends SparkBaseCatalog
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 @@ public class SparkCatalog extends SparkBaseCatalog
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) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
index fa895d698d..2b5b217837 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
+++
b/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 =>
