feat(lance): support TIME type

Paimon Java accepts TIME columns in Lance data files and writes them as
time32[ms], but Paimon C++ rejected TIME in Lance schema validation,
both at top level and nested in ARRAY or ROW. Accept TIME there, and
create empty integer statistics for a TIME column in LanceStatsExtractor
so that finalizing a Lance data file with one does not fail.

Document TIME in the data types guide, which still listed it as not
supported: its time32[ms] mapping, how the declared precision is kept,
and which file formats read or write it.

Author: SteNicholas

docs/source/user_guide/data_types.rst

@@ -146,13 +146,24 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
    * - ``TIME``
 
        ``TIME(p)``
-     - Not Supported
+     - Time32
      - Data type of a time without time zone consisting of hour:minute:second[.fractional] with up to nanosecond precision and values ranging from 00:00:00.000000000 to 23:59:59.999999999.
 
        The type can be declared using ``TIME(p)`` where p is the number of digits of
        fractional seconds (precision). p must have a value between 0 and 9
        (both inclusive). If no precision is specified, p is equal to 0.
 
+       Paimon C++ maps every ``TIME(p)`` to an Arrow ``Time32`` in milliseconds
+       (``time32[ms]``) and accepts no other Arrow time type. The declared
+       precision p is kept in the ``paimon.time.precision`` field metadata; a
+       ``Time32`` field without it is ``TIME(0)``. As in Paimon Java, values are
+       stored as milliseconds of the day whatever p is, so ``TIME(6)`` and
+       ``TIME(9)`` values keep only millisecond precision.
+
+       Paimon C++ reads TIME columns from Parquet, Lance and Vortex data files
+       and writes them to Lance and Vortex data files. ORC and Avro data files
+       do not support TIME, and a TIME column cannot be a partition key.
+
    * - ``TIMESTAMP``
 
        ``TIMESTAMP(p)``

src/paimon/core/schema/schema_validation.cpp

@@ -1001,6 +1001,7 @@ Status SchemaValidation::ValidateLanceDataField(const std::shared_ptr<arrow::Fie
         case arrow::Type::FLOAT:
         case arrow::Type::DOUBLE:
         case arrow::Type::DATE32:
+        case arrow::Type::TIME32:
         case arrow::Type::STRING:
         case arrow::Type::BINARY:
         case arrow::Type::DECIMAL128:

src/paimon/core/schema/schema_validation_test.cpp

@@ -437,9 +437,13 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
         arrow::field("string", arrow::utf8()),
         arrow::field("binary", arrow::binary()),
         arrow::field("date", arrow::date32()),
+        arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI)),
         arrow::field("timestamp", arrow::timestamp(arrow::TimeUnit::MICRO)),
         arrow::field("decimal", arrow::decimal128(38, 2)),
         arrow::field("array", arrow::list(arrow::float32())),
+        arrow::field("nested_time",
+                     arrow::struct_({arrow::field(
+                         "values", arrow::list(arrow::time32(arrow::TimeUnit::MILLI)))})),
         arrow::field("row", arrow::struct_({arrow::field("value", arrow::int32())}),
                      /*nullable=*/false),
         arrow::field("nullable_row", arrow::struct_({arrow::field("value", arrow::int32())})),
