feat: add FromView/ToView api for BlobView

include/paimon/data/blob_view.h

@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "paimon/catalog/identifier.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// Coordinates of a BLOB value stored in an upstream table.
+struct PAIMON_EXPORT BlobView {
+    Identifier identifier;
+    int32_t field_id;
+    int64_t row_id;
+
+    bool operator==(const BlobView& other) const;
+    bool operator!=(const BlobView& other) const;
+
+    std::string ToString() const;
+
+    /// Deserializes BLOB view bytes into a BlobView.
+    static Result<BlobView> FromView(const char* buffer, uint64_t size);
+
+    /// Serializes a BlobView into BLOB view bytes.
+    static Result<PAIMON_UNIQUE_PTR<Bytes>> ToView(const BlobView& blob_view,
+                                                   const std::shared_ptr<MemoryPool>& pool);
+};
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -28,6 +28,7 @@ set(PAIMON_COMMON_SRCS
     common/data/binary_string.cpp
     common/data/blob.cpp
     common/data/blob_descriptor.cpp
+    common/data/blob_view.cpp
     common/data/blob_view_struct.cpp
     common/data/blob_utils.cpp
     common/data/columnar/columnar_array.cpp
@@ -566,6 +567,7 @@ if(PAIMON_BUILD_TESTS)
                     common/data/timestamp_test.cpp
                     common/data/blob_test.cpp
                     common/data/blob_descriptor_test.cpp
+                    common/data/blob_view_test.cpp
                     common/data/blob_view_struct_test.cpp
                     common/data/blob_utils_test.cpp
                     common/data/variant/generic_variant_test.cpp

src/paimon/common/data/blob_utils_test.cpp

@@ -359,7 +359,7 @@ TEST_F(BlobUtilsTest, ValidateBlobViewFieldsFieldNotPresent) {
 TEST_F(BlobUtilsTest, ValidateBlobViewFieldsWithValidViewStruct) {
     // A BlobViewStruct value is accepted for a view field.
     BlobViewStruct view_struct(Identifier("db", "tbl"), /*field_id=*/2, /*row_id=*/5);
-    auto serialized = view_struct.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct.Serialize(pool_));
 
     arrow::LargeBinaryBuilder builder;
     ASSERT_TRUE(builder.Append(serialized->data(), serialized->size()).ok());

src/paimon/common/data/blob_view.cpp

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+#include "paimon/data/blob_view.h"
+
+#include <memory>
+
+#include "fmt/format.h"
+#include "paimon/common/data/blob_view_struct.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+bool BlobView::operator==(const BlobView& other) const {
+    return identifier == other.identifier && field_id == other.field_id && row_id == other.row_id;
+}
+
+bool BlobView::operator!=(const BlobView& other) const {
+    return !(*this == other);
+}
+
+std::string BlobView::ToString() const {
+    return fmt::format("BlobView{{identifier={}, fieldId={}, rowId={}}}", identifier.GetFullName(),
+                       field_id, row_id);
+}
+
+Result<BlobView> BlobView::FromView(const char* buffer, uint64_t size) {
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BlobViewStruct> view_struct,
+                           BlobViewStruct::Deserialize(buffer, size));
+    return BlobView{view_struct->GetIdentifier(), view_struct->FieldId(), view_struct->RowId()};
+}
+
+Result<PAIMON_UNIQUE_PTR<Bytes>> BlobView::ToView(const BlobView& blob_view,
+                                                  const std::shared_ptr<MemoryPool>& pool) {
+    if (pool == nullptr) {
+        return Status::Invalid("memory pool is nullptr");
+    }
+
+    BlobViewStruct view_struct(blob_view.identifier, blob_view.field_id, blob_view.row_id);
+    return view_struct.Serialize(pool);
+}
+
+}  // namespace paimon

src/paimon/common/data/blob_view_struct.cpp

