test: expand map blob read coverage

Author: XiaoHongbo-Hope

src/paimon/format/blob/blob_file_batch_reader_test.cpp

@@ -18,6 +18,7 @@
 
 #include "paimon/format/blob/blob_file_batch_reader.h"
 
+#include <algorithm>
 #include <string_view>
 
 #include "arrow/api.h"
@@ -156,6 +157,28 @@ class BlobFileBatchReaderTest : public testing::Test, public ::testing::WithPara
         return std::string(value->data(), value->size());
     }
 
+    void CheckMapBlobReadFails(const std::string& file_bytes,
+                               const std::shared_ptr<arrow::DataType>& key_type,
+                               const std::string& expected_message) {
+        auto dir = paimon::test::UniqueTestDirectory::Create();
+        ASSERT_TRUE(dir);
+        const std::string file_path = dir->Str() + "/corrupt-map.blob";
+        std::shared_ptr<FileSystem> file_system = std::make_shared<LocalFileSystem>();
+        ASSERT_OK(file_system->WriteFile(file_path, file_bytes, /*overwrite=*/true));
+
+        auto map_type = arrow::map(key_type, BlobUtils::ToArrowField("value", /*nullable=*/true));
+        auto schema = arrow::schema({arrow::field("blob_map", map_type)});
+        ::ArrowSchema c_schema;
+        ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input, file_system->Open(file_path));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<BlobFileBatchReader> reader,
+                             BlobFileBatchReader::Create(
+                                 input, /*batch_size=*/1, /*blob_as_descriptor=*/false,
+                                 /*emit_placeholder_sentinel=*/false, pool_, GetArrowPool(pool_)));
+        ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt));
+        ASSERT_NOK_WITH_MSG(reader->NextBatch(), expected_message);
+    }
+
  private:
     std::string blob_field_name_;
     std::shared_ptr<MemoryPool> pool_;
@@ -413,6 +436,51 @@ TEST_F(BlobFileBatchReaderTest, RejectsNullMapKey) {
     ASSERT_NOK_WITH_MSG(reader->NextBatch(), "MAP<..., BLOB> keys cannot be null");
 }
 
+TEST_F(BlobFileBatchReaderTest, RejectsCorruptMapPayloadMetadata) {
+    // The Java golden's first bin starts with the outer BLOB magic (4 bytes), followed by a
+    // 45-byte MAP payload: header [4, 13), key/value data [13, 35), indexes [35, 41), and the
+    // two index lengths [41, 49). Mutating the payload does not require updating the outer CRC,
+    // which BlobFileBatchReader intentionally does not validate.
+    const std::string golden = MapBlobGoldenBytes();
+
+    std::string corrupted = golden;
+    corrupted[4] = 0;
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "invalid MAP<..., BLOB> payload magic number");
+
+    corrupted = golden;
+    corrupted[8] = 2;
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "unsupported MAP<..., BLOB> payload version");
+
+    corrupted = golden;
+    std::fill(corrupted.begin() + 9, corrupted.begin() + 13, static_cast<char>(0xFF));
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "invalid MAP<..., BLOB> entry count");
+
+    corrupted = golden;
+    corrupted[41] = static_cast<char>(0xFF);
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "invalid MAP<..., BLOB> key index length");
+
+    corrupted = golden;
+    corrupted[9] = 2;
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "entry count does not match key index length");
+
+    CheckMapBlobReadFails(golden, arrow::int32(), "invalid MAP<..., BLOB> fixed-width key length");
+
+    corrupted = golden;
+    corrupted[38] = 0x0C;
+    corrupted[39] = 0x0B;
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "value lengths exceed the payload data length");
+
+    corrupted = golden;
+    corrupted[38] = 0x08;
+    corrupted[39] = 0x07;
+    CheckMapBlobReadFails(corrupted, arrow::utf8(),
+                          "key/value lengths do not match the payload data length");
+
+    corrupted = golden;
+    std::copy_n(corrupted.begin() + 13, 5, corrupted.begin() + 18);
+    CheckMapBlobReadFails(corrupted, arrow::utf8(), "payload: duplicate key");
+}
+
 TEST_P(BlobFileBatchReaderTest, TestPushdownBitmap) {
     std::string test_data_path = paimon::test::GetDataDir() + "/db_with_blob.db/table_with_blob/";
     auto dir = paimon::test::UniqueTestDirectory::Create();

test/inte/blob_table_inte_test.cpp

@@ -535,6 +535,50 @@ class BlobTableInteTest : public testing::Test, public ::testing::WithParamInter
             });
     }
 