@@ -472,13 +476,9 @@ TEST(SchemaValidationTest, TestLanceDataTypes) {
         arrow::field("map", arrow::map(arrow::int32(), arrow::utf8())),
         arrow::field("ltz", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")),
         VariantTypeUtils::ToArrowField("variant"),
-        arrow::field("time_millis", arrow::time32(arrow::TimeUnit::MILLI)),
-        arrow::field("nested_time",
-                     arrow::struct_({arrow::field(
-                         "values", arrow::list(arrow::time32(arrow::TimeUnit::MILLI)))})),
     };
-    std::vector<std::string> expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP", "type VARIANT",
-                                                "type time32", "type time32"};
+    std::vector<std::string> expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP",
+                                                "type VARIANT"};
     for (size_t i = 0; i < unsupported_fields.size(); ++i) {
         ASSERT_OK_AND_ASSIGN(
             table_schema,

src/paimon/format/lance/lance_file_format_test.cpp

@@ -29,6 +29,7 @@
 #include "arrow/ipc/json_simple.h"
 #include "gtest/gtest.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/types/data_type.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -176,6 +177,11 @@ TEST(LanceUtilsTest, ConvertsObjectStoreOptions) {
 }
 
 TEST_F(LanceFileFormatTest, WriteThenReadSupportedTypes) {
+    auto time_field = [](const std::string& name, int32_t precision) {
+        return arrow::field(
+            name, arrow::time32(arrow::TimeUnit::MILLI), /*nullable=*/true,
+            arrow::key_value_metadata({DataType::kTimePrecision}, {std::to_string(precision)}));
+    };
     arrow::FieldVector fields = {
         arrow::field("bool_col", arrow::boolean()),
         arrow::field("i8_col", arrow::int8()),
@@ -187,22 +193,30 @@ TEST_F(LanceFileFormatTest, WriteThenReadSupportedTypes) {
         arrow::field("string_col", arrow::utf8()),
         arrow::field("binary_col", arrow::binary()),
         arrow::field("date_col", arrow::date32()),
+        time_field("time_col", 3),
         arrow::field("timestamp_col", arrow::timestamp(arrow::TimeUnit::MICRO)),
         arrow::field("decimal_col", arrow::decimal128(12, 2)),
         arrow::field("array_col", arrow::list(arrow::float32())),
         arrow::field("row_col", arrow::struct_({arrow::field("value", arrow::int32())})),
         arrow::field(
             "vector_col",
             arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 2)),
+        arrow::field("time_array_col", arrow::list(time_field("item", 0))),
+        arrow::field("time_row_col", arrow::struct_({time_field("value", 6)})),
+        arrow::field("time_row_array_col",
+                     arrow::struct_({arrow::field("values", arrow::list(time_field("item", 9)))})),
     };
     std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
     std::shared_ptr<arrow::Array> expected =
         arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
-          [true,-1,-2,-3,-4,1.5,2.5,"one","abc",1,
-           "1970-01-01 00:00:00.000001","12.34",[1.0,2.0],[7],[3.0,4.0]],
-          [false,1,2,3,4,3.5,4.5,"two","xyz",2,
-           "2030-12-31 23:59:59.999999","-12.34",[],[8],[5.0,6.0]],
-          [null,null,null,null,null,null,null,null,null,null,null,null,null,[null],null]
+          [true,-1,-2,-3,-4,1.5,2.5,"one","abc",1,0,
+           "1970-01-01 00:00:00.000001","12.34",[1.0,2.0],[7],[3.0,4.0],
+           [0,null,45296000],[45296123],[[3723123,null]]],
+          [false,1,2,3,4,3.5,4.5,"two","xyz",2,86399999,
+           "2030-12-31 23:59:59.999999","-12.34",[],[8],[5.0,6.0],
+           [],[86399999],[[]]],
+          [null,null,null,null,null,null,null,null,null,null,null,null,null,null,[null],null,
+           null,[null],[null]]
         ])")
             .ValueOrDie();
     std::string path = PathUtil::JoinPath(directory_->Str(), "supported-types.lance");
@@ -219,6 +233,19 @@ TEST_F(LanceFileFormatTest, WriteThenReadSupportedTypes) {
         ASSERT_TRUE(actual->Equals(arrow::ChunkedArray(expected))) << actual->ToString() << "\nvs\n"
                                                                    << expected->ToString();
     }
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileBatchReader> reader, OpenReader(path, 2));
+    ::ArrowSchema ffi_schema = {};
+    ASSERT_TRUE(arrow::ExportSchema(*schema, &ffi_schema).ok());
+    ASSERT_OK(reader->SetReadSchema(&ffi_schema, /*predicate=*/nullptr,
+                                    /*selection_bitmap=*/std::nullopt));
+    ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatch batch, reader->NextBatch());
+    arrow::Result<std::shared_ptr<arrow::Array>> batch_result =
+        arrow::ImportArray(batch.first.get(), batch.second.get());
+    ASSERT_TRUE(batch_result.ok()) << batch_result.status().ToString();
+    std::shared_ptr<arrow::Array> batch_array = std::move(batch_result).MoveValueUnsafe();
+    ASSERT_TRUE(batch_array->type()->Equals(arrow::struct_(fields), /*check_metadata=*/true))
+        << batch_array->type()->ToString(/*show_metadata=*/true);
 }
 
 TEST_F(LanceFileFormatTest, RejectsNullTopLevelRows) {
@@ -351,10 +378,12 @@ TEST_F(LanceFileFormatTest, EmptyProjectionAndEmptySelection) {
 
 TEST_F(LanceFileFormatTest, ExtractStatistics) {
     arrow::FieldVector fields = {arrow::field("id", arrow::int32()),
-                                 arrow::field("name", arrow::utf8())};
+                                 arrow::field("name", arrow::utf8()),
+                                 arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI))};
     std::shared_ptr<arrow::Schema> schema = arrow::schema(fields);
     std::shared_ptr<arrow::Array> data =
