[Feature][Connector-Milvus] support per-field timezone for naive timestamp to Timestamptz conversion

alohaha22 · claude · alohaha22 · commit 330c0e6f9682 · 2026-04-16T17:15:52.000+08:00
When source columns are tz-naive (e.g. PostgreSQL 'timestamp' without
timezone, MySQL 'datetime'), users can now specify a 'timezone' property
per field in field_schema to control how the wall-clock value is
interpreted when writing to Milvus Timestamptz fields. Accepts IANA zone
IDs (e.g. 'Asia/Shanghai') and UTC offsets (e.g. '+08:00').

Without the override, falls back to JVM systemDefault for backward
compatibility. OffsetDateTime and java.sql.Timestamp values are not
affected — they already carry absolute instants. String values use the
configured timezone when set, or the original behavior when not set.

Timezone values are validated at job submission time in MilvusSinkFactory
to fail fast on invalid zone IDs.

Config example:
  field_schema = [
    {field_name = "created_at", data_type = 26, timezone = "Asia/Shanghai"}
  ]

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
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
@@ -18,6 +18,8 @@
 package org.apache.seatunnel.connectors.seatunnel.milvus.sink;
 
 import com.google.auto.service.AutoService;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
@@ -27,8 +29,12 @@
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
+import org.apache.seatunnel.connectors.seatunnel.milvus.sink.catalog.MilvusFieldSchema;
 import org.apache.seatunnel.connectors.seatunnel.milvus.sink.config.MilvusSinkConfig;
 