+    Result<std::shared_ptr<arrow::MapArray>> NormalizeMapBlobValues(
+        const std::shared_ptr<arrow::MapArray>& map_array, bool blob_as_descriptor) const {
+        const auto& values = checked_cast<const arrow::LargeBinaryArray&>(*map_array->items());
+        auto fs = std::make_shared<LocalFileSystem>();
+        arrow::LargeBinaryBuilder builder;
+        for (int64_t i = 0; i < values.length(); ++i) {
+            if (values.IsNull(i)) {
+                PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull());
+                continue;
+            }
+            std::string_view stored = values.GetView(i);
+            if (!blob_as_descriptor) {
+                PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(stored));
+                continue;
+            }
+            PAIMON_ASSIGN_OR_RAISE(
+                std::unique_ptr<Blob> blob,
+                Blob::FromDescriptor(stored.data(), static_cast<int64_t>(stored.size())));
+            PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> data, blob->ToData(fs, pool_));
+            PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append(data->data(), data->size()));
+        }
+        std::shared_ptr<arrow::Array> normalized_values;
+        PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&normalized_values));
+        return std::make_shared<arrow::MapArray>(map_array->type(), map_array->length(),
+                                                 map_array->value_offsets(), map_array->keys(),
+                                                 normalized_values, map_array->null_bitmap(),
+                                                 map_array->null_count(), map_array->offset());
+    }
+
+    void CheckMapBlobColumn(const std::shared_ptr<arrow::StructArray>& rows,
+                            const std::string& field_name, const std::string& expected_json,
+                            bool blob_as_descriptor) const {
+        auto map_array =
+            std::dynamic_pointer_cast<arrow::MapArray>(rows->GetFieldByName(field_name));
+        ASSERT_TRUE(map_array) << field_name;
+        ASSERT_OK_AND_ASSIGN(auto normalized,
+                             NormalizeMapBlobValues(map_array, blob_as_descriptor));
+        auto expected = arrow::ipc::internal::json::ArrayFromJSON(map_array->type(), expected_json)
+                            .ValueOrDie();
+        ASSERT_TRUE(expected->Equals(normalized))
+            << field_name << " expected: " << expected->ToString()
+            << " actual: " << normalized->ToString();
+    }
+
     /// Verify DataFileMeta properties from a scan plan.
     /// Each vector element corresponds to one expected DataFileMeta (ordered by file index).
     static void VerifyDataFileMetas(
@@ -4334,4 +4378,111 @@ TEST_P(BlobTableInteTest, TestReadBlobDescriptorFieldFromJava) {
     ASSERT_TRUE(resolved->Equals(expected_with_rk));
 }
 
+TEST_P(BlobTableInteTest, TestReadMapBlobTableFromJava) {
+    if (GetParam() != "parquet") {
+        GTEST_SKIP() << "the Java fixture uses Parquet";
+    }
+    const std::string table_path = GetDataDir() + "/parquet/map_blob_java.db/map_blob_java";
+    const std::vector<std::string> read_fields = {"id",
+                                                  "string_payloads",
+                                                  "boolean_payloads",
+                                                  "tinyint_payloads",
+                                                  "smallint_payloads",
+                                                  "int_payloads",
+                                                  "bigint_payloads",
+                                                  "date_payloads",
+                                                  "binary_payloads",
+                                                  "compact_decimal_payloads",
+                                                  "large_decimal_payloads"};
+
+    for (int64_t snapshot_id : {1, 3}) {
+        ScanContextBuilder scan_builder(table_path);
+        scan_builder.AddOption(Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id));
+        ASSERT_OK_AND_ASSIGN(auto scan_context, scan_builder.Finish());
+        ASSERT_OK_AND_ASSIGN(auto table_scan, TableScan::Create(std::move(scan_context)));
+        ASSERT_OK_AND_ASSIGN(auto plan, table_scan->CreatePlan());
+
+        size_t string_layer_count = 0;
+        for (const auto& split : plan->Splits()) {
+            auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+            ASSERT_TRUE(data_split);
+            for (const auto& file : data_split->DataFiles()) {
+                if (file->write_cols ==
+                    std::optional<std::vector<std::string>>({"string_payloads"})) {
+                    ++string_layer_count;
+                }
+            }
+        }
+        ASSERT_EQ(snapshot_id == 1 ? 1 : 3, string_layer_count);
+
+        for (bool blob_as_descriptor : {false, true}) {
+            std::map<std::string, std::string> read_options = {
+                {Options::BLOB_AS_DESCRIPTOR, blob_as_descriptor ? "true" : "false"}};
+            ASSERT_OK_AND_ASSIGN(auto result, ReadTable(table_path, read_fields, plan,
+                                                        /*predicate=*/nullptr, read_options));
+            ASSERT_TRUE(result);
+            auto combined = arrow::Concatenate(result->chunks()).ValueOrDie();
+            auto rows = std::dynamic_pointer_cast<arrow::StructArray>(combined);
+            ASSERT_TRUE(rows);
+            ASSERT_EQ(4, rows->length());
+            const auto& ids = checked_cast<const arrow::Int32Array&>(*rows->GetFieldByName("id"));
+            for (int64_t i = 0; i < ids.length(); ++i) {
+                ASSERT_EQ(i + 1, ids.Value(i));
+            }
+
+            CheckMapBlobColumn(
+                rows, "string_payloads",
+                R"json([[["", "string-empty"], ["alpha", "string-alpha"]], [], null, [["omega", "string-omega"]]])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "boolean_payloads",
+                R"json([[[false, "bool-false"], [true, "bool-true"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "tinyint_payloads",
+                R"json([[[-128, "tiny-min"], [-1, "tiny-negative"], [127, "tiny-max"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "smallint_payloads",
+                R"json([[[-32768, "small-min"], [-1, "small-negative"], [32767, "small-max"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "int_payloads",
+                R"json([[[-2147483648, "int-min"], [-1, "int-negative"], [2147483647, "int-max"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "bigint_payloads",
+                R"json([[[-9223372036854775808, "big-min"], [-1, "big-negative"], [9223372036854775807, "big-max"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "date_payloads",
+                R"json([[[-1, "date-negative"], [0, "date-epoch"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "compact_decimal_payloads",
+                R"json([[["-99999999.99", "compact-negative"], ["99999999.99", "compact-positive"]], null, null, null])json",
+                blob_as_descriptor);
+            CheckMapBlobColumn(
+                rows, "large_decimal_payloads",
+                R"json([[["-999999999999999999.99", "large-negative"], ["999999999999999999.99", "large-positive"]], null, null, null])json",
+                blob_as_descriptor);
+
+            auto binary_map =
+                std::dynamic_pointer_cast<arrow::MapArray>(rows->GetFieldByName("binary_payloads"));
+            ASSERT_TRUE(binary_map);
+            ASSERT_OK_AND_ASSIGN(auto normalized_binary,
+                                 NormalizeMapBlobValues(binary_map, blob_as_descriptor));
+            const auto& binary_keys =
+                checked_cast<const arrow::BinaryArray&>(*normalized_binary->keys());
+            const auto& binary_values =
+                checked_cast<const arrow::LargeBinaryArray&>(*normalized_binary->items());
+            ASSERT_EQ(2, normalized_binary->value_length(0));
+            ASSERT_EQ("", binary_keys.GetString(0));
+            ASSERT_EQ(std::string("\0\xff\1\2", 4), binary_keys.GetString(1));
+            ASSERT_EQ("binary-empty", binary_values.GetString(0));
+            ASSERT_EQ("binary-bytes", binary_values.GetString(1));
+        }
+    }
+}
+
 }  // namespace paimon::test

test/test_data/parquet/map_blob_java.db/map_blob_java/README.md

@@ -0,0 +1,26 @@
+<!--
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+-->
+
+# Java MAP&lt;K, BLOB&gt; fixture
+
+Generated by Apache Paimon Java at commit `a176eba1c6f9b0402eceea641bf435b05976470b`.
+
+Snapshot 1 contains a full write. Snapshots 2 and 3 each add a partial write containing
+`BlobMapPlaceholder` values for `string_payloads`, so snapshot 3 exercises fallback across three
+sequence layers. The remaining map columns cover BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, DATE,
+BINARY, compact DECIMAL, and large DECIMAL keys, including negative and boundary values.