-        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([[1,"a"],[2,null]])")
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields),
+                                                  R"([[1,"a",3723123],[2,null,null]])")
             .ValueOrDie();
     std::string path = PathUtil::JoinPath(directory_->Str(), "stats.lance");
     ASSERT_OK(WriteFile(path, schema, data, /*batch_size=*/2));
@@ -365,9 +394,10 @@ TEST_F(LanceFileFormatTest, ExtractStatistics) {
                          format_->CreateStatsExtractor(&ffi_schema));
     ASSERT_OK_AND_ASSIGN(auto result, extractor->ExtractWithFileInfo(file_system_, path, pool_));
     ASSERT_EQ(result.second.GetRowCount(), 2);
-    ASSERT_EQ(result.first.size(), 2);
+    ASSERT_EQ(result.first.size(), 3);
     ASSERT_EQ(result.first[0]->ToString(), "min null, max null, null count null");
     ASSERT_EQ(result.first[1]->ToString(), "min null, max null, null count null");
+    ASSERT_EQ(result.first[2]->ToString(), "min null, max null, null count null");
 }
 
 TEST_F(LanceFileFormatTest, RejectsIncompatibleReadTypes) {

src/paimon/format/lance/lance_stats_extractor.cpp

@@ -86,6 +86,8 @@ Result<std::unique_ptr<ColumnStats>> LanceStatsExtractor::CreateEmptyStats(
             return ColumnStats::CreateStringColumnStats(std::nullopt, std::nullopt, std::nullopt);
         case arrow::Type::DATE32:
             return ColumnStats::CreateDateColumnStats(std::nullopt, std::nullopt, std::nullopt);
+        case arrow::Type::TIME32:
+            return ColumnStats::CreateIntColumnStats(std::nullopt, std::nullopt, std::nullopt);
         case arrow::Type::TIMESTAMP: {
             std::shared_ptr<arrow::TimestampType> timestamp_type =
                 checked_pointer_cast<arrow::TimestampType>(type);

test/inte/write_and_read_inte_test.cpp

@@ -5978,37 +5978,52 @@ TEST(VortexPrimaryKeyWriteInteTest, TestPKWriteFinalize) {
                                                     R"([[0, "a", 1],[0, "b", 2]])"));
     ASSERT_TRUE(success);
 }
+#endif
 
+#if defined(PAIMON_ENABLE_LANCE) || defined(PAIMON_ENABLE_VORTEX)
 // Regression: append-table finalization always extracts statistics, and TIME has no statistics
 // factory of its own, so the extractor must still yield one entry for a TIME column.
-TEST(VortexTimeColumnWriteInteTest, TestAppendTimeColumnFinalize) {
-    auto dir = UniqueTestDirectory::Create("local");
-    ASSERT_NE(dir, nullptr);
-    arrow::FieldVector fields = {arrow::field("f_time", arrow::time32(arrow::TimeUnit::MILLI)),
-                                 arrow::field("f1", arrow::int32())};
-    auto schema = arrow::schema(fields);
-    std::map<std::string, std::string> options = {
-        {Options::FILE_FORMAT, "vortex"}, {Options::BUCKET, "-1"}, {Options::FILE_SYSTEM, "local"}};
-    ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
-                                                         /*primary_keys=*/{}, options,
-                                                         /*is_streaming_mode=*/false));
-    int64_t commit_identifier = 0;
-    ASSERT_OK_AND_ASSIGN(
-        std::unique_ptr<RecordBatch> batch,
-        TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[3723123, 1],[null, 2]])",
-                                    /*partition_map=*/{}, /*bucket=*/0, {}));
-    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
-                         helper->WriteAndCommit(std::move(batch), commit_identifier++,
-                                                /*expected_commit_messages=*/std::nullopt));
-    arrow::FieldVector fields_with_row_kind = fields;
-    fields_with_row_kind.insert(fields_with_row_kind.begin(),
-                                arrow::field("_VALUE_KIND", arrow::int8()));
-    ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> splits,
-                         helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
-    ASSERT_OK_AND_ASSIGN(bool success,
-                         helper->ReadAndCheckResult(arrow::struct_(fields_with_row_kind), splits,
-                                                    R"([[0, 3723123, 1],[0, null, 2]])"));
-    ASSERT_TRUE(success);
+TEST(TimeColumnWriteInteTest, TestAppendTimeColumnFinalize) {
+    std::vector<std::string> file_formats;
+#ifdef PAIMON_ENABLE_LANCE
+    file_formats.push_back("lance");
+#endif
+#ifdef PAIMON_ENABLE_VORTEX
+    file_formats.push_back("vortex");
+#endif
+    for (const auto& file_format : file_formats) {
+        SCOPED_TRACE(file_format);
+        auto dir = UniqueTestDirectory::Create("local");
+        ASSERT_NE(dir, nullptr);
+        arrow::FieldVector fields = {arrow::field("f_time", arrow::time32(arrow::TimeUnit::MILLI)),
+                                     arrow::field("f1", arrow::int32())};
+        auto schema = arrow::schema(fields);
+        std::map<std::string, std::string> options = {{Options::FILE_FORMAT, file_format},
+                                                      {Options::BUCKET, "-1"},
+                                                      {Options::FILE_SYSTEM, "local"}};
+        ASSERT_OK_AND_ASSIGN(auto helper,
+                             TestHelper::Create(dir->Str(), schema, /*partition_keys=*/{},
+                                                /*primary_keys=*/{}, options,
+                                                /*is_streaming_mode=*/false));
+        int64_t commit_identifier = 0;
+        ASSERT_OK_AND_ASSIGN(
+            std::unique_ptr<RecordBatch> batch,
+            TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[3723123, 1],[null, 2]])",
+                                        /*partition_map=*/{}, /*bucket=*/0, {}));
+        ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                             helper->WriteAndCommit(std::move(batch), commit_identifier++,
+                                                    /*expected_commit_messages=*/std::nullopt));
+        arrow::FieldVector fields_with_row_kind = fields;
+        fields_with_row_kind.insert(fields_with_row_kind.begin(),
+                                    arrow::field("_VALUE_KIND", arrow::int8()));
+        ASSERT_OK_AND_ASSIGN(
+            std::vector<std::shared_ptr<Split>> splits,
+            helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt));
+        ASSERT_OK_AND_ASSIGN(
+            bool success, helper->ReadAndCheckResult(arrow::struct_(fields_with_row_kind), splits,
+                                                     R"([[0, 3723123, 1],[0, null, 2]])"));
+        ASSERT_TRUE(success);
+    }
 }
 #endif
 

test/test_data/lance/append_java_compat.db/append_java_compat/README.md

@@ -54,5 +54,6 @@ Commit - snapshot 2 (4 rows total)
 At this Java revision, nullable ROW declarations are accepted but actual null ROW parents
 do not survive the Lance round trip (a separate probe read id=2's null parent as non-null).
 Both struct parents here are valid, including id=2 where their children are NULL.
-TIME32 is omitted because the C++ table path does not support it; Java's Lance validator
-rejects local-zoned timestamps, MAP, MULTISET, VARIANT and BLOB.
+TIME32 is omitted because the C++ table path did not support it when this fixture was
+written. At this revision, Java's Lance validator rejects local-zoned timestamps, MAP,
+MULTISET, VARIANT and BLOB.