feat(blob): support blob-view-field for cross-table blob reference re…

include/paimon/catalog/identifier.h

@@ -36,15 +36,20 @@ class PAIMON_EXPORT Identifier {
     explicit Identifier(const std::string& table);
     Identifier(const std::string& database, const std::string& table);
 
-    bool operator==(const Identifier& other);
+    bool operator==(const Identifier& other) const;
     const std::string& GetDatabaseName() const;
     const std::string& GetTableName() const;
     Result<std::string> GetDataTableName() const;
     Result<std::optional<std::string>> GetBranchName() const;
     Result<std::string> GetBranchNameOrDefault() const;
     Result<std::optional<std::string>> GetSystemTableName() const;
     Result<bool> IsSystemTable() const;
+    std::string GetFullName() const;
     std::string ToString() const;
+    int32_t HashCode() const;
+
+ public:
+    static Result<Identifier> FromString(const std::string& full_name);
 
  private:
     Status SplitTableName() const;
@@ -58,3 +63,13 @@ class PAIMON_EXPORT Identifier {
 };
 
 }  // namespace paimon
+
+namespace std {
+template <>
+struct hash<paimon::Identifier> {
+    size_t operator()(const paimon::Identifier& identifier) const {
+        return identifier.HashCode();
+    }
+};
+
+}  // namespace std

src/paimon/CMakeLists.txt

