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 0ba8dc3954a0..04e9acb5ce21 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.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)); } } 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 4e18c2318a56..16fc1806d1e2 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.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 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 fa895d698d0b..2b5b21783759 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 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", "") + } + 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", ", '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", ", '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 =>