fix(read): support Parquet TIME columns

src/paimon/common/types/data_type.cpp

@@ -41,6 +41,27 @@
 
 namespace paimon {
 
+Result<int32_t> DataType::GetTimePrecision(const arrow::Field& field) {
+    return GetTimePrecision(field.type(), field.metadata());
+}
+
+Result<int32_t> DataType::GetTimePrecision(
+    const std::shared_ptr<arrow::DataType>& type,
+    const std::shared_ptr<const arrow::KeyValueMetadata>& metadata) {
+    if (type->id() != arrow::Type::TIME32 ||
+        checked_cast<const arrow::Time32Type&>(*type).unit() != arrow::TimeUnit::MILLI) {
+        return Status::Invalid("Only millisecond TIME is supported: ", type->ToString());
+    }
+    if (!metadata || !metadata->Contains(kTimePrecision)) {
+        return 0;
+    }
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::string precision, metadata->Get(kTimePrecision));
+    if (precision.size() != 1 || precision[0] < '0' || precision[0] > '9') {
+        return Status::Invalid("Invalid TIME precision metadata: ", precision);
+    }
+    return precision[0] - '0';
+}
+
 DataType::DataType(const std::shared_ptr<arrow::DataType>& type, bool nullable,
                    const std::shared_ptr<const arrow::KeyValueMetadata>& metadata)
     : type_(type), nullable_(nullable), metadata_(metadata) {}
