fix(parquet): allow reading nested list/map columns whose leaf types differ only in representation (#172)

Author: zhf999

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)++);

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/"