This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e21e4f7ed2 [Feature][Connector-V2][Milvus] Support scalar field write
null (#11216)
e21e4f7ed2 is described below
commit e21e4f7ed2b3ae60d7e6d33fb16662f0375a9bfe
Author: loupipalien <[email protected]>
AuthorDate: Sun Jul 12 20:16:28 2026 +0800
[Feature][Connector-V2][Milvus] Support scalar field write null (#11216)
---
docs/en/connectors/sink/Milvus.md | 1 +
docs/zh/connectors/sink/Milvus.md | 1 +
.../seatunnel/milvus/catalog/MilvusCatalog.java | 7 +-
.../seatunnel/milvus/config/MilvusSinkOptions.java | 6 +
.../seatunnel/milvus/sink/MilvusSinkFactory.java | 1 +
.../milvus/utils/sink/MilvusSinkConverter.java | 77 +++++-
.../milvus/utils/source/MilvusSourceConverter.java | 1 +
.../milvus/catalog/MilvusCatalogTest.java | 16 ++
.../milvus/sink/MilvusSinkFactoryTest.java | 1 +
.../milvus/utils/sink/MilvusSinkConverterTest.java | 261 +++++++++++++++++++--
.../utils/source/MilvusSourceConverterTest.java | 44 ++++
.../e2e/connector/v2/milvus/MilvusIT.java | 89 ++++++-
.../test/resources/fake-to-milvus-nullable.conf | 73 ++++++
.../seatunnel/format/json/JsonToRowConverters.java | 29 +++
.../format/json/JsonRowDataSerDeSchemaTest.java | 35 +++
15 files changed, 617 insertions(+), 25 deletions(-)
diff --git a/docs/en/connectors/sink/Milvus.md
b/docs/en/connectors/sink/Milvus.md
index 832bb9dae1..59abb90a3c 100644
--- a/docs/en/connectors/sink/Milvus.md
+++ b/docs/en/connectors/sink/Milvus.md
@@ -59,6 +59,7 @@ Common use cases:
| enable_auto_id | boolean | No | false
| Whether Milvus should generate primary key values automatically. If
this is `true`, do not send values for the primary key field.
|
| enable_upsert | boolean | No | true
| Whether to write by upsert instead of insert. Upsert needs a
primary key in the collection schema.
|
| enable_dynamic_field | boolean | No | true
| Whether to enable Milvus dynamic fields when SeaTunnel creates a
collection.
|
+| enable_nullable_field | boolean | No | false
| Enable nullable fields. Requires Milvus 2.5 or later. When enabled,
SeaTunnel creates the field as a Milvus nullable field and allows writing
`null` values when the SeaTunnel column is nullable and the field is used as a
Milvus scalar field. |
| batch_size | int | No | 1000
| Number of records buffered before one write. In streaming jobs,
data can also be flushed when a checkpoint is triggered.
|
| rate_limit | int | No | 100000
| Milvus collection write-rate limit in MB/s. When greater than 0,
the connector sets `collection.insertRate.max.mb` and
`collection.upsertRate.max.mb` while the writer is running.
|
| partition_key | String | No | -
| Field name used as the Milvus partition key when SeaTunnel creates
a collection.
|
diff --git a/docs/zh/connectors/sink/Milvus.md
b/docs/zh/connectors/sink/Milvus.md
index bc4fdb03f6..59f192054b 100644
--- a/docs/zh/connectors/sink/Milvus.md
+++ b/docs/zh/connectors/sink/Milvus.md
@@ -59,6 +59,7 @@ SeaTunnel 表结构创建目标集合，并支持向量字段、动态字段、
| enable_auto_id | boolean | 否 | false
| 是否让 Milvus 自动生成主键。设置为 `true` 时，不要再向主键字段写值。
|
| enable_upsert | boolean | 否 | true
| 是否使用更新插入模式写入，而不是普通插入模式。更新插入需要集合中有主键。
|
| enable_dynamic_field | boolean | 否 | true
| SeaTunnel 创建集合时，是否启用 Milvus 动态字段。
|
+| enable_nullable_field | boolean | 否 | false
| 是否启用可空字段。需要 Milvus 2.5 或更高版本。开启后，当 SeaTunnel 列定义为 nullable 且作为 Milvus
标量字段，SeaTunnel 会将该字段创建为 Milvus nullable 字段并允许写入 `null` 值。|
| batch_size | int | 否 | 1000
| 写入前缓存的记录数。流任务中，检查点触发时也会刷新写入。
|
| rate_limit | int | 否 | 100000
| Milvus collection 写入限速，单位 MB/s。大于 0 时，写入器运行期间会设置
`collection.insertRate.max.mb` 和 `collection.upsertRate.max.mb`。 |
| partition_key | String | 否 | -
| SeaTunnel 创建集合时使用的 Milvus 分区键字段名。
|
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalog.java
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalog.java
index d4860e9c0d..9ab904fa87 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalog.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalog.java
@@ -160,6 +160,10 @@ public class MilvusCatalog implements Catalog {
@Override
public boolean tableExists(TablePath tablePath) throws CatalogException {
+ if (!databaseExists(tablePath.getDatabaseName())) {
+ return false;
+ }
+
R<Boolean> response =
this.client.hasCollection(
HasCollectionParam.newBuilder()
@@ -302,7 +306,8 @@ public class MilvusCatalog implements Catalog {
column,
tableSchema.getPrimaryKey(),
partitionKeyField,
- config.get(MilvusSinkOptions.ENABLE_AUTO_ID));
+ config.get(MilvusSinkOptions.ENABLE_AUTO_ID),
+
config.get(MilvusSinkOptions.ENABLE_NULLABLE_FIELD));
fieldTypes.add(fieldType);
}
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSinkOptions.java
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSinkOptions.java
index 4113be01b8..75ac1d922d 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSinkOptions.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/config/MilvusSinkOptions.java
@@ -77,6 +77,12 @@ public class MilvusSinkOptions extends MilvusBaseOptions {
.defaultValue(true)
.withDescription("Enable dynamic field");
+ public static final Option<Boolean> ENABLE_NULLABLE_FIELD =
+ Options.key("enable_nullable_field")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable nullable field");
+
public static final Option<Integer> BATCH_SIZE =
Options.key("batch_size")
.intType()
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactory.java
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactory.java
index 8cfab9d139..f1ed7fe122 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactory.java
@@ -50,6 +50,7 @@ public class MilvusSinkFactory implements TableSinkFactory {
MilvusSinkOptions.PARTITION_KEY,
MilvusSinkOptions.ENABLE_UPSERT,
MilvusSinkOptions.ENABLE_DYNAMIC_FIELD,
+ MilvusSinkOptions.ENABLE_NULLABLE_FIELD,
MilvusSinkOptions.ENABLE_AUTO_ID,
MilvusSinkOptions.BATCH_SIZE,
MilvusSinkOptions.RATE_LIMIT,
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverter.java
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverter.java
index ae7fb816f9..e9f5d26cfc 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverter.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverter.java
@@ -32,10 +32,12 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.api.table.type.SqlType;
import org.apache.seatunnel.common.utils.JsonUtils;
import org.apache.seatunnel.common.utils.VectorUtils;
+import org.apache.seatunnel.connectors.seatunnel.milvus.catalog.MilvusOptions;
import
org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectionErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectorException;
import com.google.gson.Gson;
+import com.google.gson.JsonNull;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import io.milvus.grpc.DataType;
@@ -50,6 +52,8 @@ import java.util.stream.Collectors;
import static
org.apache.seatunnel.api.table.catalog.PrimaryKey.isPrimaryKeyField;
import static
org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSinkOptions.ENABLE_AUTO_ID;
import static
org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSinkOptions.ENABLE_DYNAMIC_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSinkOptions.ENABLE_NULLABLE_FIELD;
+import static
org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSinkOptions.PARTITION_KEY;
public class MilvusSinkConverter {
private static final Gson gson = new Gson();
@@ -126,7 +130,11 @@ public class MilvusSinkConverter {
}
public static FieldType convertToFieldType(
- Column column, PrimaryKey primaryKey, String partitionKeyField,
Boolean autoId) {
+ Column column,
+ PrimaryKey primaryKey,
+ String partitionKeyField,
+ Boolean autoId,
+ Boolean enableNullableField) {
SeaTunnelDataType<?> seaTunnelDataType = column.getDataType();
DataType milvusDataType;
if (column.getSinkType() != null) {
@@ -135,7 +143,15 @@ public class MilvusSinkConverter {
milvusDataType =
convertSqlTypeToDataType(seaTunnelDataType.getSqlType());
}
FieldType.Builder build =
-
FieldType.newBuilder().withName(column.getName()).withDataType(milvusDataType);
+ FieldType.newBuilder()
+ .withName(column.getName())
+ .withDataType(milvusDataType)
+ .withNullable(
+ allowNullableField(
+ column,
+ primaryKey,
+ partitionKeyField,
+ enableNullableField));
if (StringUtils.isNotEmpty(column.getComment())) {
build.withDescription(column.getComment());
}
@@ -210,6 +226,36 @@ public class MilvusSinkConverter {
return build.build();
}
+ private static boolean allowNullableField(
+ Column column,
+ PrimaryKey primaryKey,
+ String partitionKeyField,
+ Boolean enableNullableField) {
+ return Boolean.TRUE.equals(enableNullableField)
+ && column.isNullable()
+ && supportNullableField(column, primaryKey, partitionKeyField);
+ }
+
+ private static boolean supportNullableField(
+ Column column, PrimaryKey primaryKey, String partitionKeyField) {
+ if (isPrimaryKeyField(primaryKey, column.getName())) {
+ return false;
+ }
+ if (StringUtils.equals(column.getName(), partitionKeyField)) {
+ return false;
+ }
+ switch (column.getDataType().getSqlType()) {
+ case FLOAT_VECTOR:
+ case BINARY_VECTOR:
+ case FLOAT16_VECTOR:
+ case BFLOAT16_VECTOR:
+ case SPARSE_FLOAT_VECTOR:
+ return false;
+ default:
+ return true;
+ }
+ }
+
public static DataType convertSqlTypeToDataType(SqlType sqlType) {
switch (sqlType) {
case BOOLEAN:
@@ -260,6 +306,8 @@ public class MilvusSinkConverter {
SeaTunnelRowType seaTunnelRowType = catalogTable.getSeaTunnelRowType();
PrimaryKey primaryKey = catalogTable.getTableSchema().getPrimaryKey();
Boolean autoId = config.get(ENABLE_AUTO_ID);
+ Boolean enableNullableField = config.get(ENABLE_NULLABLE_FIELD);
+ String partitionKeyField = getPartitionKeyField(catalogTable, config);
JsonObject data = new JsonObject();
Gson gson = new Gson();
@@ -273,9 +321,20 @@ public class MilvusSinkConverter {
SeaTunnelDataType<?> fieldType = seaTunnelRowType.getFieldType(i);
Object value = element.getField(i);
- if (null == value) {
- throw new MilvusConnectorException(
- MilvusConnectionErrorCode.FIELD_IS_NULL, fieldName);
+ if (value == null) {
+ Column column =
catalogTable.getTableSchema().getColumn(fieldName);
+ if (!allowNullableField(
+ column, primaryKey, partitionKeyField,
enableNullableField)) {
+ throw new MilvusConnectorException(
+ MilvusConnectionErrorCode.FIELD_IS_NULL,
fieldName);
+ }
+ if (dynamicField != null
+ && dynamicField.equals(fieldName)
+ && config.get(ENABLE_DYNAMIC_FIELD)) {
+ continue;
+ }
+ data.add(fieldName, JsonNull.INSTANCE);
+ continue;
}
// if the field is dynamic field, then parse the dynamic field
if (dynamicField != null
@@ -295,4 +354,12 @@ public class MilvusSinkConverter {
}
return data;
}
+
+ private String getPartitionKeyField(CatalogTable catalogTable,
ReadonlyConfig config) {
+ String partitionKeyField =
catalogTable.getOptions().get(MilvusOptions.PARTITION_KEY_FIELD);
+ if (StringUtils.isNotEmpty(config.get(PARTITION_KEY))) {
+ partitionKeyField = config.get(PARTITION_KEY);
+ }
+ return partitionKeyField;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverter.java
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverter.java
index 6308e3a343..dd4ed46686 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverter.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverter.java
@@ -262,6 +262,7 @@ public class MilvusSourceConverter {
PhysicalColumn.PhysicalColumnBuilder builder =
PhysicalColumn.builder();
builder.name(fieldSchema.getName());
builder.sourceType(dataType.name());
+ builder.nullable(fieldSchema.getNullable());
builder.comment(fieldSchema.getDescription());
switch (dataType) {
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalogTest.java
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalogTest.java
index 1116cafe56..ae0d3c53bc 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalogTest.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/catalog/MilvusCatalogTest.java
@@ -26,6 +26,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import io.milvus.client.MilvusServiceClient;
+import io.milvus.grpc.ListDatabasesResponse;
import io.milvus.grpc.ShowPartitionsResponse;
import io.milvus.param.R;
import io.milvus.param.RpcStatus;
@@ -47,6 +48,21 @@ import static org.mockito.Mockito.when;
class MilvusCatalogTest {
+ @Test
+ void tableExistsReturnsFalseWhenDatabaseDoesNotExist() throws Exception {
+ MilvusServiceClient client = mock(MilvusServiceClient.class);
+ @SuppressWarnings("unchecked")
+ R<ListDatabasesResponse> listDatabasesR = mock(R.class);
+ when(listDatabasesR.getData())
+
.thenReturn(ListDatabasesResponse.newBuilder().addDbNames("default").build());
+ when(client.listDatabases()).thenReturn(listDatabasesR);
+
+ MilvusCatalog catalog = createCatalogWithClient(client);
+ Assertions.assertFalse(catalog.tableExists(TablePath.of("missing_db",
null, "coll")));
+
+ verify(client, never()).hasCollection(any());
+ }
+
@Test
void createPartitionInternalSkipsEmptyString() throws Exception {
MilvusCatalog catalog =
createCatalogWithClient(mockClientWithDefaultPartitions());
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactoryTest.java
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactoryTest.java
index 1b5542366e..ef7cd9f4d0 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSinkFactoryTest.java
@@ -105,6 +105,7 @@ public class MilvusSinkFactoryTest {
assertTrue(declaredOptions.contains(MilvusSinkOptions.LOAD_COLLECTION));
assertTrue(declaredOptions.contains(MilvusSinkOptions.CREATE_INDEX));
assertTrue(declaredOptions.contains(MilvusSinkOptions.RATE_LIMIT));
+
assertTrue(declaredOptions.contains(MilvusSinkOptions.ENABLE_NULLABLE_FIELD));
}
private SeaTunnelSink<?, ?, ?, ?> createSeaTunnelSink(Map<String, Object>
options) {
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverterTest.java
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverterTest.java
index 3e94609255..95effd3618 100644
---
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverterTest.java
+++
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/sink/MilvusSinkConverterTest.java
@@ -17,52 +17,277 @@
package org.apache.seatunnel.connectors.seatunnel.milvus.utils.sink;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.BasicType;
-import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.CommonOptions;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.VectorType;
+import
org.apache.seatunnel.connectors.seatunnel.milvus.config.MilvusSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectorException;
import org.junit.jupiter.api.Test;
+import com.google.gson.JsonNull;
import io.milvus.grpc.DataType;
import io.milvus.param.collection.FieldType;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class MilvusSinkConverterTest {
@Test
void returnsReconvertedTypeWhenSinkTypeNotNull() {
- Column column = mock(Column.class);
- when(column.getName()).thenReturn("col1");
- when(column.getDataType()).thenReturn((SeaTunnelDataType)
BasicType.SHORT_TYPE);
- when(column.getSinkType()).thenReturn("Int64");
+ Column column = columnWithSinkType("col1", "Int64");
- FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null);
+ FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null, false);
assertEquals(DataType.Int64, result.getDataType());
}
@Test
void returnsReconvertedTypeWhenSinkTypeIsNull() {
- Column column = mock(Column.class);
- when(column.getSinkType()).thenReturn(null);
- when(column.getDataType()).thenReturn((SeaTunnelDataType)
BasicType.SHORT_TYPE);
- when(column.getName()).thenReturn("col1");
- FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null);
+ Column column = PhysicalColumn.of("col1", BasicType.SHORT_TYPE, 0L,
true, null, "");
+ FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null, false);
assertEquals(DataType.Int16, result.getDataType());
}
@Test
void returnsReconvertedTypeWhenTypesNotNull() {
- Column column = mock(Column.class);
- when(column.getSinkType()).thenReturn("Int64");
- when(column.getDataType()).thenReturn((SeaTunnelDataType)
BasicType.SHORT_TYPE);
- when(column.getName()).thenReturn("col1");
- FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null);
+ Column column = columnWithSinkType("col1", "Int64");
+ FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null, false);
assertEquals(DataType.Int64, result.getDataType());
}
+
+ @Test
+ void convertsNullableColumnToNullableMilvusField() {
+ Column column =
+ PhysicalColumn.of("nullable_col", BasicType.STRING_TYPE, 0L,
true, null, "");
+
+ FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null, true);
+
+ assertTrue(result.isNullable());
+ }
+
+ @Test
+ void keepsNotNullableColumnNotNullableWhenNullableFieldEnabled() {
+ Column column =
+ PhysicalColumn.of("nullable_col", BasicType.STRING_TYPE, 0L,
false, null, "");
+
+ FieldType result = MilvusSinkConverter.convertToFieldType(column,
null, null, null, true);
+
+ assertFalse(result.isNullable());
+ }
+
+ @Test
+ void keepsPrimaryKeyAndVectorFieldNotNullableWhenNullableFieldEnabled() {
+ Column primaryKeyColumn = PhysicalColumn.of("id", BasicType.LONG_TYPE,
0L, false, null, "");
+ FieldType primaryKey =
+ MilvusSinkConverter.convertToFieldType(
+ primaryKeyColumn,
+ PrimaryKey.of("id", Collections.singletonList("id")),
+ null,
+ false,
+ true);
+
+ Column vectorColumn =
+ PhysicalColumn.of("vector", VectorType.VECTOR_FLOAT_TYPE, 0L,
4, false, null, "");
+ FieldType vector =
+ MilvusSinkConverter.convertToFieldType(vectorColumn, null,
null, null, true);
+
+ assertEquals(false, primaryKey.isNullable());
+ assertEquals(false, vector.isNullable());
+ }
+
+ @Test
+ void keepsPartitionKeyNotNullableWhenNullableFieldEnabled() {
+ Column partitionKeyColumn =
+ PhysicalColumn.of("partition_col", BasicType.STRING_TYPE, 0L,
true, null, "");
+
+ FieldType partitionKey =
+ MilvusSinkConverter.convertToFieldType(
+ partitionKeyColumn, null, "partition_col", null, true);
+
+ assertFalse(partitionKey.isNullable());
+ assertTrue(partitionKey.isPartitionKey());
+ }
+
+ @Test
+ void throwsWhenFieldValueIsNullByDefault() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {null});
+
+ MilvusConnectorException exception =
+ assertThrows(
+ MilvusConnectorException.class,
+ () ->
+ new MilvusSinkConverter()
+ .buildMilvusData(
+ catalogTable(),
+
ReadonlyConfig.fromMap(Collections.emptyMap()),
+ Collections.emptyList(),
+ null,
+ row));
+
+ assertEquals("MILVUS-10", exception.getSeaTunnelErrorCode().getCode());
+ }
+
+ @Test
+ void keepsNullFieldValueInMilvusDataWhenNullableFieldEnabled() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {null});
+
+ assertEquals(
+ JsonNull.INSTANCE,
+ new MilvusSinkConverter()
+ .buildMilvusData(
+ catalogTable(),
+ nullableFieldConfig(),
+ Collections.emptyList(),
+ null,
+ row)
+ .get("nullable_col"));
+ }
+
+ @Test
+ void throwsWhenNotNullableFieldValueIsNullAndNullableFieldEnabled() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {null});
+
+ MilvusConnectorException exception =
+ assertThrows(
+ MilvusConnectorException.class,
+ () ->
+ new MilvusSinkConverter()
+ .buildMilvusData(
+ catalogTable(false),
+ nullableFieldConfig(),
+ Collections.emptyList(),
+ null,
+ row));
+
+ assertEquals("MILVUS-10", exception.getSeaTunnelErrorCode().getCode());
+ }
+
+ @Test
+ void throwsWhenPartitionKeyValueIsNullAndNullableFieldEnabled() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {null});
+
+ MilvusConnectorException exception =
+ assertThrows(
+ MilvusConnectorException.class,
+ () ->
+ new MilvusSinkConverter()
+ .buildMilvusData(
+ catalogTable(),
+ nullablePartitionKeyConfig(),
+ Collections.emptyList(),
+ null,
+ row));
+
+ assertEquals("MILVUS-10", exception.getSeaTunnelErrorCode().getCode());
+ }
+
+ @Test
+ void throwsWhenDynamicFieldValueIsNullByDefault() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {null});
+
+ MilvusConnectorException exception =
+ assertThrows(
+ MilvusConnectorException.class,
+ () ->
+ new MilvusSinkConverter()
+ .buildMilvusData(
+ dynamicCatalogTable(),
+
ReadonlyConfig.fromMap(Collections.emptyMap()),
+ Collections.emptyList(),
+ "dynamic_col",
+ row));
+
+ assertEquals("MILVUS-10", exception.getSeaTunnelErrorCode().getCode());
+ }
+
+ private CatalogTable catalogTable() {
+ return catalogTable(true);
+ }
+
+ private CatalogTable catalogTable(boolean nullable) {
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "nullable_col",
+ BasicType.STRING_TYPE,
+ 0L,
+ nullable,
+ null,
+ ""))
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "");
+ }
+
+ private CatalogTable dynamicCatalogTable() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(CommonOptions.METADATA.getName(), true);
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ TableSchema.builder()
+ .column(
+ new PhysicalColumn(
+ "dynamic_col",
+ BasicType.STRING_TYPE,
+ 0L,
+ null,
+ false,
+ null,
+ "",
+ null,
+ null,
+ options))
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "");
+ }
+
+ private ReadonlyConfig nullableFieldConfig() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(MilvusSinkOptions.ENABLE_NULLABLE_FIELD.key(), true);
+ return ReadonlyConfig.fromMap(options);
+ }
+
+ private ReadonlyConfig nullablePartitionKeyConfig() {
+ Map<String, Object> options = new HashMap<>();
+ options.put(MilvusSinkOptions.ENABLE_NULLABLE_FIELD.key(), true);
+ options.put(MilvusSinkOptions.PARTITION_KEY.key(), "nullable_col");
+ return ReadonlyConfig.fromMap(options);
+ }
+
+ private Column columnWithSinkType(String name, String sinkType) {
+ return new PhysicalColumn(
+ name,
+ BasicType.SHORT_TYPE,
+ 0L,
+ null,
+ true,
+ null,
+ "",
+ sinkType,
+ null,
+ Collections.emptyMap());
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverterTest.java
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverterTest.java
new file mode 100644
index 0000000000..c323400566
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/utils/source/MilvusSourceConverterTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.milvus.utils.source;
+
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+
+import org.junit.jupiter.api.Test;
+
+import io.milvus.grpc.DataType;
+import io.milvus.grpc.FieldSchema;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class MilvusSourceConverterTest {
+
+ @Test
+ void convertColumnPreservesNullable() {
+ FieldSchema fieldSchema =
+ FieldSchema.newBuilder()
+ .setName("nullable_col")
+ .setDataType(DataType.VarChar)
+ .setNullable(true)
+ .build();
+
+ PhysicalColumn column =
MilvusSourceConverter.convertColumn(fieldSchema);
+
+ assertTrue(column.isNullable());
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java
index fe78508d7b..dd934c263d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/milvus/MilvusIT.java
@@ -54,9 +54,11 @@ import io.milvus.client.MilvusServiceClient;
import io.milvus.grpc.DataType;
import io.milvus.grpc.DescribeCollectionResponse;
import io.milvus.grpc.DescribeIndexResponse;
+import io.milvus.grpc.FieldData;
import io.milvus.grpc.FieldSchema;
import io.milvus.grpc.IndexDescription;
import io.milvus.grpc.KeyValuePair;
+import io.milvus.grpc.ListDatabasesResponse;
import io.milvus.grpc.MutationResult;
import io.milvus.grpc.QueryResults;
import io.milvus.param.ConnectParam;
@@ -100,13 +102,14 @@ import java.util.stream.Stream;
public class MilvusIT extends TestSuiteBase implements TestResource {
private static final String HOST = "milvus-e2e";
- private static final String MILVUS_IMAGE =
"milvusdb/milvus:2.4-20240711-7e2a9d6b";
+ private static final String MILVUS_IMAGE = "milvusdb/milvus:v2.5.11";
private static final String TOKEN = "root:Milvus";
private MilvusContainer container;
private MilvusServiceClient milvusClient;
private static final String COLLECTION_NAME = "simple_example";
private static final String COLLECTION_NAME_1 = "simple_example_1";
private static final String COLLECTION_NAME_2 = "simple_example_2";
+ private static final String COLLECTION_NAME_NULLABLE =
"simple_example_nullable";
private static final String COLLECTION_NAME_WITH_PARTITIONKEY =
"simple_example_with_partitionkey";
private static final String COLLECTION_NAME_WITH_PARTITIONS =
"simple_example_with_partitions";
@@ -642,6 +645,38 @@ public class MilvusIT extends TestSuiteBase implements
TestResource {
Assertions.assertTrue(fields.contains(TITLE_FIELD));
}
+ @TestTemplate
+ public void testFakeToMilvusWithNullableField(TestContainer container)
+ throws IOException, InterruptedException {
+ String databaseName = "test_nullable";
+ catalog.createDatabase(TablePath.of(databaseName,
COLLECTION_NAME_NULLABLE), true);
+
+ Container.ExecResult execResult =
container.executeJob("/fake-to-milvus-nullable.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
+
+ waitCollectionReady(databaseName, COLLECTION_NAME_NULLABLE,
VECTOR_FIELD);
+ Assertions.assertEquals(2, countCollectionEntities(databaseName,
COLLECTION_NAME_NULLABLE));
+
+ FieldSchema titleFieldSchema =
+ describeField(databaseName, COLLECTION_NAME_NULLABLE,
TITLE_FIELD);
+ Assertions.assertTrue(titleFieldSchema.getNullable());
+
+ FieldData nullTitleFieldData =
+ queryField(databaseName, COLLECTION_NAME_NULLABLE,
TITLE_FIELD, ID_FIELD + " == 1");
+ Assertions.assertEquals(
+ Collections.singletonList(false),
nullTitleFieldData.getValidDataList());
+ Assertions.assertEquals(
+ Collections.singletonList(""),
+ nullTitleFieldData.getScalars().getStringData().getDataList());
+
+ FieldData titleFieldData =
+ queryField(databaseName, COLLECTION_NAME_NULLABLE,
TITLE_FIELD, ID_FIELD + " == 2");
+ Assertions.assertEquals(Collections.singletonList(true),
titleFieldData.getValidDataList());
+ Assertions.assertEquals(
+ Collections.singletonList("nullable title"),
+ titleFieldData.getScalars().getStringData().getDataList());
+ }
+
@TestTemplate
public void testMultiFakeToMilvus(TestContainer container)
throws IOException, InterruptedException {
@@ -885,6 +920,26 @@ public class MilvusIT extends TestSuiteBase implements
TestResource {
private void waitCollectionReady(
String databaseName, String collectionName, String
vectorFieldName) {
+ // assert database exist
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ R<ListDatabasesResponse> listDatabasesResponse =
+ this.milvusClient.listDatabases();
+ Assertions.assertEquals(
+ R.Status.Success.getCode(),
+ listDatabasesResponse.getStatus(),
+
Optional.ofNullable(listDatabasesResponse.getException())
+ .map(Exception::getMessage)
+ .orElse(""));
+ return listDatabasesResponse
+ .getData()
+ .getDbNamesList()
+ .contains(databaseName);
+ });
+
// assert table exist
Awaitility.await()
.atMost(60, TimeUnit.SECONDS)
@@ -955,4 +1010,36 @@ public class MilvusIT extends TestSuiteBase implements
TestResource {
.getDataList()
.get(0);
}
+
+ private FieldSchema describeField(
+ String databaseName, String collectionName, String fieldName) {
+ R<DescribeCollectionResponse> describeCollectionResponse =
+ this.milvusClient.describeCollection(
+ DescribeCollectionParam.newBuilder()
+ .withDatabaseName(databaseName)
+ .withCollectionName(collectionName)
+ .build());
+ Assertions.assertEquals(R.Status.Success.getCode(),
describeCollectionResponse.getStatus());
+ return
describeCollectionResponse.getData().getSchema().getFieldsList().stream()
+ .filter(field -> fieldName.equals(field.getName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Field not found: " +
fieldName));
+ }
+
+ private FieldData queryField(
+ String databaseName, String collectionName, String fieldName,
String expr) {
+ R<QueryResults> queryResults =
+ milvusClient.query(
+ QueryParam.newBuilder()
+ .withDatabaseName(databaseName)
+ .withCollectionName(collectionName)
+ .withExpr(expr)
+
.withOutFields(Collections.singletonList(fieldName))
+ .build());
+ Assertions.assertEquals(R.Status.Success.getCode(),
queryResults.getStatus());
+ return queryResults.getData().getFieldsDataList().stream()
+ .filter(fieldData ->
fieldName.equals(fieldData.getFieldName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Field data not found: "
+ fieldName));
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/resources/fake-to-milvus-nullable.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/resources/fake-to-milvus-nullable.conf
new file mode 100644
index 0000000000..5681ee930f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-milvus-e2e/src/test/resources/fake-to-milvus-nullable.conf
@@ -0,0 +1,73 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ schema = {
+ table = "simple_example_nullable"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ nullable = false
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "nullable title"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, [0.1, 0.2, 0.3, 0.4], null]
+ },
+ {
+ kind = INSERT
+ fields = [2, [0.2, 0.3, 0.4, 0.5], "nullable title"]
+ }
+ ]
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://milvus-e2e:19530";
+ token = "root:Milvus"
+ database = "test_nullable"
+ enable_nullable_field = true
+ }
+}
diff --git
a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
index d5f46f5202..c4aaf5e436 100644
---
a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
+++
b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/JsonToRowConverters.java
@@ -32,6 +32,7 @@ import
org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.common.utils.DateTimeUtils;
import org.apache.seatunnel.common.utils.DateUtils;
import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
import java.io.IOException;
@@ -198,6 +199,13 @@ public class JsonToRowConverters implements Serializable {
return convertToBigDecimal(jsonNode);
}
};
+ case FLOAT_VECTOR:
+ return new JsonToObjectConverter() {
+ @Override
+ public Object convert(JsonNode jsonNode, String fieldName)
{
+ return convertToFloatVector(jsonNode, fieldName);
+ }
+ };
case ARRAY:
return createArrayConverter((ArrayType<?, ?>) type);
case MAP:
@@ -342,6 +350,27 @@ public class JsonToRowConverters implements Serializable {
return bigDecimal;
}
+ private Object convertToFloatVector(JsonNode jsonNode, String fieldName) {
+ if (!jsonNode.isArray()) {
+ throw new SeaTunnelJsonFormatException(
+ CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE,
+ String.format("Field '%s' expects an array for
FLOAT_VECTOR.", fieldName));
+ }
+ Float[] values = new Float[jsonNode.size()];
+ for (int i = 0; i < jsonNode.size(); i++) {
+ JsonNode element = jsonNode.get(i);
+ if (element == null || element.isNull() || !element.isNumber()) {
+ throw new SeaTunnelJsonFormatException(
+ CommonErrorCodeDeprecated.UNSUPPORTED_DATA_TYPE,
+ String.format(
+ "Field '%s' expects numeric values for
FLOAT_VECTOR, but element at index %d is '%s'.",
+ fieldName, i, element));
+ }
+ values[i] = convertToFloat(element);
+ }
+ return VectorUtils.toByteBuffer(values);
+ }
+
public JsonToObjectConverter createRowConverter(SeaTunnelRowType rowType) {
final JsonToObjectConverter[] fieldConverters =
Arrays.stream(rowType.getFieldTypes())
diff --git
a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
index 4194df4adf..66d038e105 100644
---
a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
+++
b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/JsonRowDataSerDeSchemaTest.java
@@ -35,6 +35,7 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.exception.CommonError;
import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.common.utils.VectorUtils;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
import org.junit.jupiter.api.Assertions;
@@ -42,6 +43,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.math.BigDecimal;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.LocalDate;
@@ -63,9 +65,12 @@ import static
org.apache.seatunnel.api.table.type.BasicType.INT_TYPE;
import static org.apache.seatunnel.api.table.type.BasicType.LONG_TYPE;
import static org.apache.seatunnel.api.table.type.BasicType.SHORT_TYPE;
import static org.apache.seatunnel.api.table.type.BasicType.STRING_TYPE;
+import static org.apache.seatunnel.api.table.type.VectorType.VECTOR_FLOAT_TYPE;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class JsonRowDataSerDeSchemaTest {
@@ -577,6 +582,36 @@ public class JsonRowDataSerDeSchemaTest {
assertEquals(expect, keyMap.keySet().iterator().next());
}
+ @Test
+ public void testFloatVectorConverter() throws JsonProcessingException {
+ JsonToRowConverters converters = new JsonToRowConverters(true, false);
+ JsonToRowConverters.JsonToObjectConverter converter =
+ converters.createConverter(VECTOR_FLOAT_TYPE);
+
+ ByteBuffer result =
+ (ByteBuffer)
+ converter.convert(
+ JsonUtils.stringToJsonNode("[0.1, 0.2, 0.3,
0.4]"), "vector");
+
+ assertArrayEquals(new Float[] {0.1f, 0.2f, 0.3f, 0.4f},
VectorUtils.toFloatArray(result));
+ }
+
+ @Test
+ public void testFloatVectorConverterRejectsInvalidElement() throws
JsonProcessingException {
+ JsonToRowConverters converters = new JsonToRowConverters(true, false);
+ JsonToRowConverters.JsonToObjectConverter converter =
+ converters.createConverter(VECTOR_FLOAT_TYPE);
+
+ SeaTunnelJsonFormatException exception =
+ assertThrows(
+ SeaTunnelJsonFormatException.class,
+ () ->
+ converter.convert(
+ JsonUtils.stringToJsonNode("[0.1,
null]"), "vector"));
+
+ assertTrue(exception.getMessage().contains("element at index 1"));
+ }
+
@Test
public void testParseUnsupportedDateTimeFormat() throws IOException {
SeaTunnelRowType rowType =