@@ -25,6 +25,7 @@ set(PAIMON_COMMON_SRCS
     common/data/binary_string.cpp
     common/data/blob.cpp
     common/data/blob_descriptor.cpp
+    common/data/blob_view_struct.cpp
     common/data/blob_utils.cpp
     common/data/columnar/columnar_array.cpp
     common/data/columnar/columnar_map.cpp
@@ -110,6 +111,7 @@ set(PAIMON_COMMON_SRCS
     common/reader/predicate_batch_reader.cpp
     common/reader/prefetch_file_batch_reader_impl.cpp
     common/reader/reader_utils.cpp
+    common/reader/blob_view_resolving_batch_reader.cpp
     common/reader/complete_row_kind_batch_reader.cpp
     common/reader/data_evolution_file_reader.cpp
     common/sst/block_handle.cpp
@@ -332,6 +334,7 @@ set(PAIMON_CORE_SRCS
     core/table/system/system_table_schema.cpp
     core/tag/tag.cpp
     core/utils/branch_manager.cpp
+    core/utils/blob_view_lookup.cpp
     core/utils/consumer_manager.cpp
     core/utils/field_mapping.cpp
     core/utils/file_store_path_factory.cpp
@@ -408,6 +411,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_struct_test.cpp
                     common/data/blob_utils_test.cpp
                     common/executor/default_executor_test.cpp
                     common/format/column_stats_test.cpp
@@ -470,6 +474,7 @@ if(PAIMON_BUILD_TESTS)
                     common/reader/prefetch_file_batch_reader_impl_test.cpp
                     common/reader/reader_utils_test.cpp
                     common/reader/complete_row_kind_batch_reader_test.cpp
+                    common/reader/blob_view_resolving_batch_reader_test.cpp
                     common/reader/data_evolution_file_reader_test.cpp
                     common/reader/data_evolution_array_test.cpp
                     common/reader/data_evolution_row_test.cpp
@@ -715,6 +720,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/table_scan_test.cpp
                     core/table/system/system_table_test.cpp
                     core/tag/tag_test.cpp
+                    core/utils/blob_view_lookup_test.cpp
                     core/utils/branch_manager_test.cpp
                     core/utils/consumer_manager_test.cpp
                     core/utils/file_store_path_factory_cache_test.cpp

src/paimon/common/catalog/catalog_context.h

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <map>
+#include <string>
+
+#include "paimon/fs/file_system.h"
+
+namespace paimon {
+struct CatalogContext {
+    CatalogContext(const std::string& _root_path,
+                   const std::map<std::string, std::string>& _options,
+                   const std::shared_ptr<FileSystem>& _file_system)
+        : root_path(_root_path), options(_options), file_system(_file_system) {}
+
+    std::string root_path;
+    std::map<std::string, std::string> options;
+    std::shared_ptr<FileSystem> file_system;
+};
+
+}  // namespace paimon

src/paimon/common/data/blob_descriptor.cpp

@@ -50,7 +50,7 @@ Result<std::unique_ptr<BlobDescriptor>> BlobDescriptor::Create(int8_t version,
 PAIMON_UNIQUE_PTR<Bytes> BlobDescriptor::Serialize(const std::shared_ptr<MemoryPool>& pool) const {
     MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
     out.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN);
-    out.WriteValue<int8_t>(version_);
+    out.WriteValue<int8_t>(kCurrentVersion);
     out.WriteValue<int64_t>(kMagic);
     out.WriteValue<int32_t>(static_cast<int32_t>(uri_.size()));
 

src/paimon/common/data/blob_descriptor_test.cpp

@@ -56,6 +56,10 @@ TEST_F(BlobDescriptorTest, TestDeserializeCompatibilityForJavaWithVersion1) {
     ASSERT_EQ(descriptor->Uri(), "test_uri");
     ASSERT_EQ(descriptor->Offset(), 1024);
     ASSERT_EQ(descriptor->Length(), 2048);
+
+    PAIMON_UNIQUE_PTR<Bytes> cpp_serialized = descriptor->Serialize(pool_);
+    auto cpp_serialized_string = std::string(cpp_serialized->data(), cpp_serialized->size());
+    ASSERT_NE(cpp_serialized_string, java_serialized);
 }
 
 TEST_F(BlobDescriptorTest, TestDeserializeCompatibilityForJavaWithVersion2) {

src/paimon/common/data/blob_view_struct.cpp

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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/common/data/blob_view_struct.h"
+
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/common/utils/murmurhash_utils.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/byte_order.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/status.h"
+
+namespace paimon {
+PAIMON_UNIQUE_PTR<Bytes> BlobViewStruct::Serialize(const std::shared_ptr<MemoryPool>& pool) const {
+    MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+    out.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN);
+
+    out.WriteValue<int8_t>(kCurrentVersion);
+    out.WriteValue<int64_t>(kMagic);
+    std::string identifier = identifier_.GetFullName();
+    out.WriteValue<int32_t>(static_cast<int32_t>(identifier.size()));
+    auto uri_bytes = std::make_shared<Bytes>(identifier, pool.get());
+    out.WriteBytes(uri_bytes);
+    out.WriteValue<int32_t>(field_id_);
+    out.WriteValue<int64_t>(row_id_);
+    return MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get());
+}
+
+Result<std::unique_ptr<BlobViewStruct>> BlobViewStruct::Deserialize(const char* buffer,
+                                                                    uint64_t size) {
+    auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, size);
+    DataInputStream in(std::move(input_stream));
+    in.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN);
+
+    PAIMON_ASSIGN_OR_RAISE(int8_t version, in.ReadValue<int8_t>());
+    if (version != kCurrentVersion) {
+        return Status::Invalid(fmt::format(
+            "Expecting BlobViewStruct version to be {}, but found {}.", kCurrentVersion, version));
+    }
+    PAIMON_ASSIGN_OR_RAISE(int64_t magic, in.ReadValue<int64_t>());
+    if (kMagic != magic) {
+        return Status::Invalid(
+            fmt::format("Invalid BlobViewStruct: missing magic header. Expected magic: {}, "
+                        "but found {}",
+                        kMagic, magic));
+    }
+    PAIMON_ASSIGN_OR_RAISE(int32_t length, in.ReadValue<int32_t>());
+    std::string identifier_str(length, '\0');
+    PAIMON_RETURN_NOT_OK(in.Read(identifier_str.data(), identifier_str.size()));
+    PAIMON_ASSIGN_OR_RAISE(int32_t field_id, in.ReadValue<int32_t>());
+    PAIMON_ASSIGN_OR_RAISE(int64_t row_id, in.ReadValue<int64_t>());
+    PAIMON_ASSIGN_OR_RAISE(Identifier identifier, Identifier::FromString(identifier_str));
+    return std::make_unique<BlobViewStruct>(identifier, field_id, row_id);
+}
+
+Result<bool> BlobViewStruct::IsBlobViewStruct(const char* buffer, uint64_t size) {
+    if (size < kMinViewLength) {
+        return false;
+    }
+    auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, size);
+    DataInputStream in(std::move(input_stream));
+    in.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN);
+
+    PAIMON_ASSIGN_OR_RAISE(int8_t version, in.ReadValue<int8_t>());
+    if (version != kCurrentVersion) {
+        return false;
+    }
+    PAIMON_ASSIGN_OR_RAISE(int64_t magic, in.ReadValue<int64_t>());
+    return kMagic == magic;
+}
+
+std::string BlobViewStruct::ToString() const {
+    return fmt::format("BlobViewStruct{{identifier={}, fieldId={}, rowId={}}}",
+                       identifier_.GetFullName(), field_id_, row_id_);
+}
+
+bool BlobViewStruct::operator==(const BlobViewStruct& other) const {
+    if (this == &other) {
+        return true;
+    }
+    return field_id_ == other.field_id_ && row_id_ == other.row_id_ &&
+           identifier_ == other.identifier_;
+}
+
+bool BlobViewStruct::operator!=(const BlobViewStruct& other) const {
+    return !(*this == other);
+}
+
+int32_t BlobViewStruct::HashCode() const {
+    int32_t hash =
+        MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&field_id_),
+                                         /*offset=*/0, sizeof(field_id_), identifier_.HashCode());
+    return MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&row_id_),
+                                            /*offset=*/0, sizeof(row_id_), hash);
+}
+}  // namespace paimon