+import java.time.ZoneId;
+import java.util.List;
+
 @AutoService(Factory.class)
 public class MilvusSinkFactory implements TableSinkFactory {
 
@@ -51,10 +57,35 @@ public OptionRule optionRule() {
 
     public TableSink createSink(TableSinkFactoryContext context) {
         ReadonlyConfig config = context.getOptions();
+        validateFieldTimezones(config);
         CatalogTable catalogTable = renameCatalogTable(config, context.getCatalogTable());
         return () -> new MilvusSink(config, catalogTable);
     }
 
+    private void validateFieldTimezones(ReadonlyConfig config) {
+        List<Object> rawFieldSchema = config.get(MilvusSinkConfig.FIELD_SCHEMA);
+        if (rawFieldSchema == null || rawFieldSchema.isEmpty()) {
+            return;
+        }
+        Gson gson = new Gson();
+        List<MilvusFieldSchema> fieldSchemaList = gson.fromJson(
+                gson.toJson(rawFieldSchema),
+                new TypeToken<List<MilvusFieldSchema>>() {}.getType());
+        for (MilvusFieldSchema fs : fieldSchemaList) {
+            if (fs.getTimezone() != null && !fs.getTimezone().isEmpty()) {
+                try {
+                    ZoneId.of(fs.getTimezone());
+                } catch (Exception e) {
+                    throw new IllegalArgumentException(
+                            "Invalid timezone '" + fs.getTimezone()
+                                    + "' for field '" + fs.getEffectiveFieldName()
+                                    + "'. Use IANA zone ID (e.g. 'Asia/Shanghai') "
+                                    + "or UTC offset (e.g. '+08:00').", e);
+                }
+            }
+        }
+    }
+
     private CatalogTable renameCatalogTable(
             ReadonlyConfig config, CatalogTable sourceCatalogTable) {
         TableIdentifier sourceTableId = sourceCatalogTable.getTableId();
diff --git a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/catalog/MilvusFieldSchema.java b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/catalog/MilvusFieldSchema.java
@@ -64,6 +64,9 @@ public class MilvusFieldSchema {
     @SerializedName("enable_match")
     private Boolean enableMatch;
 
+    @SerializedName("timezone")
+    private String timezone;
+
     /**
      * Get the effective field name (field_name if specified, otherwise source_field_name)
      */
diff --git a/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/utils/MilvusSinkConverter.java b/seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/utils/MilvusSinkConverter.java
@@ -43,6 +43,9 @@
 import org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectionErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.milvus.sink.catalog.MilvusFieldSchema;
+import org.apache.seatunnel.connectors.seatunnel.milvus.sink.config.MilvusSinkConfig;
+
+import lombok.extern.slf4j.Slf4j;
 
 import java.nio.ByteBuffer;
 import java.time.LocalDateTime;
@@ -51,11 +54,13 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+@Slf4j
 public class MilvusSinkConverter {
     private static final Gson gson = new Gson();
 
@@ -68,16 +73,19 @@ public class MilvusSinkConverter {
      * file; it lives entirely on {@code GeometryConverter}.
      */
     private final GeometryConverter geometryConverter;
+    private final Map<String, ZoneId> fieldTimezoneOverrides;
 
     public MilvusSinkConverter() {
-        // Matches the production default (GEOMETRY_CONVERT_MODE = "passthrough"):
-        // Geometry strings flow byte-for-byte to Milvus. Tests that need the
-        // parsing path should construct with new MilvusSinkConverter(GeometryConverter.PARSE).
-        this(GeometryConverter.PASSTHROUGH);
+        this(GeometryConverter.PASSTHROUGH, Collections.emptyMap());
     }
 
     public MilvusSinkConverter(GeometryConverter geometryConverter) {
+        this(geometryConverter, Collections.emptyMap());
+    }
+
+    public MilvusSinkConverter(GeometryConverter geometryConverter, Map<String, ZoneId> fieldTimezoneOverrides) {
         this.geometryConverter = geometryConverter;
+        this.fieldTimezoneOverrides = fieldTimezoneOverrides;
     }
 
     /**
@@ -87,7 +95,31 @@ public MilvusSinkConverter(GeometryConverter geometryConverter) {
      * they need without polluting the writer with sub-domain knowledge.
      */
     public static MilvusSinkConverter fromConfig(ReadonlyConfig config) {
-        return new MilvusSinkConverter(GeometryConverter.fromConfig(config));
+        GeometryConverter geo = GeometryConverter.fromConfig(config);
+        Map<String, ZoneId> tzOverrides = parseFieldTimezones(config);
+        return new MilvusSinkConverter(geo, tzOverrides);
+    }
+
+    private static Map<String, ZoneId> parseFieldTimezones(ReadonlyConfig config) {
+        List<Object> rawFieldSchema = config.get(MilvusSinkConfig.FIELD_SCHEMA);
+        if (rawFieldSchema == null || rawFieldSchema.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        List<MilvusFieldSchema> fieldSchemaList = gson.fromJson(
+                gson.toJson(rawFieldSchema),
+                new TypeToken<List<MilvusFieldSchema>>() {}.getType());
+        Map<String, ZoneId> result = new HashMap<>();
+        for (MilvusFieldSchema fs : fieldSchemaList) {
+            if (fs.getTimezone() != null && !fs.getTimezone().isEmpty()) {
+                String name = fs.getEffectiveFieldName();
+                if (name != null) {
+                    ZoneId zone = ZoneId.of(fs.getTimezone());
+                    result.put(name, zone);
+                    log.info("Milvus field '{}' naive-to-aware timezone: {}", name, zone);
+                }
+            }
+        }
+        return result;
     }
 
     public Object convertBySeaTunnelType(
@@ -485,24 +517,39 @@ public Object convertByMilvusType(FieldSchema fieldSchema, Object value) {
             case Timestamptz:
                 // Milvus SDK requires Timestamptz as ISO 8601 String format (e.g., "2024-01-19T11:30:45Z")
                 // Reference: Milvus ParamUtils.java line 430+ requires String type for Timestamptz
+                ZoneId tzOverride = fieldTimezoneOverrides.get(fieldSchema.getName());
                 if (value instanceof java.sql.Timestamp) {
-                    // Convert java.sql.Timestamp to ISO 8601 string (UTC)
+                    // java.sql.Timestamp internally holds epoch millis (an absolute instant).
+                    // toInstant() is always correct — per-field timezone does not apply.
                     return ((java.sql.Timestamp) value).toInstant().toString();
                 } else if (value instanceof LocalDateTime) {
-                    // Convert LocalDateTime (systemDefault) to ISO 8601 string
-                    return ((LocalDateTime) value).atZone(ZoneId.systemDefault()).toInstant().toString();
+                    ZoneId zone = tzOverride != null ? tzOverride : ZoneId.systemDefault();
+                    return ((LocalDateTime) value).atZone(zone).toInstant().toString();
                 } else if (value instanceof OffsetDateTime) {
-                    // Convert OffsetDateTime to ISO 8601 string (preserves timezone)
+                    // OffsetDateTime already carries offset — toInstant() is always correct.
+                    // Per-field timezone does not apply.
                     return ((OffsetDateTime) value).toInstant().toString();
                 } else if (value instanceof String) {
-                    // Already a string - validate and/or convert to ISO 8601 format
-                    String strValue = value.toString();
-                    // Try to parse and normalize to ISO 8601
+                    String strValue = value.toString().trim();
+                    if (tzOverride != null) {
+                        // User explicitly configured timezone — parse as naive wall-clock
+                        // and apply the configured zone, consistent with LocalDateTime path.
+                        // Timestamp.valueOf only accepts "yyyy-MM-dd HH:mm:ss[.fff]" format,
+                        // so strings with timezone info (Z, +08:00, etc.) naturally fail
+                        // and pass through as-is.
+                        try {
+                            LocalDateTime ldt = java.sql.Timestamp.valueOf(
+                                    strValue.replace('T', ' ')).toLocalDateTime();
+                            return ldt.atZone(tzOverride).toInstant().toString();
+                        } catch (IllegalArgumentException e) {
+                            return strValue;
+                        }
+                    }
+                    // No timezone override — original behavior
                     try {
                         java.sql.Timestamp ts = java.sql.Timestamp.valueOf(strValue);
                         return ts.toInstant().toString();
                     } catch (IllegalArgumentException e) {
-                        // If it's already in ISO 8601 format, return as-is
                         return strValue;
                     }
                 }
diff --git a/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/utils/MilvusSinkConverterTest.java b/seatunnel-connectors-v2/connector-milvus/src/test/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/utils/MilvusSinkConverterTest.java
@@ -26,15 +26,18 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.connectors.seatunnel.milvus.exception.MilvusConnectorException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 public class MilvusSinkConverterTest {
 
@@ -134,6 +137,172 @@ public void testConvertByMilvusType_Timestamptz_InvalidStringFallback() {
         Assertions.assertEquals("not-a-date", result);
     }
 
+    // --- Per-field timezone handling for Timestamptz ---
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_LocalDateTime_PerFieldShanghai() {
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("Asia/Shanghai"));
+        MilvusSinkConverter shanghaiConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = shanghaiConverter.convertByMilvusType(schema, ldt);
+        // 10:00 Shanghai = 02:00 UTC
+        Assertions.assertEquals("2024-01-19T02:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_LocalDateTime_PerFieldUtc() {
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("UTC"));
+        MilvusSinkConverter utcConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = utcConverter.convertByMilvusType(schema, ldt);
+        Assertions.assertEquals("2024-01-19T10:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_LocalDateTime_PerFieldUtcOffset() {
+        // UTC offset format "+08:00" should work the same as IANA "Asia/Shanghai"
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("+08:00"));
+        MilvusSinkConverter offsetConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = offsetConverter.convertByMilvusType(schema, ldt);
+        // 10:00 +08:00 = 02:00 UTC
+        Assertions.assertEquals("2024-01-19T02:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_LocalDateTime_NoOverrideFallsBackToSystemDefault() {
+        // No per-field override → systemDefault
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = converter.convertByMilvusType(schema, ldt);
+        String expected = ldt.atZone(ZoneId.systemDefault()).toInstant().toString();
+        Assertions.assertEquals(expected, result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_MultipleFieldsDifferentZones() {
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("created_at", ZoneId.of("Asia/Shanghai"));
+        overrides.put("updated_at", ZoneId.of("US/Eastern"));
+        MilvusSinkConverter multiConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+
+        LocalDateTime ldt = LocalDateTime.of(2024, 6, 15, 12, 0, 0);
+
+        FieldSchema createdSchema = FieldSchema.builder()
+                .name("created_at").dataType(DataType.Timestamptz).build();
+        Object createdResult = multiConverter.convertByMilvusType(createdSchema, ldt);
+        // 12:00 Shanghai (+8) = 04:00 UTC
+        Assertions.assertEquals("2024-06-15T04:00:00Z", createdResult);
+
+        FieldSchema updatedSchema = FieldSchema.builder()
+                .name("updated_at").dataType(DataType.Timestamptz).build();
+        Object updatedResult = multiConverter.convertByMilvusType(updatedSchema, ldt);
+        // 12:00 US/Eastern (EDT = -4 in June) = 16:00 UTC
+        Assertions.assertEquals("2024-06-15T16:00:00Z", updatedResult);
+    }
+
+    @Test
+    public void testFromConfig_FieldSchemaTimezone_WiredToConverter() {
+        // Simulate field_schema config with timezone
+        Map<String, Object> fieldEntry = new HashMap<>();
+        fieldEntry.put("field_name", "event_time");
+        fieldEntry.put("data_type", 26);  // Timestamptz
+        fieldEntry.put("timezone", "Asia/Shanghai");
+
+        List<Object> fieldSchemaList = new java.util.ArrayList<>();
+        fieldSchemaList.add(fieldEntry);
+
+        Map<String, Object> configMap = new HashMap<>();
+        configMap.put("field_schema", fieldSchemaList);
+        ReadonlyConfig config = ReadonlyConfig.fromMap(configMap);
+        MilvusSinkConverter shanghaiConverter = MilvusSinkConverter.fromConfig(config);
+
+        FieldSchema schema = FieldSchema.builder()
+                .name("event_time").dataType(DataType.Timestamptz).build();
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = shanghaiConverter.convertByMilvusType(schema, ldt);
+        // 10:00 Shanghai = 02:00 UTC
+        Assertions.assertEquals("2024-01-19T02:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_String_WithOverrideUsesConfiguredTz() {
+        // String input + per-field timezone → parse as naive wall-clock, apply configured tz
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("Asia/Shanghai"));
+        MilvusSinkConverter shanghaiConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        Object result = shanghaiConverter.convertByMilvusType(schema, "2024-01-19 10:00:00");
+        // 10:00 Shanghai = 02:00 UTC
+        Assertions.assertEquals("2024-01-19T02:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_String_WithoutOverrideUsesOriginalBehavior() {
+        // String input + no per-field timezone → original Timestamp.valueOf behavior
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        Object result = converter.convertByMilvusType(schema, "2024-01-19 10:00:00");
+        Assertions.assertInstanceOf(String.class, result);
+        Assertions.assertTrue(result.toString().contains("2024-01-19"));
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_String_Iso8601PassedThrough() {
+        // ISO 8601 with offset — should pass through even with per-field timezone
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("Asia/Shanghai"));
+        MilvusSinkConverter shanghaiConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        Object result = shanghaiConverter.convertByMilvusType(schema, "2024-01-19T10:00:00Z");
+        // Already has offset Z — LocalDateTime.parse fails, falls through as-is
+        Assertions.assertEquals("2024-01-19T10:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertByMilvusType_Timestamptz_OffsetDateTime_IgnoresPerFieldTimezone() {
+        // OffsetDateTime already carries offset — per-field timezone must NOT affect it
+        Map<String, ZoneId> overrides = new HashMap<>();
+        overrides.put("ts", ZoneId.of("Asia/Shanghai"));
+        MilvusSinkConverter shanghaiConverter = new MilvusSinkConverter(
+                GeometryConverter.PASSTHROUGH, overrides);
+        FieldSchema schema = FieldSchema.builder()
+                .name("ts").dataType(DataType.Timestamptz).build();
+        OffsetDateTime odt = OffsetDateTime.of(2024, 1, 19, 10, 0, 0, 0, ZoneOffset.ofHours(5));
+        Object result = shanghaiConverter.convertByMilvusType(schema, odt);
+        // 10:00+05 = 05:00 UTC, regardless of per-field timezone setting
+        Assertions.assertEquals("2024-01-19T05:00:00Z", result);
+    }
+
+    @Test
+    public void testConvertBySeaTunnelType_TimestampTz_LocalDateTime_UsesSystemDefault() {
+        // convertBySeaTunnelType has no field name, always uses systemDefault
+        LocalDateTime ldt = LocalDateTime.of(2024, 1, 19, 10, 0, 0);
+        Object result = converter.convertBySeaTunnelType(
+                LocalTimeType.OFFSET_DATE_TIME_TYPE, false, ldt);
+        Assertions.assertInstanceOf(String.class, result);
+        String expected = ldt.atZone(ZoneId.systemDefault()).toInstant().toString();
+        Assertions.assertEquals(expected, result);
+    }
+
     // --- Consistency: both paths return same type for same input ---
 
     @Test