@@ -112,6 +133,13 @@ std::string DataType::DataTypeToString(const std::shared_ptr<arrow::DataType>& t
             return "BYTES";
         case arrow::Type::type::DATE32:
             return "DATE";
+        case arrow::Type::type::TIME32: {
+            auto precision = GetTimePrecision(type, metadata_);
+            if (!precision.ok()) {
+                throw std::invalid_argument(precision.status().ToString());
+            }
+            return fmt::format("TIME({})", precision.value());
+        }
         case arrow::Type::type::DECIMAL128: {
             auto status = DecimalUtils::CheckDecimalType(*type);
             if (!status.ok()) {

src/paimon/common/types/data_type.h

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -29,6 +30,7 @@
 
 namespace arrow {
 class DataType;
+class Field;
 class TimestampType;
 class KeyValueMetadata;
 }  // namespace arrow
@@ -37,6 +39,12 @@ namespace paimon {
 
 class DataType : public Jsonizable<DataType> {
  public:
+    static constexpr char kTimePrecision[] = "paimon.time.precision";
+
+    // Arrow carries milliseconds, while metadata preserves the declared TIME precision.
+    // Arrow-only schemas use Paimon's default precision of zero.
+    static Result<int32_t> GetTimePrecision(const arrow::Field& field);
+
     static std::unique_ptr<DataType> Create(
         const std::shared_ptr<arrow::DataType>& type, bool nullable,
         const std::shared_ptr<const arrow::KeyValueMetadata>& metadata);
@@ -58,6 +66,10 @@ class DataType : public Jsonizable<DataType> {
     std::shared_ptr<const arrow::KeyValueMetadata> metadata_;
 
  private:
+    static Result<int32_t> GetTimePrecision(
+        const std::shared_ptr<arrow::DataType>& type,
+        const std::shared_ptr<const arrow::KeyValueMetadata>& metadata);
+
     std::string TimestampToString(const std::shared_ptr<arrow::TimestampType>& type) const;
     std::string DataTypeToString(const std::shared_ptr<arrow::DataType>& type) const;
 };

src/paimon/common/types/data_type_json_parser.cpp

@@ -33,6 +33,7 @@
 #include "paimon/common/data/blob_utils.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/types/data_type.h"
 #include "paimon/common/types/vector_type.h"
 #include "paimon/common/utils/date_time_utils.h"
 #include "paimon/common/utils/string_utils.h"
@@ -84,11 +85,11 @@ struct Token {
     std::string value;
 };
 
-// Extension type attributes of a parsed atomic type. BLOB and VARIANT parse to plain arrow
-// types (large_binary / struct) and need field-level metadata markers applied by the caller.
+// Logical attributes not represented by the Arrow type are attached to the parsed field.
 struct AtomicTypeAttributes {
     bool is_blob = false;
     bool is_variant = false;
+    std::optional<int32_t> time_precision;
 };
 
 // nullptr is returned in the case of parsing failed
@@ -249,6 +250,7 @@ class TokenParser {
     Result<std::shared_ptr<arrow::DataType>> ParseStringType();
     Result<std::shared_ptr<arrow::DataType>> ParseDecimalType();
     Result<std::shared_ptr<arrow::DataType>> ParseDoubleType();
+    Result<std::shared_ptr<arrow::DataType>> ParseTimeType(AtomicTypeAttributes* attributes);
     Result<std::shared_ptr<arrow::DataType>> ParseTimestampType();
     Result<std::shared_ptr<arrow::DataType>> ParseTimestampLtzType();
     Result<std::shared_ptr<arrow::DataType>> ParseVectorType();
@@ -522,6 +524,8 @@ Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTypeByKeyword(
             return ParseDoubleType();
         case Keyword::DATE:
             return arrow::date32();
+        case Keyword::TIME:
+            return ParseTimeType(attributes);
         case Keyword::TIMESTAMP:
             return ParseTimestampType();
         case Keyword::TIMESTAMP_LTZ:
@@ -581,6 +585,22 @@ Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseDoubleType() {
     return arrow::float64();
 }
 
+Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTimeType(
+    AtomicTypeAttributes* attributes) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t precision, ParseOptionalPrecision(/*default_precision=*/0));
+    if (precision < 0 || precision > 9) {
+        return Status::Invalid("Time precision must be between 0 and 9 (both inclusive)");
+    }
+    if (HasNextToken({Keyword::WITHOUT})) {
+        PAIMON_RETURN_NOT_OK(NextToken(Keyword::WITHOUT));
+        PAIMON_RETURN_NOT_OK(NextToken(Keyword::TIME));
+        PAIMON_RETURN_NOT_OK(NextToken(Keyword::ZONE));
+    }
+    // Paimon stores TIME as milliseconds of the day, including PyPaimon's TIME(0).
+    attributes->time_precision = precision;
+    return arrow::time32(arrow::TimeUnit::MILLI);
+}
+
 Result<std::shared_ptr<arrow::DataType>> TokenParser::ParseTimestampType() {
     PAIMON_ASSIGN_OR_RAISE(int32_t precision, ParseOptionalPrecision(Timestamp::DEFAULT_PRECISION));
     bool with_timezone = false;
@@ -743,6 +763,11 @@ Result<std::shared_ptr<arrow::Field>> DataTypeJsonParser::ParseAtomicTypeField(
         return BlobUtils::ToArrowField(name, nullable);
     } else if (attributes.is_variant) {
         return VariantTypeUtils::ToArrowField(name, nullable);
+    } else if (attributes.time_precision) {
+        return arrow::field(
+            name, type, nullable,
+            arrow::KeyValueMetadata::Make({DataType::kTimePrecision},
+                                          {std::to_string(attributes.time_precision.value())}));
     } else {
         return arrow::field(name, type, nullable);
     }

src/paimon/common/types/data_type_json_parser_test.cpp

@@ -22,9 +22,11 @@
 #include <utility>
 #include <vector>
 
+#include "fmt/format.h"
 #include "gtest/gtest.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/types/data_type.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/common/utils/date_time_utils.h"
 #include "paimon/status.h"
@@ -356,4 +358,35 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) {
     }
 }
 
+TEST(DataTypeJsonParserTest, ParseTimeType) {
+    std::vector<std::string> types = {"TIME", "TIME WITHOUT TIME ZONE"};
+    for (int32_t precision = 0; precision <= 9; ++precision) {
+        types.push_back(fmt::format("TIME({})", precision));
+        types.push_back(fmt::format("TIME({}) WITHOUT TIME ZONE", precision));
+    }
+    for (const auto& type : types) {
+        for (bool nullable : {true, false}) {
+            std::string type_str = nullable ? type : type + " NOT NULL";
+            SCOPED_TRACE(type_str);
+            rapidjson::Document doc;
+            rapidjson::Value value(type_str.data(), doc.GetAllocator());
+            ASSERT_OK_AND_ASSIGN(auto field, DataTypeJsonParser::ParseType("time", value));
+            ASSERT_TRUE(field->type()->Equals(arrow::time32(arrow::TimeUnit::MILLI)));
+            ASSERT_EQ(field->nullable(), nullable);
+            auto logical_type = DataType::Create(field->type(), nullable, field->metadata());
+            ASSERT_OK_AND_ASSIGN(auto serialized, logical_type->ToJsonString());
+            int32_t precision = type.find('(') == std::string::npos ? 0 : type[5] - '0';
+            ASSERT_EQ(serialized,
+                      fmt::format("\"TIME({}){}\"", precision, nullable ? "" : " NOT NULL"));
+        }
+    }
+    for (const char* type : {"TIME(-1)", "TIME(10)", "TIME(2147483648)", "TIME()", "TIME(3, 0)",
+                             "TIME WITH TIME ZONE", "TIME(3) WITHOUT TIME"}) {
+        SCOPED_TRACE(type);
+        rapidjson::Document doc;
+        rapidjson::Value value(type, doc.GetAllocator());
+        ASSERT_NOK(DataTypeJsonParser::ParseType("time", value));
+    }
+}
+
 }  // namespace paimon::test

src/paimon/core/schema/arrow_schema_validator.cpp

@@ -28,6 +28,7 @@
 #include "paimon/common/data/variant/variant_access_utils.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/types/data_type.h"
 #include "paimon/common/types/vector_type.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/common/utils/decimal_utils.h"
@@ -123,6 +124,7 @@ Status ArrowSchemaValidator::ValidateDataTypeWithFieldId(
         case arrow::Type::type::DATE32:
         case arrow::Type::type::DECIMAL128:
         case arrow::Type::type::TIMESTAMP:
+        case arrow::Type::type::TIME32:
             return Status::OK();
         case arrow::Type::type::LIST: {
             const auto& value_field = checked_cast<arrow::BaseListType*>(type.get())->value_field();
@@ -208,6 +210,9 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr<arrow::Field>&
         case arrow::Type::type::DATE32:
         case arrow::Type::type::TIMESTAMP:
             break;
+        case arrow::Type::type::TIME32:
+            PAIMON_RETURN_NOT_OK(DataType::GetTimePrecision(*field));
+            break;
         case arrow::Type::type::DECIMAL128:
             PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*field->type()));
             break;

src/paimon/core/schema/arrow_schema_validator_test.cpp

@@ -29,6 +29,7 @@
 #include "paimon/common/data/variant/variant_defs.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/types/data_field.h"
+#include "paimon/common/types/data_type.h"
 #include "paimon/common/utils/date_time_utils.h"
 #include "paimon/testing/utils/testharness.h"
 
@@ -76,6 +77,39 @@ TEST(ArrowSchemaValidatorTest, TestVectorElementType) {
     }
 }
 
+TEST(ArrowSchemaValidatorTest, TestTimeType) {
+    for (const auto& type :
+         {arrow::time32(arrow::TimeUnit::MILLI), arrow::time32(arrow::TimeUnit::SECOND),
+          arrow::time64(arrow::TimeUnit::MICRO), arrow::time64(arrow::TimeUnit::NANO)}) {
+        SCOPED_TRACE(type->ToString());
+        for (const auto& field_type : {type, arrow::list(type)}) {
+            auto schema = DataField::ConvertDataFieldsToArrowSchema(
+                {DataField(0, arrow::field("time", field_type))});
+            if (type->Equals(arrow::time32(arrow::TimeUnit::MILLI))) {
+                ASSERT_OK(ArrowSchemaValidator::ValidateSchema(*schema));
+                ASSERT_OK(ArrowSchemaValidator::ValidateSchemaWithFieldId(*schema));
+            } else {
+                ASSERT_NOK(ArrowSchemaValidator::ValidateSchema(*schema));
+                ASSERT_NOK(ArrowSchemaValidator::ValidateSchemaWithFieldId(*schema));
+            }
+        }
+    }
+}
+
+TEST(ArrowSchemaValidatorTest, TestInvalidTimePrecision) {
+    for (const char* precision : {"", "-1", "10", "3x", "1.5", "2147483648"}) {
+        SCOPED_TRACE(precision);
+        auto field =
+            arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI), true,
+                         arrow::KeyValueMetadata::Make({DataType::kTimePrecision}, {precision}));
+        ASSERT_NOK_WITH_MSG(ArrowSchemaValidator::ValidateSchema(*arrow::schema({field})),
+                            "Invalid TIME precision metadata");
+        ASSERT_NOK_WITH_MSG(DataField(0, field).ToJsonString(), "Invalid TIME precision metadata");
+    }
+    auto seconds = DataType::Create(arrow::time32(arrow::TimeUnit::SECOND), true, nullptr);
+    ASSERT_NOK_WITH_MSG(seconds->ToJsonString(), "Only millisecond TIME is supported");
+}
+
 TEST(ArrowSchemaValidatorTest, TestValidateNoRedundantFields) {
     auto col1_field = arrow::field("col1", arrow::int64());
     auto col2_field = arrow::field("col2", arrow::int32());

src/paimon/core/schema/schema_manager_test.cpp

@@ -25,12 +25,76 @@
 
 #include "arrow/type.h"
 #include "gtest/gtest.h"
+#include "paimon/common/types/data_type.h"
+#include "paimon/common/types/data_type_json_parser.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
 
+TEST(SchemaManagerTest, TimePrecisionRoundTrip) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fs = std::make_shared<LocalFileSystem>();
+    SchemaManager manager(fs, dir->Str());
+    arrow::FieldVector fields;
+    for (int32_t precision = 0; precision <= 9; ++precision) {
+        for (bool nullable : {true, false}) {
+            std::string name = "t" + std::to_string(fields.size());
+            std::string type =
+                "TIME(" + std::to_string(precision) + ")" + (nullable ? "" : " NOT NULL");
+            rapidjson::Document doc;
+            rapidjson::Value value(type.c_str(), doc.GetAllocator());
+            ASSERT_OK_AND_ASSIGN(auto field, DataTypeJsonParser::ParseType(name, value));
+            fields.push_back(field);
+        }
+    }
+    fields.push_back(arrow::field("default_time", arrow::time32(arrow::TimeUnit::MILLI)));
+    fields.push_back(arrow::field("times", arrow::list(fields[6]->WithName("item"))));
+    fields.push_back(
+        arrow::field("mapping", std::make_shared<arrow::MapType>(fields[13]->WithName("key"),
+                                                                 fields[18]->WithName("value"))));
+    fields.push_back(arrow::field("nested", arrow::struct_({fields[6], fields[13], fields[18]})));
+    ASSERT_OK_AND_ASSIGN(auto created,
+                         manager.CreateTable(arrow::schema(fields), {}, {},
+                                             {{"file.format", "parquet"}, {"bucket", "-1"}}));
+    ASSERT_OK_AND_ASSIGN(auto serialized, created->ToJsonString());
+    SchemaManager reloaded_manager(fs, dir->Str());
+    ASSERT_OK_AND_ASSIGN(auto reloaded, reloaded_manager.ReadSchema(0));
+    ASSERT_OK_AND_ASSIGN(auto reserialized, reloaded->ToJsonString());
+    ASSERT_EQ(serialized, reserialized);
+    const auto& restored_fields = reloaded->Fields();
+    for (int32_t i = 0; i < 20; ++i) {
+        ASSERT_OK_AND_ASSIGN(auto precision,
+                             DataType::GetTimePrecision(*restored_fields[i].ArrowField()));
+        ASSERT_EQ(precision, i / 2);
+        ASSERT_EQ(restored_fields[i].ArrowField()->nullable(), i % 2 == 0);
+    }
+    ASSERT_OK_AND_ASSIGN(auto default_precision,
+                         DataType::GetTimePrecision(*restored_fields[20].ArrowField()));
+    ASSERT_EQ(default_precision, 0);
+    for (int32_t i = 21; i < 24; ++i) {
+        SCOPED_TRACE(i);
+        ASSERT_TRUE(DataField::ConvertDataFieldToArrowField(created->Fields()[i])
+                        ->Equals(DataField::ConvertDataFieldToArrowField(restored_fields[i]),
+                                 /*check_metadata=*/true));
+    }
+}
+
+TEST(SchemaManagerTest, RejectTimePartitionKey) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    SchemaManager manager(std::make_shared<LocalFileSystem>(), dir->Str());
+    auto schema = arrow::schema({arrow::field("id", arrow::int32()),
+                                 arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI))});
+    ASSERT_NOK_WITH_MSG(
+        manager.CreateTable(schema, {"time"}, {}, {{"file.format", "parquet"}, {"bucket", "-1"}}),
+        "partition field time cannot be TIME");
+    ASSERT_OK_AND_ASSIGN(auto latest, manager.Latest());
+    ASSERT_FALSE(latest.has_value());
+}
+
 TEST(SchemaManagerTest, ConcurrentHistoricalSchemaReads) {
     SchemaManager manager(
         std::make_shared<LocalFileSystem>(),

src/paimon/core/schema/schema_validation.cpp

@@ -397,6 +397,10 @@ Status SchemaValidation::ValidateNotContainSpecificType(
         auto it = fields_map.find(field_name);
         if (it != fields_map.end()) {
             auto field = it->second;
+            if (field->type()->id() == arrow::Type::TIME32) {
+                return Status::Invalid(
+                    fmt::format("partition field {} cannot be TIME", field_name));
+            }
             if (IsComplexType(field)) {
                 return Status::Invalid(
                     fmt::format("partition field {} cannot be TIMESTAMP/DECIMAL/BLOB", field_name));