This is an automated email from the ASF dual-hosted git repository. SteNicholas pushed a commit to branch release-0.3 in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
commit 32f095f642ca6e4e04b9e2ffc3ec7128fa2d2952 Author: Zhou Hongfeng <[email protected]> AuthorDate: Tue Aug 4 18:30:23 2026 +0800 fix(parquet): allow reading nested list/map columns whose leaf types differ only in representation (#172) --- .../format/parquet/parquet_file_batch_reader.cpp | 83 +++++++++++- .../parquet/parquet_file_batch_reader_test.cpp | 150 +++++++++++++++++++++ test/inte/write_and_read_inte_test.cpp | 109 +++++++++++++++ 3 files changed, 337 insertions(+), 5 deletions(-) diff --git a/src/paimon/format/parquet/parquet_file_batch_reader.cpp b/src/paimon/format/parquet/parquet_file_batch_reader.cpp index ab142b7..c0cd40e 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader.cpp @@ -64,6 +64,67 @@ class Predicate; namespace paimon::parquet { +namespace { +// LIST/MAP do not support pruning fields from their nested value types, but physical and +// logical leaf types may still differ (for example, Parquet reports LTZ timestamps as UTC +// while Paimon exposes them in the local timezone). Compare only the nested projection shape +// here so those representation differences are handled by the normal cast path. +bool HasSameNestedProjectionShape(const std::shared_ptr<arrow::DataType>& read_type, + const std::shared_ptr<arrow::DataType>& file_type) { + const bool read_is_nested = ArrowSchemaValidator::IsNestedType(read_type); + const bool file_is_nested = ArrowSchemaValidator::IsNestedType(file_type); + if (!read_is_nested || !file_is_nested) { + if (read_is_nested || file_is_nested) { + return false; + } + // ParquetTimestampConverter explicitly supports timestamp unit and timezone + // conversion after reading. Other atomic type differences remain unsupported here. + if (read_type->id() == arrow::Type::TIMESTAMP && + file_type->id() == arrow::Type::TIMESTAMP) { + const auto& read_timestamp = static_cast<const arrow::TimestampType&>(*read_type); + const auto& file_timestamp = static_cast<const arrow::TimestampType&>(*file_type); + return read_timestamp.unit() == file_timestamp.unit() || + (file_timestamp.unit() == arrow::TimeUnit::MILLI && + read_timestamp.unit() == arrow::TimeUnit::SECOND); + } + return read_type->Equals(file_type); + } + if (read_type->id() != file_type->id()) { + return false; + } + + switch (file_type->id()) { + case arrow::Type::STRUCT: { + if (read_type->num_fields() != file_type->num_fields()) { + return false; + } + for (int32_t i = 0; i < file_type->num_fields(); ++i) { + const auto& read_child = read_type->field(i); + const auto& file_child = file_type->field(i); + if (read_child->name() != file_child->name() || + !HasSameNestedProjectionShape(read_child->type(), file_child->type())) { + return false; + } + } + return true; + } + case arrow::Type::LIST: { + const auto& read_list = static_cast<const arrow::ListType&>(*read_type); + const auto& file_list = static_cast<const arrow::ListType&>(*file_type); + return HasSameNestedProjectionShape(read_list.value_type(), file_list.value_type()); + } + case arrow::Type::MAP: { + const auto& read_map = static_cast<const arrow::MapType&>(*read_type); + const auto& file_map = static_cast<const arrow::MapType&>(*file_type); + return HasSameNestedProjectionShape(read_map.key_type(), file_map.key_type()) && + HasSameNestedProjectionShape(read_map.item_type(), file_map.item_type()); + } + default: + return false; + } +} +} // namespace + ParquetFileBatchReader::ParquetFileBatchReader( std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream, std::unique_ptr<FileReaderWrapper>&& reader, const std::map<std::string, std::string>& options, @@ -667,18 +728,30 @@ Status ParquetFileBatchReader::CollectLeafIndices(const std::shared_ptr<arrow::D SkipLeafIndices(file_child->type(), leaf_index); } } - } else if (file_type->id() == arrow::Type::LIST || file_type->id() == arrow::Type::MAP) { + } else if (file_type->id() == arrow::Type::LIST) { // Keep behavior aligned with ORC path: list/map inner partial projection // is currently unsupported and should fail-fast. - if (!read_type->Equals(file_type)) { + if (!HasSameNestedProjectionShape(read_type, file_type)) { return Status::Invalid(fmt::format( "Parquet does not support partial projection inside list/map: src {} vs target {}", file_type->ToString(), read_type->ToString())); } - for (int32_t i = 0; i < file_type->num_fields(); i++) { - PAIMON_RETURN_NOT_OK(CollectLeafIndices( - read_type->field(i)->type(), file_type->field(i)->type(), leaf_index, indices)); + const auto& read_list = static_cast<const arrow::ListType&>(*read_type); + const auto& file_list = static_cast<const arrow::ListType&>(*file_type); + PAIMON_RETURN_NOT_OK(CollectLeafIndices(read_list.value_type(), file_list.value_type(), + leaf_index, indices)); + } else if (file_type->id() == arrow::Type::MAP) { + if (!HasSameNestedProjectionShape(read_type, file_type)) { + return Status::Invalid(fmt::format( + "Parquet does not support partial projection inside list/map: src {} vs target {}", + file_type->ToString(), read_type->ToString())); } + const auto& read_map = static_cast<const arrow::MapType&>(*read_type); + const auto& file_map = static_cast<const arrow::MapType&>(*file_type); + PAIMON_RETURN_NOT_OK( + CollectLeafIndices(read_map.key_type(), file_map.key_type(), leaf_index, indices)); + PAIMON_RETURN_NOT_OK( + CollectLeafIndices(read_map.item_type(), file_map.item_type(), leaf_index, indices)); } else { // Leaf column — collect its index. indices->push_back((*leaf_index)++); diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp index dd32346..59e5a25 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp @@ -646,6 +646,156 @@ TEST_F(ParquetFileBatchReaderTest, TestReadSchemaWithMapSelectedKeysMetadata) { << "expected: " << expected_array->ToString() << "\nactual: " << result_array->ToString(); } +TEST_F(ParquetFileBatchReaderTest, TestNestedListTimestampTimezoneAndMapFieldName) { + const std::string timezone = "Asia/Shanghai"; + paimon::test::TimezoneGuard timezone_guard(timezone); + + auto write_attrs_type = + std::make_shared<arrow::MapType>(arrow::field("key", arrow::utf8(), /*nullable=*/false), + arrow::field("attrs", arrow::utf8())); + auto write_element_type = arrow::struct_({ + arrow::field("key", arrow::utf8()), + arrow::field("attrs", write_attrs_type), + arrow::field("updated_at", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), + }); + auto write_schema = arrow::schema( + {arrow::field("annotations", arrow::list(arrow::field("element", write_element_type)))}); + + const std::string data_json = R"([ + [[ ["ann-1", [["source", "model"]], "2026-07-16 12:00:00.000001"] ]], + [[ ["ann-2", [], "2026-07-16 12:00:00.000002"], + ["ann-3", null, null] ]], + [null] + ])"; + auto write_array = std::dynamic_pointer_cast<arrow::StructArray>( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(write_schema->fields()), data_json) + .ValueOrDie()); + WriteArray(file_path_, write_array, write_schema, /*write_batch_size=*/write_array->length(), + /*enable_dictionary=*/false, /*max_row_group_length=*/write_array->length()); + + auto read_element_type = arrow::struct_({ + arrow::field("key", arrow::utf8()), + arrow::field("attrs", arrow::map(arrow::utf8(), arrow::utf8())), + arrow::field("updated_at", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), + }); + auto read_schema = arrow::schema( + {arrow::field("annotations", arrow::list(arrow::field("element", read_element_type)))}); + auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(read_schema->fields()), data_json) + .ValueOrDie()); + + auto parquet_batch_reader = + PrepareParquetFileBatchReader(file_path_, read_schema, /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt, /*batch_size=*/2); + ASSERT_OK_AND_ASSIGN(auto result_array, paimon::test::ReadResultCollector::CollectResult( + parquet_batch_reader.get())); + auto expected_chunked_array = arrow::ChunkedArray::Make({expected_array}).ValueOrDie(); + ASSERT_TRUE(result_array->Equals(expected_chunked_array)) + << "expected: " << expected_chunked_array->ToString() + << "\nactual: " << result_array->ToString(); + + auto projected_element_type = arrow::struct_({ + arrow::field("key", arrow::utf8()), + arrow::field("attrs", arrow::map(arrow::utf8(), arrow::utf8())), + }); + auto partial_read_schema = arrow::schema({arrow::field( + "annotations", arrow::list(arrow::field("element", projected_element_type)))}); + auto c_partial_read_schema = std::make_unique<ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*partial_read_schema, c_partial_read_schema.get()).ok()); + ASSERT_NOK_WITH_MSG( + parquet_batch_reader->SetReadSchema(c_partial_read_schema.get(), /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt), + "Parquet does not support partial projection inside list/map"); + + auto mismatched_element_type = arrow::struct_({ + arrow::field("key", arrow::utf8()), + arrow::field("attrs", arrow::map(arrow::utf8(), arrow::utf8())), + arrow::field("updated_at", arrow::utf8()), + }); + auto mismatched_read_schema = arrow::schema({arrow::field( + "annotations", arrow::list(arrow::field("element", mismatched_element_type)))}); + auto c_mismatched_read_schema = std::make_unique<ArrowSchema>(); + ASSERT_TRUE(arrow::ExportSchema(*mismatched_read_schema, c_mismatched_read_schema.get()).ok()); + ASSERT_NOK_WITH_MSG( + parquet_batch_reader->SetReadSchema(c_mismatched_read_schema.get(), /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt), + "Parquet does not support partial projection inside list/map"); + + auto unsupported_timestamp_element_type = arrow::struct_({ + arrow::field("key", arrow::utf8()), + arrow::field("attrs", arrow::map(arrow::utf8(), arrow::utf8())), + arrow::field("updated_at", arrow::timestamp(arrow::TimeUnit::NANO, timezone)), + }); + auto unsupported_timestamp_schema = arrow::schema({arrow::field( + "annotations", arrow::list(arrow::field("element", unsupported_timestamp_element_type)))}); + auto c_unsupported_timestamp_schema = std::make_unique<ArrowSchema>(); + ASSERT_TRUE( + arrow::ExportSchema(*unsupported_timestamp_schema, c_unsupported_timestamp_schema.get()) + .ok()); + ASSERT_NOK_WITH_MSG(parquet_batch_reader->SetReadSchema(c_unsupported_timestamp_schema.get(), + /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt), + "Parquet does not support partial projection inside list/map"); +} + +TEST_F(ParquetFileBatchReaderTest, TestNestedTimestampSecondReadFromMilliFile) { + const std::string timezone = "Asia/Shanghai"; + paimon::test::TimezoneGuard timezone_guard(timezone); + + // Parquet has no second-precision timestamp, so the writer stores second timestamps as + // milliseconds. Reading them back with a second-precision schema must cast milli to second, + // including for timestamp leaves nested inside list/struct/map. + auto event_type = arrow::struct_({ + arrow::field("name", arrow::utf8()), + arrow::field("ts_sec", arrow::timestamp(arrow::TimeUnit::SECOND)), + arrow::field("ts_tz_sec", arrow::timestamp(arrow::TimeUnit::SECOND, timezone)), + }); + auto schema = arrow::schema({ + arrow::field("events", arrow::list(arrow::field("element", event_type))), + arrow::field("marks", arrow::map(arrow::utf8(), arrow::timestamp(arrow::TimeUnit::SECOND))), + }); + + const std::string data_json = R"([ + [[ ["e-1", "2026-07-16 12:00:01", "2026-07-16 12:00:02"] ], + [["begin", "2026-07-16 12:00:03"]]], + [[ ["e-2", "2026-07-16 12:00:04", null], + ["e-3", null, "2026-07-16 12:00:05"] ], []], + [[null], null] + ])"; + auto write_array = std::dynamic_pointer_cast<arrow::StructArray>( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(schema->fields()), data_json) + .ValueOrDie()); + WriteArray(file_path_, write_array, schema, /*write_batch_size=*/write_array->length(), + /*enable_dictionary=*/false, /*max_row_group_length=*/write_array->length()); + + auto parquet_batch_reader = + PrepareParquetFileBatchReader(file_path_, schema, /*predicate=*/nullptr, + /*selection_bitmap=*/std::nullopt, /*batch_size=*/2); + + // The nested second timestamps are physically stored as milliseconds in the file. + ASSERT_OK_AND_ASSIGN(auto c_file_schema, parquet_batch_reader->GetFileSchema()); + auto file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOr(nullptr); + ASSERT_TRUE(file_schema); + auto file_event_type = + static_cast<const arrow::ListType&>(*file_schema->field(0)->type()).value_type(); + ASSERT_EQ(arrow::Type::STRUCT, file_event_type->id()); + ASSERT_EQ(arrow::TimeUnit::MILLI, + static_cast<const arrow::TimestampType&>(*file_event_type->field(1)->type()).unit()); + ASSERT_EQ(arrow::TimeUnit::MILLI, + static_cast<const arrow::TimestampType&>(*file_event_type->field(2)->type()).unit()); + auto file_mark_type = + static_cast<const arrow::MapType&>(*file_schema->field(1)->type()).item_type(); + ASSERT_EQ(arrow::TimeUnit::MILLI, + static_cast<const arrow::TimestampType&>(*file_mark_type).unit()); + + ASSERT_OK_AND_ASSIGN( + std::shared_ptr<arrow::ChunkedArray> result_array, + paimon::test::ReadResultCollector::CollectResult(parquet_batch_reader.get())); + auto expected_array = arrow::ChunkedArray::Make({write_array}).ValueOrDie(); + ASSERT_TRUE(result_array->Equals(expected_array)) + << "expected: " << expected_array->ToString() << "\nactual: " << result_array->ToString(); +} + TEST_F(ParquetFileBatchReaderTest, TestGetFileSchemaWithFieldId) { std::string file_name = paimon::test::GetDataDir() + "parquet/parquet_append_table.db/parquet_append_table/bucket-0/" diff --git a/test/inte/write_and_read_inte_test.cpp b/test/inte/write_and_read_inte_test.cpp index a18fe9d..53920ba 100644 --- a/test/inte/write_and_read_inte_test.cpp +++ b/test/inte/write_and_read_inte_test.cpp @@ -806,6 +806,115 @@ TEST_P(WriteAndReadInteTest, TestPkTimestampType) { ASSERT_TRUE(success); } +/// End-to-end coverage for second-precision timestamps nested inside list/struct/map. +/// Parquet has no second-precision timestamp, so the writer stores those leaves as milli and +/// the reader has to convert milli back to second for every nested leaf. +TEST_P(WriteAndReadInteTest, TestAppendNestedTimestampSecondPrecision) { + auto [file_format, file_system] = GetParam(); + TimezoneGuard timezone_guard("Asia/Shanghai"); + auto timezone = DateTimeUtils::GetLocalTimezoneName(); + auto event_type = arrow::struct_({ + arrow::field("name", arrow::utf8()), + arrow::field("ts_sec", arrow::timestamp(arrow::TimeUnit::SECOND)), + arrow::field("ts_ltz_sec", arrow::timestamp(arrow::TimeUnit::SECOND, timezone)), + }); + arrow::FieldVector fields = { + arrow::field("events", arrow::list(arrow::field("element", event_type))), + arrow::field("marks", arrow::map(arrow::utf8(), arrow::timestamp(arrow::TimeUnit::SECOND))), + }; + std::map<std::string, std::string> options = { + {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT, file_format}, + {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "-1"}, + {Options::FILE_SYSTEM, file_system}, {"orc.timestamp-ltz.legacy.type", "false"}}; + if (file_system == "jindo") { + options = AddOptionsForJindo(options); + } + ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(test_dir_, arrow::schema(fields), + /*partition_keys=*/{}, /*primary_keys=*/{}, + options, /*is_streaming_mode=*/false)); + std::string data = R"([ + [[["e-1", "1970-01-01 00:00:01", "1970-01-01 00:00:02"]], + [["begin", "1970-01-01 00:00:03"]]], + [[["e-2", "1970-01-01 00:00:04", null], ["e-3", null, "1970-01-01 00:00:05"]], []], + [[null], null] + ])"; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), data, + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*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>> data_splits, + helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt)); + std::string expected_data = R"([ + [0, [["e-1", "1970-01-01 00:00:01", "1970-01-01 00:00:02"]], + [["begin", "1970-01-01 00:00:03"]]], + [0, [["e-2", "1970-01-01 00:00:04", null], ["e-3", null, "1970-01-01 00:00:05"]], []], + [0, [null], null] + ])"; + ASSERT_OK_AND_ASSIGN( + bool success, helper->ReadAndCheckResult(arrow::struct_(fields_with_row_kind), data_splits, + expected_data)); + ASSERT_TRUE(success); +} + +/// End-to-end coverage for TIMESTAMP_LTZ(6) nested inside list/struct/map. The Parquet reader +/// reports LTZ leaves as UTC while the read schema carries the local timezone, so read schema and +/// file schema differ only in the timezone of those leaves; the micro precision stays unchanged. +TEST_P(WriteAndReadInteTest, TestAppendNestedTimestampLtzMicroTimezoneOnly) { + auto [file_format, file_system] = GetParam(); + // Pin a non-UTC timezone so the read schema really differs from what the file reports. + TimezoneGuard timezone_guard("Asia/Shanghai"); + auto timezone = DateTimeUtils::GetLocalTimezoneName(); + auto event_type = arrow::struct_({ + arrow::field("name", arrow::utf8()), + arrow::field("ts_ltz_micro", arrow::timestamp(arrow::TimeUnit::MICRO, timezone)), + }); + arrow::FieldVector fields = { + arrow::field("events", arrow::list(arrow::field("element", event_type))), + arrow::field("marks", + arrow::map(arrow::utf8(), arrow::timestamp(arrow::TimeUnit::MICRO, timezone))), + }; + std::map<std::string, std::string> options = { + {Options::MANIFEST_FORMAT, "avro"}, {Options::FILE_FORMAT, file_format}, + {Options::TARGET_FILE_SIZE, "1024"}, {Options::BUCKET, "-1"}, + {Options::FILE_SYSTEM, file_system}, {"orc.timestamp-ltz.legacy.type", "false"}}; + if (file_system == "jindo") { + options = AddOptionsForJindo(options); + } + ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(test_dir_, arrow::schema(fields), + /*partition_keys=*/{}, /*primary_keys=*/{}, + options, /*is_streaming_mode=*/false)); + std::string data = R"([ + [[["e-1", "2026-07-16 12:00:00.000001"]], [["begin", "2026-07-16 12:00:00.000002"]]], + [[["e-2", null], ["e-3", "2026-07-16 12:00:00.000003"]], []], + [[null], null] + ])"; + ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), data, + /*partition_map=*/{}, /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*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>> data_splits, + helper->NewScan(StartupMode::LatestFull(), /*snapshot_id=*/std::nullopt)); + std::string expected_data = R"([ + [0, [["e-1", "2026-07-16 12:00:00.000001"]], [["begin", "2026-07-16 12:00:00.000002"]]], + [0, [["e-2", null], ["e-3", "2026-07-16 12:00:00.000003"]], []], + [0, [null], null] + ])"; + ASSERT_OK_AND_ASSIGN( + bool success, helper->ReadAndCheckResult(arrow::struct_(fields_with_row_kind), data_splits, + expected_data)); + ASSERT_TRUE(success); +} + TEST_P(WriteAndReadInteTest, TestPKWithSequenceFieldInPKField) { arrow::FieldVector fields = { arrow::field("p1", arrow::utf8()),