@@ -18,6 +18,7 @@
  */
 #include "paimon/common/data/blob_view_struct.h"
 
+#include <limits>
 #include <utility>
 
 #include "fmt/format.h"
@@ -31,7 +32,13 @@
 #include "paimon/status.h"
 
 namespace paimon {
-PAIMON_UNIQUE_PTR<Bytes> BlobViewStruct::Serialize(const std::shared_ptr<MemoryPool>& pool) const {
+Result<PAIMON_UNIQUE_PTR<Bytes>> BlobViewStruct::Serialize(
+    const std::shared_ptr<MemoryPool>& pool) const {
+    if (identifier_.GetDatabaseName() == Identifier::kUnknownDatabase) {
+        return Status::Invalid(
+            fmt::format("Blob view upstream table identifier must include database name: {}",
+                        identifier_.GetFullName()));
+    }
     MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
     out.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN);
 
@@ -74,7 +81,7 @@ Result<std::unique_ptr<BlobViewStruct>> BlobViewStruct::Deserialize(const char*
 }
 
 Result<bool> BlobViewStruct::IsBlobViewStruct(const char* buffer, uint64_t size) {
-    if (size < kMinViewLength) {
+    if (buffer == nullptr || size < kMinViewLength) {
         return false;
     }
     auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, size);

src/paimon/common/data/blob_view_struct.h

@@ -53,7 +53,7 @@ class BlobViewStruct {
 
     static Result<std::unique_ptr<BlobViewStruct>> Deserialize(const char* buffer, uint64_t size);
     static Result<bool> IsBlobViewStruct(const char* buffer, uint64_t size);
-    PAIMON_UNIQUE_PTR<Bytes> Serialize(const std::shared_ptr<MemoryPool>& pool) const;
+    Result<PAIMON_UNIQUE_PTR<Bytes>> Serialize(const std::shared_ptr<MemoryPool>& pool) const;
     std::string ToString() const;
     int32_t HashCode() const;
 

src/paimon/common/data/blob_view_struct_test.cpp

@@ -46,7 +46,7 @@ TEST_F(BlobViewStructTest, TestConstructorAndGetters) {
 }
 
 TEST_F(BlobViewStructTest, TestSerializeDeserializeRoundTrip) {
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     ASSERT_NE(serialized, nullptr);
     ASSERT_GT(serialized->size(), 0u);
 
@@ -59,14 +59,14 @@ TEST_F(BlobViewStructTest, TestSerializeDeserializeRoundTrip) {
 }
 
 TEST_F(BlobViewStructTest, TestDeserializeWithInvalidVersion) {
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     (*serialized)[0] = '\x02';  // invalid version (current is 1).
     ASSERT_NOK_WITH_MSG(BlobViewStruct::Deserialize(serialized->data(), serialized->size()),
                         "Expecting BlobViewStruct version to be 1, but found 2");
 }
 
 TEST_F(BlobViewStructTest, TestDeserializeWithInvalidMagic) {
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     (*serialized)[1] = '\x00';
     ASSERT_NOK_WITH_MSG(BlobViewStruct::Deserialize(serialized->data(), serialized->size()),
                         "missing magic header");
@@ -77,6 +77,12 @@ TEST_F(BlobViewStructTest, TestToString) {
     ASSERT_EQ(debug_str, "BlobViewStruct{identifier=test_db.test_table, fieldId=7, rowId=1024}");
 }
 
+TEST_F(BlobViewStructTest, TestSerializeRejectsUnknownDatabase) {
+    BlobViewStruct view_struct(Identifier("test_table"), /*field_id=*/7, /*row_id=*/1024);
+    ASSERT_NOK_WITH_MSG(view_struct.Serialize(pool_),
+                        "upstream table identifier must include database name");
+}
+
 TEST_F(BlobViewStructTest, TestEqual) {
     {
         // test equal itself
@@ -110,7 +116,7 @@ TEST_F(BlobViewStructTest, TestEqual) {
 }
 
 TEST_F(BlobViewStructTest, TestIsBlobViewStructValid) {
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     ASSERT_OK_AND_ASSIGN(bool result,
                          BlobViewStruct::IsBlobViewStruct(serialized->data(), serialized->size()));
     ASSERT_TRUE(result);
@@ -130,7 +136,7 @@ TEST_F(BlobViewStructTest, TestIsBlobViewStructWithTooShortBuffer) {
 
 TEST_F(BlobViewStructTest, TestIsBlobViewStructWithFutureVersion) {
     // Version > CURRENT_VERSION should return false (not an error)
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     (*serialized)[0] = '\x02';  // set version to 2 (> CURRENT_VERSION)
     ASSERT_OK_AND_ASSIGN(bool result,
                          BlobViewStruct::IsBlobViewStruct(serialized->data(), serialized->size()));
@@ -139,7 +145,7 @@ TEST_F(BlobViewStructTest, TestIsBlobViewStructWithFutureVersion) {
 
 TEST_F(BlobViewStructTest, TestIsBlobViewStructWithWrongMagic) {
     // Wrong magic number should return false
-    auto serialized = view_struct_.Serialize(pool_);
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, view_struct_.Serialize(pool_));
     // Corrupt the magic bytes (bytes 1-8)
     (*serialized)[1] = '\x00';
     (*serialized)[2] = '\x00';

src/paimon/common/data/blob_view_test.cpp

@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+#include "paimon/data/blob_view.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/catalog/identifier.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class BlobViewTest : public testing::Test {
+ public:
+    std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+    BlobView blob_view_{Identifier("default", "source"), /*field_id=*/7, /*row_id=*/5};
+};
+
+TEST_F(BlobViewTest, TestFieldsAndEquality) {
+    ASSERT_EQ(blob_view_.identifier, Identifier("default", "source"));
+    ASSERT_EQ(blob_view_.field_id, 7);
+    ASSERT_EQ(blob_view_.row_id, 5);
+
+    BlobView equal{Identifier("default", "source"), /*field_id=*/7, /*row_id=*/5};
+    BlobView different_identifier{Identifier("other", "source"), /*field_id=*/7, /*row_id=*/5};
+    BlobView different_field{Identifier("default", "source"), /*field_id=*/8, /*row_id=*/5};
+    BlobView different_row{Identifier("default", "source"), /*field_id=*/7, /*row_id=*/6};
+    ASSERT_EQ(blob_view_, equal);
+    ASSERT_NE(blob_view_, different_identifier);
+    ASSERT_NE(blob_view_, different_field);
+    ASSERT_NE(blob_view_, different_row);
+}
+
+TEST_F(BlobViewTest, TestToString) {
+    ASSERT_EQ(blob_view_.ToString(), "BlobView{identifier=default.source, fieldId=7, rowId=5}");
+}
+
+TEST_F(BlobViewTest, TestRoundTrip) {
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> serialized, BlobView::ToView(blob_view_, pool_));
+    ASSERT_OK_AND_ASSIGN(BlobView restored,
+                         BlobView::FromView(serialized->data(), serialized->size()));
+    ASSERT_EQ(restored, blob_view_);
+}
+
+TEST_F(BlobViewTest, TestJavaSerializationCompatibility) {
+    // Serialized by Java BlobViewStruct(Identifier.fromString("default.source"), 7, 5L).
+    std::vector<char> java_serialized = {
+        1,  87,  69,  73,  86,  66, 79,  76, 66, 14, 0, 0, 0, 100, 101, 102, 97, 117, 108, 116,
+        46, 115, 111, 117, 114, 99, 101, 7,  0,  0,  0, 5, 0, 0,   0,   0,   0,  0,   0};
+
+    ASSERT_OK_AND_ASSIGN(BlobView restored,
+                         BlobView::FromView(java_serialized.data(), java_serialized.size()));
+    ASSERT_EQ(restored, blob_view_);
+
+    ASSERT_OK_AND_ASSIGN(PAIMON_UNIQUE_PTR<Bytes> cpp_serialized,
+                         BlobView::ToView(blob_view_, pool_));
+    ASSERT_EQ(std::string(cpp_serialized->data(), cpp_serialized->size()),
+              std::string(java_serialized.data(), java_serialized.size()));
+}
+
+TEST_F(BlobViewTest, TestToViewRejectsInvalidArguments) {
+    BlobView unknown_database{Identifier("source"), /*field_id=*/7, /*row_id=*/5};
+    ASSERT_NOK_WITH_MSG(BlobView::ToView(unknown_database, pool_),
+                        "upstream table identifier must include database name");
+    ASSERT_NOK_WITH_MSG(BlobView::ToView(blob_view_, /*pool=*/nullptr), "memory pool is nullptr");
+}
+
+}  // namespace paimon::test

src/paimon/common/reader/blob_view_resolving_batch_reader_test.cpp

@@ -91,11 +91,12 @@ class BlobViewResolvingBatchReaderTest : public ::testing::Test {
         bool exhausted_ = false;
     };
 
-    std::string MakeBlobViewStructBytes(const std::string& database, const std::string& table,
-                                        int32_t field_id, int64_t row_id) const {
+    Result<std::string> MakeBlobViewStructBytes(const std::string& database,
+                                                const std::string& table, int32_t field_id,
+                                                int64_t row_id) const {
         Identifier identifier(database, table);
         BlobViewStruct view_struct(identifier, field_id, row_id);
-        auto bytes = view_struct.Serialize(pool_);
+        PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> bytes, view_struct.Serialize(pool_));
         return std::string(bytes->data(), bytes->size());
     }
 
@@ -145,7 +146,8 @@ TEST_F(BlobViewResolvingBatchReaderTest, TestEofBatch) {
 }
 
 TEST_F(BlobViewResolvingBatchReaderTest, TestEmptyReadBlobViewFields) {
-    std::string view_bytes = MakeBlobViewStructBytes("db", "table", /*field_id=*/1, /*row_id=*/7);
+    ASSERT_OK_AND_ASSIGN(std::string view_bytes,
+                         MakeBlobViewStructBytes("db", "table", /*field_id=*/1, /*row_id=*/7));
     std::shared_ptr<arrow::StructArray> struct_array = BuildStructArray({view_bytes}, {true});
 
     bool resolver_called = false;
@@ -166,8 +168,10 @@ TEST_F(BlobViewResolvingBatchReaderTest, TestEmptyReadBlobViewFields) {
 }
 
 TEST_F(BlobViewResolvingBatchReaderTest, TestResolvesBlobViewColumn) {
-    auto row0_view = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/100);
-    auto row1_view = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/200);
+    ASSERT_OK_AND_ASSIGN(std::string row0_view,
+                         MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/100));
+    ASSERT_OK_AND_ASSIGN(std::string row1_view,
+                         MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/200));
     std::shared_ptr<arrow::StructArray> src_struct =
         BuildStructArray({row0_view, row1_view}, {true, true});
 
@@ -203,7 +207,8 @@ TEST_F(BlobViewResolvingBatchReaderTest, TestResolvesBlobViewColumn) {
 }
 
 TEST_F(BlobViewResolvingBatchReaderTest, TestResolverError) {
-    auto view_bytes = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/1, /*row_id=*/5);
+    ASSERT_OK_AND_ASSIGN(std::string view_bytes,
+                         MakeBlobViewStructBytes("db", "tbl", /*field_id=*/1, /*row_id=*/5));
     std::shared_ptr<arrow::StructArray> src_struct = BuildStructArray({view_bytes}, {true});
     auto resolver = BlobViewResolver([](const BlobViewStruct&) -> Result<std::shared_ptr<Bytes>> {
         return Status::Invalid("cache miss");