From fc6e39a166a18d054f612d23984713c36b288cad Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 21 Sep 2026 15:14:03 +0800 Subject: [PATCH 1/3] fix(read): support Parquet TIME columns Parse TIME schemas as millisecond Arrow values and validate time32[ms]. Skip cast lookup for identical Arrow types so existing Parquet TIME columns can be read without losing milliseconds. Add parser, schema validation, field mapping and Parquet read regression tests. Signed-off-by: QuakeWang --- .../common/types/data_type_json_parser.cpp | 17 ++++ .../types/data_type_json_parser_test.cpp | 27 ++++++ .../core/schema/arrow_schema_validator.cpp | 8 ++ .../schema/arrow_schema_validator_test.cpp | 19 ++++ .../core/schema/schema_validation_test.cpp | 13 ++- src/paimon/core/utils/field_mapping.cpp | 9 +- src/paimon/core/utils/field_mapping_test.cpp | 18 ++++ test/inte/read_inte_test.cpp | 92 +++++++++++++++++++ 8 files changed, 193 insertions(+), 10 deletions(-) diff --git a/src/paimon/common/types/data_type_json_parser.cpp b/src/paimon/common/types/data_type_json_parser.cpp index d04e5ade5..f2dbe41d2 100644 --- a/src/paimon/common/types/data_type_json_parser.cpp +++ b/src/paimon/common/types/data_type_json_parser.cpp @@ -250,6 +250,7 @@ class TokenParser { Result> ParseStringType(); Result> ParseDecimalType(); Result> ParseDoubleType(); + Result> ParseTimeType(); Result> ParseTimestampType(); Result> ParseTimestampLtzType(); Result> ParseVectorType(); @@ -523,6 +524,8 @@ Result> TokenParser::ParseTypeByKeyword( return ParseDoubleType(); case Keyword::DATE: return arrow::date32(); + case Keyword::TIME: + return ParseTimeType(); case Keyword::TIMESTAMP: return ParseTimestampType(); case Keyword::TIMESTAMP_LTZ: @@ -582,6 +585,20 @@ Result> TokenParser::ParseDoubleType() { return arrow::float64(); } +Result> TokenParser::ParseTimeType() { + 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). + return arrow::time32(arrow::TimeUnit::MILLI); +} + Result> TokenParser::ParseTimestampType() { PAIMON_ASSIGN_OR_RAISE(int32_t precision, ParseOptionalPrecision(Timestamp::DEFAULT_PRECISION)); bool with_timezone = false; diff --git a/src/paimon/common/types/data_type_json_parser_test.cpp b/src/paimon/common/types/data_type_json_parser_test.cpp index e5dfbc21e..8cf9c902c 100644 --- a/src/paimon/common/types/data_type_json_parser_test.cpp +++ b/src/paimon/common/types/data_type_json_parser_test.cpp @@ -22,6 +22,7 @@ #include #include +#include "fmt/format.h" #include "gtest/gtest.h" #include "paimon/common/data/variant/variant_type_utils.h" #include "paimon/common/utils/checked_cast.h" @@ -229,4 +230,30 @@ TEST(DataTypeJsonParserTest, ParseTypeAtomicTypeSuccess) { } } +TEST(DataTypeJsonParserTest, ParseTimeType) { + std::vector 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); + } + } + 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 diff --git a/src/paimon/core/schema/arrow_schema_validator.cpp b/src/paimon/core/schema/arrow_schema_validator.cpp index c008d222d..850bf2541 100644 --- a/src/paimon/core/schema/arrow_schema_validator.cpp +++ b/src/paimon/core/schema/arrow_schema_validator.cpp @@ -123,6 +123,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(type.get())->value_field(); @@ -208,6 +209,13 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr& case arrow::Type::type::DATE32: case arrow::Type::type::TIMESTAMP: break; + case arrow::Type::type::TIME32: + if (checked_cast(*field->type()).unit() != + arrow::TimeUnit::MILLI) { + return Status::Invalid("Only millisecond TIME is supported: ", + field->type()->ToString()); + } + break; case arrow::Type::type::DECIMAL128: PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*field->type())); break; diff --git a/src/paimon/core/schema/arrow_schema_validator_test.cpp b/src/paimon/core/schema/arrow_schema_validator_test.cpp index 283b61ec2..e1d6bbf1b 100644 --- a/src/paimon/core/schema/arrow_schema_validator_test.cpp +++ b/src/paimon/core/schema/arrow_schema_validator_test.cpp @@ -76,6 +76,25 @@ 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, TestValidateNoRedundantFields) { auto col1_field = arrow::field("col1", arrow::int64()); auto col2_field = arrow::field("col2", arrow::int32()); diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp index 547289877..21525e7b9 100644 --- a/src/paimon/core/schema/schema_validation_test.cpp +++ b/src/paimon/core/schema/schema_validation_test.cpp @@ -289,9 +289,13 @@ 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 expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP", - "type VARIANT"}; + std::vector expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP", "type VARIANT", + "type time32", "type time32"}; for (size_t i = 0; i < unsupported_fields.size(); ++i) { ASSERT_OK_AND_ASSIGN( table_schema, @@ -303,14 +307,13 @@ TEST(SchemaValidationTest, TestLanceDataTypes) { for (const auto& field : arrow::FieldVector{ arrow::field("time_seconds", arrow::time32(arrow::TimeUnit::SECOND)), - 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)))}))}) { + "values", arrow::list(arrow::time32(arrow::TimeUnit::SECOND)))}))}) { ASSERT_NOK_WITH_MSG( TableSchema::Create(/*schema_id=*/0, arrow::schema({field}), /*partition_keys=*/{}, /*primary_keys=*/{}, options), - "Unknown or unsupported arrow type: time32"); + "Only millisecond TIME is supported"); } for (const auto& [option_key, option_value] : std::vector>{ diff --git a/src/paimon/core/utils/field_mapping.cpp b/src/paimon/core/utils/field_mapping.cpp index be7287dd6..61a40365b 100644 --- a/src/paimon/core/utils/field_mapping.cpp +++ b/src/paimon/core/utils/field_mapping.cpp @@ -177,11 +177,6 @@ Result>> FieldMappingBuilder::CreateDa std::vector> cast_executors; cast_executors.reserve(read_fields.size()); for (size_t i = 0; i < read_fields.size(); i++) { - PAIMON_ASSIGN_OR_RAISE(FieldType read_type, - FieldTypeUtils::ConvertToFieldType(read_fields[i].Type()->id())); - PAIMON_ASSIGN_OR_RAISE(FieldType data_type, - FieldTypeUtils::ConvertToFieldType(data_fields[i].Type()->id())); - if (!read_fields[i].Type()->Equals(data_fields[i].Type())) { auto read_type_id = read_fields[i].Type()->id(); if (read_type_id == arrow::Type::STRUCT || read_type_id == arrow::Type::LIST || @@ -191,6 +186,10 @@ Result>> FieldMappingBuilder::CreateDa cast_executors.push_back(nullptr); continue; } + PAIMON_ASSIGN_OR_RAISE(FieldType read_type, + FieldTypeUtils::ConvertToFieldType(read_fields[i].Type()->id())); + PAIMON_ASSIGN_OR_RAISE(FieldType data_type, + FieldTypeUtils::ConvertToFieldType(data_fields[i].Type()->id())); auto executor_factory = CastExecutorFactory::GetCastExecutorFactory(); auto cast_executor = executor_factory->GetCastExecutor(/*src=*/data_type, /*target=*/read_type); diff --git a/src/paimon/core/utils/field_mapping_test.cpp b/src/paimon/core/utils/field_mapping_test.cpp index 1ba674cd7..a970a31ea 100644 --- a/src/paimon/core/utils/field_mapping_test.cpp +++ b/src/paimon/core/utils/field_mapping_test.cpp @@ -677,4 +677,22 @@ TEST_F(FieldMappingTest, TestMapSelectedKeysMetadataPropagatedToDataSchema) { ASSERT_FALSE(custom_metadata_result.ok()); } +TEST_F(FieldMappingTest, TestTimeWithoutCast) { + std::vector fields = { + DataField(0, arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI)))}; + auto schema = DataField::ConvertDataFieldsToArrowSchema(fields); + ASSERT_OK_AND_ASSIGN(auto builder, FieldMappingBuilder::Create(schema, /*partition_keys=*/{}, + /*predicate=*/nullptr)); + ASSERT_OK_AND_ASSIGN(auto mapping, builder->CreateFieldMapping(fields)); + const auto& info = mapping->non_partition_info; + ASSERT_EQ(info.non_partition_read_schema, fields); + ASSERT_EQ(info.non_partition_data_schema, fields); + ASSERT_EQ(info.cast_executors.size(), 1); + ASSERT_EQ(info.cast_executors[0], nullptr); + + std::vector int_fields = {DataField(0, arrow::field("time", arrow::int32()))}; + ASSERT_NOK(FieldMappingBuilder::CreateDataCastExecutors(fields, int_fields)); + ASSERT_NOK(FieldMappingBuilder::CreateDataCastExecutors(int_fields, fields)); +} + } // namespace paimon::test diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 284cd362f..859f09c16 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -34,6 +34,7 @@ #include "arrow/api.h" #include "arrow/array/array_base.h" #include "arrow/c/abi.h" +#include "arrow/c/bridge.h" #include "arrow/ipc/json_simple.h" #include "gtest/gtest.h" #include "paimon/catalog/catalog.h" @@ -64,6 +65,9 @@ #include "paimon/defs.h" #include "paimon/file_store_commit.h" #include "paimon/file_store_write.h" +#include "paimon/format/file_format.h" +#include "paimon/format/file_format_factory.h" +#include "paimon/format/format_writer.h" #include "paimon/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" @@ -88,6 +92,94 @@ namespace paimon::test { +TEST(TimeReadInteTest, TestParquetTimeColumns) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto fs = std::make_shared(); + std::string table_path = dir->Str(); + std::string bucket_path = PathUtil::JoinPath(table_path, "bucket-0"); + ASSERT_OK(fs->Mkdirs(bucket_path)); + ASSERT_OK(fs->Mkdirs(PathUtil::JoinPath(table_path, "schema"))); + // PyPaimon writes millisecond values with a TIME(0) table schema. + ASSERT_OK(fs->WriteFile(PathUtil::JoinPath(table_path, "schema/schema-0"), R"json({ + "version": 3, "id": 0, + "fields": [{"id": 0, "name": "t0", "type": "TIME(0)"}, + {"id": 1, "name": "t3", "type": "TIME(3)"}], + "highestFieldId": 1, "partitionKeys": [], "primaryKeys": [], + "options": {"file.format": "parquet", "bucket": "-1"}, "timeMillis": 0 + })json", + /*overwrite=*/false)); + + auto time_type = arrow::time32(arrow::TimeUnit::MILLI); + auto file_schema = + arrow::schema({arrow::field("t0", time_type), arrow::field("t3", time_type)}); + arrow::Time32Builder time_builder(time_type, arrow::default_memory_pool()); + ASSERT_TRUE(time_builder.AppendValues({0, 123, 86399999}).ok()); + ASSERT_TRUE(time_builder.AppendNull().ok()); + std::shared_ptr times; + ASSERT_TRUE(time_builder.Finish(×).ok()); + auto rows_result = arrow::StructArray::Make({times, times}, file_schema->fields()); + ASSERT_TRUE(rows_result.ok()); + auto rows = rows_result.ValueOrDie(); + + // Generate only the data file; this test does not require Paimon table-write support. + ASSERT_OK_AND_ASSIGN(auto format, FileFormatFactory::Get("parquet", {})); + ArrowSchema c_schema; + ASSERT_TRUE(arrow::ExportSchema(*file_schema, &c_schema).ok()); + ASSERT_OK_AND_ASSIGN(auto writer_builder, format->CreateWriterBuilder(&c_schema, 4)); + std::string file_path = PathUtil::JoinPath(bucket_path, "data-time.parquet"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr out, + fs->Create(file_path, /*overwrite=*/false)); + ASSERT_OK_AND_ASSIGN( + auto writer, writer_builder->WithMemoryPool(GetDefaultPool())->Build(out, "uncompressed")); + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*rows, &c_array).ok()); + ASSERT_OK(writer->AddBatch(&c_array)); + ASSERT_OK(writer->Finish()); + ASSERT_OK(out->Close()); + + ASSERT_OK_AND_ASSIGN(auto file_status, fs->GetFileStatus(file_path)); + ASSERT_OK_AND_ASSIGN(auto meta, + DataFileMeta::ForAppend("data-time.parquet", file_status.GetLen(), 4, + SimpleStats::EmptyStats(), 0, 3, + /*schema_id=*/0, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, + /*external_path=*/std::nullopt, + /*first_row_id=*/std::nullopt, + /*write_cols=*/std::nullopt)); + DataSplitImpl::Builder split_builder(BinaryRow::EmptyRow(), 0, bucket_path, {meta}); + ASSERT_OK_AND_ASSIGN(auto split, split_builder.WithSnapshot(1).RawConvertible(true).Build()); + auto row_kinds = arrow::MakeArrayFromScalar(arrow::Int8Scalar(0), times->length()); + ASSERT_TRUE(row_kinds.ok()); + for (int32_t batch_size : {1, 3}) { + for (bool projected : {false, true}) { + ReadContextBuilder context_builder(table_path); + context_builder.AddOption("read.batch-size", std::to_string(batch_size)); + if (projected) { + context_builder.SetReadFieldNames({"t0"}); + } + ASSERT_OK_AND_ASSIGN(auto context, context_builder.Finish()); + ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(context))); + ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(split)); + ASSERT_OK_AND_ASSIGN(auto result, + ReadResultCollector::CollectResult(std::move(reader))); + ASSERT_TRUE(result); + arrow::ArrayVector expected_columns = {row_kinds.ValueOrDie(), times}; + arrow::FieldVector expected_fields = {SpecialFields::ValueKind().ArrowField(), + file_schema->field(0)}; + if (!projected) { + expected_columns.push_back(times); + expected_fields.push_back(file_schema->field(1)); + } + auto expected = arrow::StructArray::Make(expected_columns, expected_fields); + ASSERT_TRUE(expected.ok()); + ASSERT_TRUE( + result->Equals(std::make_shared(expected.ValueOrDie()))) + << result->ToString(); + } + } +} + struct TestParam { bool enable_prefetch; std::string enable_adaptive_prefetch_strategy; From c23c2679364ec0e0fd436d0eb3254a558f45e9aa Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Mon, 21 Sep 2026 22:53:16 +0800 Subject: [PATCH 2/3] fix(schema): preserve TIME precision during serialization Store declared TIME precision in Arrow field metadata and use it when serializing schemas. Validate precision metadata and default Arrow-only TIME fields to precision zero. Cover schema persistence and nested precision round trips. Verify Parquet TIME values across Java, Python and Rust fixtures and update ORC and Avro unsupported-type expectations. Signed-off-by: QuakeWang --- src/paimon/common/types/data_type.cpp | 24 +++++++++ src/paimon/common/types/data_type.h | 9 ++++ .../common/types/data_type_json_parser.cpp | 18 +++++-- .../types/data_type_json_parser_test.cpp | 6 +++ .../core/schema/arrow_schema_validator.cpp | 8 ++- .../schema/arrow_schema_validator_test.cpp | 15 ++++++ .../core/schema/schema_manager_test.cpp | 53 +++++++++++++++++++ test/inte/paimon_read_compat_inte_test.cpp | 37 +++++++++++-- 8 files changed, 155 insertions(+), 15 deletions(-) diff --git a/src/paimon/common/types/data_type.cpp b/src/paimon/common/types/data_type.cpp index 2bf5d73cb..8fa932a25 100644 --- a/src/paimon/common/types/data_type.cpp +++ b/src/paimon/common/types/data_type.cpp @@ -41,6 +41,23 @@ namespace paimon { +Result DataType::GetTimePrecision( + const std::shared_ptr& type, + const std::shared_ptr& metadata) { + if (type->id() != arrow::Type::TIME32 || + checked_cast(*type).unit() != arrow::TimeUnit::MILLI) { + return Status::Invalid("Only millisecond TIME is supported: ", type->ToString()); + } + if (!metadata || !metadata->Contains(TIME_PRECISION)) { + return 0; + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::string precision, metadata->Get(TIME_PRECISION)); + 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& type, bool nullable, const std::shared_ptr& metadata) : type_(type), nullable_(nullable), metadata_(metadata) {} @@ -112,6 +129,13 @@ std::string DataType::DataTypeToString(const std::shared_ptr& 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()) { diff --git a/src/paimon/common/types/data_type.h b/src/paimon/common/types/data_type.h index 173960b4c..7e7d62e01 100644 --- a/src/paimon/common/types/data_type.h +++ b/src/paimon/common/types/data_type.h @@ -19,6 +19,7 @@ #pragma once +#include #include #include @@ -37,6 +38,14 @@ namespace paimon { class DataType : public Jsonizable { public: + static constexpr char TIME_PRECISION[] = "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 GetTimePrecision( + const std::shared_ptr& type, + const std::shared_ptr& metadata); + static std::unique_ptr Create( const std::shared_ptr& type, bool nullable, const std::shared_ptr& metadata); diff --git a/src/paimon/common/types/data_type_json_parser.cpp b/src/paimon/common/types/data_type_json_parser.cpp index f2dbe41d2..2845196a3 100644 --- a/src/paimon/common/types/data_type_json_parser.cpp +++ b/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/rapidjson_util.h" @@ -85,11 +86,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 time_precision; }; // nullptr is returned in the case of parsing failed @@ -250,7 +251,7 @@ class TokenParser { Result> ParseStringType(); Result> ParseDecimalType(); Result> ParseDoubleType(); - Result> ParseTimeType(); + Result> ParseTimeType(AtomicTypeAttributes* attributes); Result> ParseTimestampType(); Result> ParseTimestampLtzType(); Result> ParseVectorType(); @@ -525,7 +526,7 @@ Result> TokenParser::ParseTypeByKeyword( case Keyword::DATE: return arrow::date32(); case Keyword::TIME: - return ParseTimeType(); + return ParseTimeType(attributes); case Keyword::TIMESTAMP: return ParseTimestampType(); case Keyword::TIMESTAMP_LTZ: @@ -585,7 +586,8 @@ Result> TokenParser::ParseDoubleType() { return arrow::float64(); } -Result> TokenParser::ParseTimeType() { +Result> 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)"); @@ -596,6 +598,7 @@ Result> TokenParser::ParseTimeType() { 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); } @@ -685,6 +688,11 @@ Result> 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::TIME_PRECISION}, + {std::to_string(attributes.time_precision.value())})); } else { return arrow::field(name, type, nullable); } diff --git a/src/paimon/common/types/data_type_json_parser_test.cpp b/src/paimon/common/types/data_type_json_parser_test.cpp index 8cf9c902c..1b1a9572a 100644 --- a/src/paimon/common/types/data_type_json_parser_test.cpp +++ b/src/paimon/common/types/data_type_json_parser_test.cpp @@ -25,6 +25,7 @@ #include "fmt/format.h" #include "gtest/gtest.h" #include "paimon/common/data/variant/variant_type_utils.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" @@ -245,6 +246,11 @@ TEST(DataTypeJsonParserTest, ParseTimeType) { 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)", diff --git a/src/paimon/core/schema/arrow_schema_validator.cpp b/src/paimon/core/schema/arrow_schema_validator.cpp index 850bf2541..007b8e698 100644 --- a/src/paimon/core/schema/arrow_schema_validator.cpp +++ b/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" @@ -210,11 +211,8 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr& case arrow::Type::type::TIMESTAMP: break; case arrow::Type::type::TIME32: - if (checked_cast(*field->type()).unit() != - arrow::TimeUnit::MILLI) { - return Status::Invalid("Only millisecond TIME is supported: ", - field->type()->ToString()); - } + PAIMON_RETURN_NOT_OK( + DataType::GetTimePrecision(field->type(), field->metadata()).status()); break; case arrow::Type::type::DECIMAL128: PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*field->type())); diff --git a/src/paimon/core/schema/arrow_schema_validator_test.cpp b/src/paimon/core/schema/arrow_schema_validator_test.cpp index e1d6bbf1b..4ac96cd71 100644 --- a/src/paimon/core/schema/arrow_schema_validator_test.cpp +++ b/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" @@ -95,6 +96,20 @@ TEST(ArrowSchemaValidatorTest, TestTimeType) { } } +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::TIME_PRECISION}, {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()); diff --git a/src/paimon/core/schema/schema_manager_test.cpp b/src/paimon/core/schema/schema_manager_test.cpp index 0b078462b..f9d4a0a61 100644 --- a/src/paimon/core/schema/schema_manager_test.cpp +++ b/src/paimon/core/schema/schema_manager_test.cpp @@ -25,12 +25,65 @@ #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(); + 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(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].Type(), + restored_fields[i].ArrowField()->metadata())); + 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].Type(), + restored_fields[20].ArrowField()->metadata())); + 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, ConcurrentHistoricalSchemaReads) { SchemaManager manager( std::make_shared(), diff --git a/test/inte/paimon_read_compat_inte_test.cpp b/test/inte/paimon_read_compat_inte_test.cpp index 5948f0ef7..7ac4337bb 100644 --- a/test/inte/paimon_read_compat_inte_test.cpp +++ b/test/inte/paimon_read_compat_inte_test.cpp @@ -406,6 +406,23 @@ TEST_P(PaimonReadCompatInteTest, ReadsCompatibleTypeValues) { } } +TEST_P(PaimonReadCompatInteTest, ReadsTimeValues) { + const CompatibilityParam& param = GetParam(); + if (param.file_format != "parquet") { + GTEST_SKIP() << "TIME reading is only supported for Parquet"; + } + for (int32_t precision : {0, 3, 6, 9}) { + std::string field_name = "f_time_" + std::to_string(precision); + ASSERT_OK_AND_ASSIGN(auto result, + ReadTable(param, param.writer_prefix + "_time_types", {field_name})); + auto rows = GetOnlyStructChunk(result); + ASSERT_TRUE(rows); + ASSERT_EQ(rows->length(), 2); + AssertFieldEqualsJson(rows, field_name, arrow::time32(arrow::TimeUnit::MILLI), + precision == 0 ? "[45296000, null]" : "[45296123, null]"); + } +} + TEST_P(PaimonReadCompatInteTest, ReadsBlobValues) { const CompatibilityParam& param = GetParam(); @@ -566,20 +583,30 @@ std::vector UnsupportedReadParams() { const std::vector read_cases = { {"ArrayBlob", "array_blob_types", "f_array_blob", "BLOB field must be a top-level field or the direct value of a top-level MAP field"}, - {"TimePrecision0", "time_types", "f_time_0", "Unsupported type: TIME"}, - {"TimePrecision3", "time_types", "f_time_3", "Unsupported type: TIME"}, - {"TimePrecision6", "time_types", "f_time_6", "Unsupported type: TIME"}, - {"TimePrecision9", "time_types", "f_time_9", "Unsupported type: TIME"}, + {"TimePrecision0", "time_types", "f_time_0", ""}, + {"TimePrecision3", "time_types", "f_time_3", ""}, + {"TimePrecision6", "time_types", "f_time_6", ""}, + {"TimePrecision9", "time_types", "f_time_9", ""}, }; std::vector result; for (const CompatibilityParam& param : CompatibilityParams()) { for (const UnsupportedReadCase& read_case : read_cases) { + bool is_time = read_case.table_suffix == "time_types"; + if (is_time && param.file_format == "parquet") { + continue; + } // Java Avro cannot create TIME(6/9), so its negative table contains TIME(0/3) only. if (param.file_format == "avro" && (read_case.name == "TimePrecision6" || read_case.name == "TimePrecision9")) { continue; } - result.push_back({param.file_format, param.writer_prefix, read_case}); + auto expected_case = read_case; + if (is_time) { + expected_case.expected_error = param.file_format == "orc" + ? "Unknown or unsupported Arrow type: time32[ms]" + : "invalid avro logical type"; + } + result.push_back({param.file_format, param.writer_prefix, expected_case}); } } return result; From 725ec94e9da81ee83171bd74b9a1141670606534 Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Tue, 22 Sep 2026 14:12:14 +0800 Subject: [PATCH 3/3] fix(types): address TIME review feedback Accept Arrow fields in the TIME precision accessor and reject TIME partition keys during schema validation. Remove redundant read integration coverage and add partition validation tests. Signed-off-by: QuakeWang --- src/paimon/common/types/data_type.cpp | 8 +- src/paimon/common/types/data_type.h | 11 ++- .../common/types/data_type_json_parser.cpp | 2 +- .../core/schema/arrow_schema_validator.cpp | 3 +- .../schema/arrow_schema_validator_test.cpp | 2 +- .../core/schema/schema_manager_test.cpp | 21 ++++- src/paimon/core/schema/schema_validation.cpp | 4 + .../core/schema/schema_validation_test.cpp | 17 ++++ test/inte/read_inte_test.cpp | 92 ------------------- 9 files changed, 53 insertions(+), 107 deletions(-) diff --git a/src/paimon/common/types/data_type.cpp b/src/paimon/common/types/data_type.cpp index 8fa932a25..f5e562955 100644 --- a/src/paimon/common/types/data_type.cpp +++ b/src/paimon/common/types/data_type.cpp @@ -41,6 +41,10 @@ namespace paimon { +Result DataType::GetTimePrecision(const arrow::Field& field) { + return GetTimePrecision(field.type(), field.metadata()); +} + Result DataType::GetTimePrecision( const std::shared_ptr& type, const std::shared_ptr& metadata) { @@ -48,10 +52,10 @@ Result DataType::GetTimePrecision( checked_cast(*type).unit() != arrow::TimeUnit::MILLI) { return Status::Invalid("Only millisecond TIME is supported: ", type->ToString()); } - if (!metadata || !metadata->Contains(TIME_PRECISION)) { + if (!metadata || !metadata->Contains(kTimePrecision)) { return 0; } - PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::string precision, metadata->Get(TIME_PRECISION)); + 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); } diff --git a/src/paimon/common/types/data_type.h b/src/paimon/common/types/data_type.h index 7e7d62e01..1e49ce7e1 100644 --- a/src/paimon/common/types/data_type.h +++ b/src/paimon/common/types/data_type.h @@ -30,6 +30,7 @@ namespace arrow { class DataType; +class Field; class TimestampType; class KeyValueMetadata; } // namespace arrow @@ -38,13 +39,11 @@ namespace paimon { class DataType : public Jsonizable { public: - static constexpr char TIME_PRECISION[] = "paimon.time.precision"; + 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 GetTimePrecision( - const std::shared_ptr& type, - const std::shared_ptr& metadata); + static Result GetTimePrecision(const arrow::Field& field); static std::unique_ptr Create( const std::shared_ptr& type, bool nullable, @@ -67,6 +66,10 @@ class DataType : public Jsonizable { std::shared_ptr metadata_; private: + static Result GetTimePrecision( + const std::shared_ptr& type, + const std::shared_ptr& metadata); + std::string TimestampToString(const std::shared_ptr& type) const; std::string DataTypeToString(const std::shared_ptr& type) const; }; diff --git a/src/paimon/common/types/data_type_json_parser.cpp b/src/paimon/common/types/data_type_json_parser.cpp index dedd0bcc8..1416ae333 100644 --- a/src/paimon/common/types/data_type_json_parser.cpp +++ b/src/paimon/common/types/data_type_json_parser.cpp @@ -766,7 +766,7 @@ Result> DataTypeJsonParser::ParseAtomicTypeField( } else if (attributes.time_precision) { return arrow::field( name, type, nullable, - arrow::KeyValueMetadata::Make({DataType::TIME_PRECISION}, + arrow::KeyValueMetadata::Make({DataType::kTimePrecision}, {std::to_string(attributes.time_precision.value())})); } else { return arrow::field(name, type, nullable); diff --git a/src/paimon/core/schema/arrow_schema_validator.cpp b/src/paimon/core/schema/arrow_schema_validator.cpp index 007b8e698..daf364c66 100644 --- a/src/paimon/core/schema/arrow_schema_validator.cpp +++ b/src/paimon/core/schema/arrow_schema_validator.cpp @@ -211,8 +211,7 @@ Status ArrowSchemaValidator::ValidateField(const std::shared_ptr& case arrow::Type::type::TIMESTAMP: break; case arrow::Type::type::TIME32: - PAIMON_RETURN_NOT_OK( - DataType::GetTimePrecision(field->type(), field->metadata()).status()); + PAIMON_RETURN_NOT_OK(DataType::GetTimePrecision(*field)); break; case arrow::Type::type::DECIMAL128: PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*field->type())); diff --git a/src/paimon/core/schema/arrow_schema_validator_test.cpp b/src/paimon/core/schema/arrow_schema_validator_test.cpp index 4ac96cd71..6a9a55f38 100644 --- a/src/paimon/core/schema/arrow_schema_validator_test.cpp +++ b/src/paimon/core/schema/arrow_schema_validator_test.cpp @@ -101,7 +101,7 @@ TEST(ArrowSchemaValidatorTest, TestInvalidTimePrecision) { SCOPED_TRACE(precision); auto field = arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI), true, - arrow::KeyValueMetadata::Make({DataType::TIME_PRECISION}, {precision})); + 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"); diff --git a/src/paimon/core/schema/schema_manager_test.cpp b/src/paimon/core/schema/schema_manager_test.cpp index f9d4a0a61..cd4823891 100644 --- a/src/paimon/core/schema/schema_manager_test.cpp +++ b/src/paimon/core/schema/schema_manager_test.cpp @@ -66,15 +66,13 @@ TEST(SchemaManagerTest, TimePrecisionRoundTrip) { 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].Type(), - restored_fields[i].ArrowField()->metadata())); + 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].Type(), - restored_fields[20].ArrowField()->metadata())); + DataType::GetTimePrecision(*restored_fields[20].ArrowField())); ASSERT_EQ(default_precision, 0); for (int32_t i = 21; i < 24; ++i) { SCOPED_TRACE(i); @@ -84,6 +82,19 @@ TEST(SchemaManagerTest, TimePrecisionRoundTrip) { } } +TEST(SchemaManagerTest, RejectTimePartitionKey) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + SchemaManager manager(std::make_shared(), 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(), diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp index d97d73a86..2a9c33222 100644 --- a/src/paimon/core/schema/schema_validation.cpp +++ b/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)); diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp index 21525e7b9..f6c08b868 100644 --- a/src/paimon/core/schema/schema_validation_test.cpp +++ b/src/paimon/core/schema/schema_validation_test.cpp @@ -694,6 +694,23 @@ TEST(SchemaValidationTest, TestSpecificPartitionKey) { } } +TEST(SchemaValidationTest, TestTimePartitionKey) { + auto schema = arrow::schema({arrow::field("id", arrow::int32()), + arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI))}); + for (const std::vector& partition_keys : + {std::vector{}, std::vector{"time"}}) { + ASSERT_OK_AND_ASSIGN(auto table_schema, + TableSchema::Create(0, schema, partition_keys, {}, + {{"file.format", "parquet"}, {"bucket", "-1"}})); + if (partition_keys.empty()) { + ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema)); + } else { + ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), + "partition field time cannot be TIME"); + } + } +} + TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) { auto f0 = arrow::field("f0", arrow::utf8()); auto f1 = BlobUtils::ToArrowField("f1"); diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 859f09c16..284cd362f 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -34,7 +34,6 @@ #include "arrow/api.h" #include "arrow/array/array_base.h" #include "arrow/c/abi.h" -#include "arrow/c/bridge.h" #include "arrow/ipc/json_simple.h" #include "gtest/gtest.h" #include "paimon/catalog/catalog.h" @@ -65,9 +64,6 @@ #include "paimon/defs.h" #include "paimon/file_store_commit.h" #include "paimon/file_store_write.h" -#include "paimon/format/file_format.h" -#include "paimon/format/file_format_factory.h" -#include "paimon/format/format_writer.h" #include "paimon/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" @@ -92,94 +88,6 @@ namespace paimon::test { -TEST(TimeReadInteTest, TestParquetTimeColumns) { - auto dir = UniqueTestDirectory::Create(); - ASSERT_TRUE(dir); - auto fs = std::make_shared(); - std::string table_path = dir->Str(); - std::string bucket_path = PathUtil::JoinPath(table_path, "bucket-0"); - ASSERT_OK(fs->Mkdirs(bucket_path)); - ASSERT_OK(fs->Mkdirs(PathUtil::JoinPath(table_path, "schema"))); - // PyPaimon writes millisecond values with a TIME(0) table schema. - ASSERT_OK(fs->WriteFile(PathUtil::JoinPath(table_path, "schema/schema-0"), R"json({ - "version": 3, "id": 0, - "fields": [{"id": 0, "name": "t0", "type": "TIME(0)"}, - {"id": 1, "name": "t3", "type": "TIME(3)"}], - "highestFieldId": 1, "partitionKeys": [], "primaryKeys": [], - "options": {"file.format": "parquet", "bucket": "-1"}, "timeMillis": 0 - })json", - /*overwrite=*/false)); - - auto time_type = arrow::time32(arrow::TimeUnit::MILLI); - auto file_schema = - arrow::schema({arrow::field("t0", time_type), arrow::field("t3", time_type)}); - arrow::Time32Builder time_builder(time_type, arrow::default_memory_pool()); - ASSERT_TRUE(time_builder.AppendValues({0, 123, 86399999}).ok()); - ASSERT_TRUE(time_builder.AppendNull().ok()); - std::shared_ptr times; - ASSERT_TRUE(time_builder.Finish(×).ok()); - auto rows_result = arrow::StructArray::Make({times, times}, file_schema->fields()); - ASSERT_TRUE(rows_result.ok()); - auto rows = rows_result.ValueOrDie(); - - // Generate only the data file; this test does not require Paimon table-write support. - ASSERT_OK_AND_ASSIGN(auto format, FileFormatFactory::Get("parquet", {})); - ArrowSchema c_schema; - ASSERT_TRUE(arrow::ExportSchema(*file_schema, &c_schema).ok()); - ASSERT_OK_AND_ASSIGN(auto writer_builder, format->CreateWriterBuilder(&c_schema, 4)); - std::string file_path = PathUtil::JoinPath(bucket_path, "data-time.parquet"); - ASSERT_OK_AND_ASSIGN(std::shared_ptr out, - fs->Create(file_path, /*overwrite=*/false)); - ASSERT_OK_AND_ASSIGN( - auto writer, writer_builder->WithMemoryPool(GetDefaultPool())->Build(out, "uncompressed")); - ArrowArray c_array; - ASSERT_TRUE(arrow::ExportArray(*rows, &c_array).ok()); - ASSERT_OK(writer->AddBatch(&c_array)); - ASSERT_OK(writer->Finish()); - ASSERT_OK(out->Close()); - - ASSERT_OK_AND_ASSIGN(auto file_status, fs->GetFileStatus(file_path)); - ASSERT_OK_AND_ASSIGN(auto meta, - DataFileMeta::ForAppend("data-time.parquet", file_status.GetLen(), 4, - SimpleStats::EmptyStats(), 0, 3, - /*schema_id=*/0, FileSource::Append(), - /*value_stats_cols=*/std::nullopt, - /*external_path=*/std::nullopt, - /*first_row_id=*/std::nullopt, - /*write_cols=*/std::nullopt)); - DataSplitImpl::Builder split_builder(BinaryRow::EmptyRow(), 0, bucket_path, {meta}); - ASSERT_OK_AND_ASSIGN(auto split, split_builder.WithSnapshot(1).RawConvertible(true).Build()); - auto row_kinds = arrow::MakeArrayFromScalar(arrow::Int8Scalar(0), times->length()); - ASSERT_TRUE(row_kinds.ok()); - for (int32_t batch_size : {1, 3}) { - for (bool projected : {false, true}) { - ReadContextBuilder context_builder(table_path); - context_builder.AddOption("read.batch-size", std::to_string(batch_size)); - if (projected) { - context_builder.SetReadFieldNames({"t0"}); - } - ASSERT_OK_AND_ASSIGN(auto context, context_builder.Finish()); - ASSERT_OK_AND_ASSIGN(auto read, TableRead::Create(std::move(context))); - ASSERT_OK_AND_ASSIGN(auto reader, read->CreateReader(split)); - ASSERT_OK_AND_ASSIGN(auto result, - ReadResultCollector::CollectResult(std::move(reader))); - ASSERT_TRUE(result); - arrow::ArrayVector expected_columns = {row_kinds.ValueOrDie(), times}; - arrow::FieldVector expected_fields = {SpecialFields::ValueKind().ArrowField(), - file_schema->field(0)}; - if (!projected) { - expected_columns.push_back(times); - expected_fields.push_back(file_schema->field(1)); - } - auto expected = arrow::StructArray::Make(expected_columns, expected_fields); - ASSERT_TRUE(expected.ok()); - ASSERT_TRUE( - result->Equals(std::make_shared(expected.ValueOrDie()))) - << result->ToString(); - } - } -} - struct TestParam { bool enable_prefetch; std::string enable_adaptive_prefetch_strategy;