src/paimon/common/data/blob_view_struct.h

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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 <functional>
+#include <memory>
+#include <string>
+
+#include "paimon/catalog/identifier.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+
+namespace paimon {
+/// Serialized metadata for a BLOB view field.
+/// A blob view only stores the coordinates needed to locate the original blob value in the
+/// upstream table: identifier, field_id and row_id. The actual blob data is
+/// resolved at read time by scanning the upstream table.
+class BlobViewStruct {
+ public:
+    BlobViewStruct(const Identifier& identifier, int32_t field_id, int64_t row_id)
+        : identifier_(identifier), field_id_(field_id), row_id_(row_id) {}
+
+    const Identifier& GetIdentifier() const {
+        return identifier_;
+    }
+
+    int32_t FieldId() const {
+        return field_id_;
+    }
+
+    int64_t RowId() const {
+        return row_id_;
+    }
+
+    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;
+    std::string ToString() const;
+    int32_t HashCode() const;
+
+    bool operator==(const BlobViewStruct& other) const;
+    bool operator!=(const BlobViewStruct& other) const;
+
+ private:
+    static constexpr int64_t kMagic = 0x424C4F4256494557l;
+    static constexpr int8_t kCurrentVersion = 1;
+    /// one byte for version, eight bytes for magic number.
+    static constexpr uint64_t kMinViewLength = 9;
+
+    Identifier identifier_;
+    int32_t field_id_;
+    int64_t row_id_;
+};
+
+/// Resolves a BlobViewStruct into the serialized BlobDescriptor bytes stored in the upstream
+/// table. Returns nullptr when the referenced source-table cell is null.
+using BlobViewResolver = std::function<Result<std::shared_ptr<Bytes>>(const BlobViewStruct&)>;
+
+}  // namespace paimon
+
+namespace std {
+template <>
+struct hash<paimon::BlobViewStruct> {
+    size_t operator()(const paimon::BlobViewStruct& blob_view_struct) const {
+        return blob_view_struct.HashCode();
+    }
+};
+
+}  // namespace std