lxy-9602 · Oct 30, 2026 · Oct 28, 2026
diff --git a/docs/source/user_guide/data_types.rst b/docs/source/user_guide/data_types.rst
@@ -194,6 +194,11 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        Paimon C++ supports precision values 0 (seconds), 3 (milliseconds), 6
        (microseconds), and 9 (nanoseconds).
 
+       As in Paimon Java, Lance data files store this type as an Arrow
+       timestamp with the ``UTC`` time zone whatever the local time zone is,
+       and reads return it in the local time zone. Only the time zone of the
+       Arrow type changes; the values since the epoch are kept as they are.
+
    * - ``ARRAY<t>``
      - List
      - Data type of an array of elements with same subtype.

diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp
@@ -1002,19 +1002,14 @@ Status SchemaValidation::ValidateLanceDataField(const std::shared_ptr<arrow::Fie
         case arrow::Type::DOUBLE:
         case arrow::Type::DATE32:
         case arrow::Type::TIME32:
+        // Lance cannot parse a time zone id containing ':', but LanceFormatWriter, like Paimon
+        // Java, writes every zoned timestamp with the "UTC" time zone.
+        case arrow::Type::TIMESTAMP:
         case arrow::Type::STRING:
         case arrow::Type::BINARY:
         case arrow::Type::DECIMAL128:
         case arrow::Type::FIXED_SIZE_LIST:
             return Status::OK();
-        case arrow::Type::TIMESTAMP: {
-            const auto& timestamp_type = checked_cast<const arrow::TimestampType&>(*type);
-            if (!timestamp_type.timezone().empty()) {
-                return Status::Invalid(
-                    "Lance file format does not support type LOCAL_ZONED_TIMESTAMP");
-            }
-            return Status::OK();
-        }
         case arrow::Type::LIST:
             return ValidateLanceDataField(type->field(0));
         case arrow::Type::STRUCT:

diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp
@@ -439,6 +439,14 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
         arrow::field("date", arrow::date32()),
         arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI)),
         arrow::field("timestamp", arrow::timestamp(arrow::TimeUnit::MICRO)),
