This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 92e7e55 fix(parquet): allow reading nested list/map columns whose
leaf types differ only in representation (#172)
92e7e55 is described below
commit 92e7e55a8de56b26a90da5f7dbf29966bad41c8e
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()),