+        arrow::field("ltz_0", arrow::timestamp(arrow::TimeUnit::SECOND, "UTC")),
+        arrow::field("ltz_3", arrow::timestamp(arrow::TimeUnit::MILLI, "Asia/Shanghai")),
+        arrow::field("ltz_6", arrow::timestamp(arrow::TimeUnit::MICRO, "GMT-10:00")),
+        arrow::field("ltz_9", arrow::timestamp(arrow::TimeUnit::NANO, "+08:00")),
+        arrow::field("ltz_array", arrow::list(arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"))),
+        arrow::field("nested_ltz",
+                     arrow::struct_({arrow::field(
+                         "values", arrow::list(arrow::timestamp(arrow::TimeUnit::MILLI, "UTC")))})),
         arrow::field("decimal", arrow::decimal128(38, 2)),
         arrow::field("array", arrow::list(arrow::float32())),
         arrow::field("nested_time",
@@ -474,11 +482,9 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
 
     arrow::FieldVector unsupported_fields = {
         arrow::field("map", arrow::map(arrow::int32(), arrow::utf8())),
-        arrow::field("ltz", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")),
         VariantTypeUtils::ToArrowField("variant"),
     };
-    std::vector<std::string> expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP",
-                                                "type VARIANT"};
+    std::vector<std::string> expected_errors = {"type MAP", "type VARIANT"};
     for (size_t i = 0; i < unsupported_fields.size(); ++i) {
         ASSERT_OK_AND_ASSIGN(
             table_schema,
@@ -504,6 +510,14 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
              {Options::CHANGELOG_FILE_FORMAT, "lance"}}) {
         std::map<std::string, std::string> alternate_format_options = {
             {Options::BUCKET, "-1"}, {Options::FILE_FORMAT, "parquet"}, {option_key, option_value}};
+        ASSERT_OK_AND_ASSIGN(
+            table_schema,
+            TableSchema::Create(
+                /*schema_id=*/0,
+                arrow::schema({arrow::field(
+                    "ltz", arrow::list(arrow::timestamp(arrow::TimeUnit::MICRO, "GMT-10:00")))}),
+                /*partition_keys=*/{}, /*primary_keys=*/{}, alternate_format_options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
         ASSERT_OK_AND_ASSIGN(
             table_schema,
             TableSchema::Create(

diff --git a/src/paimon/core/utils/nested_projection_utils.cpp b/src/paimon/core/utils/nested_projection_utils.cpp
@@ -727,6 +727,92 @@ std::shared_ptr<arrow::DataType> NormalizeLeafRepresentation(
     }
     return t;
 }
+
+std::shared_ptr<arrow::ArrayData> AlignTimestampTimezonesData(
+    const std::shared_ptr<arrow::ArrayData>& data,
+    const std::shared_ptr<arrow::DataType>& read_type) {
+    const std::shared_ptr<arrow::DataType>& type = data->type;
+    if (type->id() == arrow::Type::TIMESTAMP && read_type->id() == arrow::Type::TIMESTAMP) {
+        const auto& stored = checked_cast<const arrow::TimestampType&>(*type);
+        const auto& wanted = checked_cast<const arrow::TimestampType&>(*read_type);
+        if (stored.unit() != wanted.unit() || stored.timezone().empty() ||
+            wanted.timezone().empty() || stored.timezone() == wanted.timezone()) {
+            return data;
+        }
+        std::shared_ptr<arrow::ArrayData> retyped = data->Copy();
+        retyped->type = arrow::timestamp(wanted.unit(), wanted.timezone());
+        return retyped;
+    }
+    if (type->id() == arrow::Type::STRUCT && read_type->id() == arrow::Type::STRUCT) {
+        const auto& wanted_struct = checked_cast<const arrow::StructType&>(*read_type);
+        arrow::FieldVector fields = type->fields();
+        std::shared_ptr<arrow::ArrayData> retyped;
+        for (int32_t i = 0; i < static_cast<int32_t>(data->child_data.size()); ++i) {
+            const int32_t index = wanted_struct.GetFieldIndex(fields[i]->name());
+            if (index < 0) {
+                continue;
+            }
+            std::shared_ptr<arrow::ArrayData> child = AlignTimestampTimezonesData(
+                data->child_data[i], wanted_struct.field(index)->type());
+            if (child == data->child_data[i]) {
+                continue;
+            }
+            if (retyped == nullptr) {
+                retyped = data->Copy();
+            }
+            retyped->child_data[i] = child;
+            fields[i] = fields[i]->WithType(child->type);
+        }
+        if (retyped == nullptr) {
+            return data;
+        }
+        retyped->type = arrow::struct_(fields);
+        return retyped;
+    }
+    if ((type->id() == arrow::Type::LIST || type->id() == arrow::Type::FIXED_SIZE_LIST) &&
+        type->id() == read_type->id() && !data->child_data.empty()) {
+        std::shared_ptr<arrow::ArrayData> values =
+            AlignTimestampTimezonesData(data->child_data[0], read_type->field(0)->type());
+        if (values == data->child_data[0]) {
+            return data;
+        }
+        std::shared_ptr<arrow::ArrayData> retyped = data->Copy();
+        retyped->child_data[0] = values;
+        const std::shared_ptr<arrow::Field> value_field = type->field(0)->WithType(values->type);
+        retyped->type = type->id() == arrow::Type::LIST
+                            ? arrow::list(value_field)
+                            : arrow::fixed_size_list(
+                                  value_field,
+                                  checked_cast<const arrow::FixedSizeListType&>(*type).list_size());
+        return retyped;
+    }
+    if (type->id() == arrow::Type::MAP && read_type->id() == arrow::Type::MAP &&
+        !data->child_data.empty()) {
+        const auto& stored_map = checked_cast<const arrow::MapType&>(*type);
+        const auto& wanted_map = checked_cast<const arrow::MapType&>(*read_type);
+        const std::shared_ptr<arrow::ArrayData>& entries = data->child_data[0];
+        std::shared_ptr<arrow::ArrayData> keys =
+            AlignTimestampTimezonesData(entries->child_data[0], wanted_map.key_type());
+        std::shared_ptr<arrow::ArrayData> items =
+            AlignTimestampTimezonesData(entries->child_data[1], wanted_map.item_type());
+        if (keys == entries->child_data[0] && items == entries->child_data[1]) {
+            return data;
+        }
+        const std::shared_ptr<arrow::Field> key_field =
+            stored_map.key_field()->WithType(keys->type);
+        const std::shared_ptr<arrow::Field> item_field =
+            stored_map.item_field()->WithType(items->type);
+        std::shared_ptr<arrow::ArrayData> retyped_entries = entries->Copy();
+        retyped_entries->type = arrow::struct_({key_field, item_field});
+        retyped_entries->child_data = {keys, items};
+        std::shared_ptr<arrow::ArrayData> retyped = data->Copy();
+        retyped->type =
+            std::make_shared<arrow::MapType>(key_field, item_field, stored_map.keys_sorted());
+        retyped->child_data = {retyped_entries};
+        return retyped;
+    }
+    return data;
+}
 }  // namespace
 
 Result<std::shared_ptr<arrow::Array>> NestedProjectionUtils::AlignArrayToReadType(
@@ -840,4 +926,13 @@ Result<std::shared_ptr<arrow::Array>> NestedProjectionUtils::AlignArrayToReadTyp
     }
 }
 
+std::shared_ptr<arrow::Array> NestedProjectionUtils::AlignTimestampTimezones(
+    const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& read_type) {
+    std::shared_ptr<arrow::ArrayData> data = AlignTimestampTimezonesData(array->data(), read_type);
+    if (data == array->data()) {
+        return array;
+    }
+    return arrow::MakeArray(data);
+}
+
 }  // namespace paimon
diff --git a/src/paimon/core/utils/nested_projection_utils.h b/src/paimon/core/utils/nested_projection_utils.h
@@ -128,6 +128,16 @@ class PAIMON_EXPORT NestedProjectionUtils {
         const std::shared_ptr<arrow::Array>& array,
         const std::shared_ptr<arrow::DataType>& read_type, arrow::MemoryPool* pool);
 
+    /// Relabel a TIMESTAMP in `array` with the zone `read_type` asks for, only when both sides
+    /// have a zone and the same unit. A zoned value is an instant since the epoch, so only the
+    /// type changes. STRUCT children are matched by name, and LIST/FIXED_SIZE_LIST elements and
+    /// MAP keys and values are followed; every other difference, a unit change included, is left
+    /// as is.
+    /// @return `array` itself when no TIMESTAMP is relabelled, otherwise a shallow copy.
+    static std::shared_ptr<arrow::Array> AlignTimestampTimezones(
+        const std::shared_ptr<arrow::Array>& array,
+        const std::shared_ptr<arrow::DataType>& read_type);
+
  private:
     static Result<bool> HasNestedSubfieldProjectionType(
         const std::shared_ptr<arrow::DataType>& file_type,

diff --git a/src/paimon/core/utils/nested_projection_utils_test.cpp b/src/paimon/core/utils/nested_projection_utils_test.cpp
@@ -520,6 +520,94 @@ TEST(NestedProjectionUtilsTest, AlignArrayToReadTypeKeepsNestedLargeBinaryBlob)
     ASSERT_TRUE(blob_out->Equals(*blob));
 }
 
+TEST(NestedProjectionUtilsTest, AlignTimestampTimezonesRelabelsZonedLeavesOfSameUnit) {
+    auto make_type = [](const std::string& zone, arrow::TimeUnit::type other_unit) {
+        return arrow::struct_({
+            arrow::field("ltz", arrow::timestamp(arrow::TimeUnit::MICRO, zone)),
+            arrow::field("row", arrow::struct_({arrow::field(
+                                    "ltz", arrow::timestamp(arrow::TimeUnit::MILLI, zone))})),
+            arrow::field("list", arrow::list(arrow::timestamp(arrow::TimeUnit::MICRO, zone))),
+            arrow::field("fixed_list",
+                         arrow::fixed_size_list(arrow::timestamp(arrow::TimeUnit::MICRO, zone), 2)),
+            arrow::field("map", arrow::map(arrow::timestamp(arrow::TimeUnit::MILLI, zone),
+                                           arrow::timestamp(arrow::TimeUnit::MICRO, zone))),
+            arrow::field("other_unit", arrow::timestamp(other_unit, zone)),
+            arrow::field("ntz", arrow::timestamp(arrow::TimeUnit::MICRO)),
+        });
+    };
+    const std::string json = R"([[1, [2], [3, null], [16, null], [[11, 12], [13, null]], 4, 5],
+                                 [null, [null], null, null, null, null, null],
+                                 [6, [7], [8], [17, 18], [[14, 15]], 9, 10]])";
+    std::shared_ptr<arrow::DataType> data_type = make_type("UTC", arrow::TimeUnit::MICRO);
+    std::shared_ptr<arrow::Array> array =
+        arrow::ipc::internal::json::ArrayFromJSON(data_type, json).ValueOrDie();
+    ASSERT_EQ(NestedProjectionUtils::AlignTimestampTimezones(array, array->type()).get(),
+              array.get());
+
+    std::shared_ptr<arrow::DataType> read_type = make_type("Asia/Shanghai", arrow::TimeUnit::MILLI);
+    std::shared_ptr<arrow::Array> aligned =
+        NestedProjectionUtils::AlignTimestampTimezones(array, read_type);
+    auto expected_type = arrow::struct_({
+        arrow::field("ltz", arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai")),
+        arrow::field("row",
+                     arrow::struct_({arrow::field(
+                         "ltz", arrow::timestamp(arrow::TimeUnit::MILLI, "Asia/Shanghai"))})),
+        arrow::field("list",
+                     arrow::list(arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai"))),
+        arrow::field(
+            "fixed_list",
+            arrow::fixed_size_list(arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai"), 2)),
+        arrow::field("map", arrow::map(arrow::timestamp(arrow::TimeUnit::MILLI, "Asia/Shanghai"),
+                                       arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai"))),
+        arrow::field("other_unit", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")),
+        arrow::field("ntz", arrow::timestamp(arrow::TimeUnit::MICRO)),
+    });
+    std::shared_ptr<arrow::Array> expected =
+        arrow::ipc::internal::json::ArrayFromJSON(expected_type, json).ValueOrDie();
+    ASSERT_TRUE(aligned->type()->Equals(expected_type)) << aligned->type()->ToString();
+    ASSERT_TRUE(aligned->Equals(*expected));
+    ASSERT_TRUE(array->type()->Equals(data_type)) << array->type()->ToString();
+
+    std::shared_ptr<arrow::Array> aligned_slice =
+        NestedProjectionUtils::AlignTimestampTimezones(array->Slice(1), read_type);
+    ASSERT_TRUE(aligned_slice->Equals(*expected->Slice(1)));
+
+    std::shared_ptr<arrow::Array> zoned =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"),
+                                                  "[1, null]")
+            .ValueOrDie();
+    ASSERT_EQ(NestedProjectionUtils::AlignTimestampTimezones(
+                  zoned, arrow::timestamp(arrow::TimeUnit::MICRO))
+                  .get(),
+              zoned.get());
+    std::shared_ptr<arrow::Array> unzoned =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::timestamp(arrow::TimeUnit::MICRO),
+                                                  "[1, null]")
+            .ValueOrDie();
+    ASSERT_EQ(NestedProjectionUtils::AlignTimestampTimezones(
+                  unzoned, arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"))
+                  .get(),
+              unzoned.get());
+}
+
+TEST(NestedProjectionUtilsTest, AlignTimestampTimezonesMatchesStructChildrenByName) {
+    auto utc = arrow::timestamp(arrow::TimeUnit::MICRO, "UTC");
+    auto shanghai = arrow::timestamp(arrow::TimeUnit::MICRO, "Asia/Shanghai");
+    const std::string json = "[[1, 2], [null, null]]";
+    std::shared_ptr<arrow::Array> array =
+        arrow::ipc::internal::json::ArrayFromJSON(
+            arrow::struct_({arrow::field("unread", utc), arrow::field("read", utc)}), json)
+            .ValueOrDie();
+    std::shared_ptr<arrow::Array> aligned = NestedProjectionUtils::AlignTimestampTimezones(
+        array, arrow::struct_({arrow::field("read", shanghai)}));
+    auto expected_type =
+        arrow::struct_({arrow::field("unread", utc), arrow::field("read", shanghai)});
+    std::shared_ptr<arrow::Array> expected =
+        arrow::ipc::internal::json::ArrayFromJSON(expected_type, json).ValueOrDie();
+    ASSERT_TRUE(aligned->type()->Equals(expected_type)) << aligned->type()->ToString();
+    ASSERT_TRUE(aligned->Equals(*expected));
+}
+
 TEST(NestedProjectionUtilsTest, HasNestedSubfieldProjectionNoProjection) {
     auto file_schema = arrow::schema({
         MakeField("f0", arrow::int32(), 1),

diff --git a/src/paimon/format/lance/lance_file_batch_reader.cpp b/src/paimon/format/lance/lance_file_batch_reader.cpp
@@ -128,9 +128,14 @@ Result<BatchReader::ReadBatch> LanceFileBatchReader::AlignBatch(ReadBatch batch)
     if (array->type_id() != arrow::Type::STRUCT) {
         return Status::Invalid("Lance reader returned a non-struct Arrow batch");
     }
-    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> aligned,
-                           NestedProjectionUtils::AlignArrayToReadType(
-                               array, arrow::struct_(read_schema_->fields()), arrow_pool_.get()));
+    std::shared_ptr<arrow::DataType> read_type = arrow::struct_(read_schema_->fields());
+    // Zoned timestamps come back as UTC, as LanceFormatWriter and Paimon Java store them, while
+    // paimon asks for the local zone; the instants are the same.
+    std::shared_ptr<arrow::Array> relabelled =
+        NestedProjectionUtils::AlignTimestampTimezones(array, read_type);
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<arrow::Array> aligned,
+        NestedProjectionUtils::AlignArrayToReadType(relabelled, read_type, arrow_pool_.get()));
     auto out_array = std::make_unique<::ArrowArray>();
     auto out_schema = std::make_unique<::ArrowSchema>();
     PAIMON_RETURN_NOT_OK_FROM_ARROW(