feat(blob view): support blob-view.resolve.enabled to preserve blob view references at read time

Align with CoreOptions.BLOB_VIEW_RESOLVE_ENABLED in apache/paimon: the
option defaults to true; when set to false the read passes the serialized
BlobViewStruct bytes through without resolving them from upstream tables,
skips the prescan/resolver entirely, and no longer requires
blob-view-upstream-warehouse, so blob view values can be forwarded to
another blob-view table.

Closes alibaba#413

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: SteNicholas

include/paimon/defs.h

@@ -408,6 +408,10 @@ struct PAIMON_EXPORT Options {
     /// serialized BlobViewStruct bytes inline in data files and resolve from upstream tables at
     /// read time. No default value.
     static const char BLOB_VIEW_FIELD[];
+    /// "blob-view.resolve.enabled" - Whether to resolve blob-view-field values from upstream
+    /// tables at read time. Set to false to preserve serialized BlobViewStruct bytes when
+    /// forwarding blob view values to another blob-view table. Default value is "true".
+    static const char BLOB_VIEW_RESOLVE_ENABLED[];
     /// "blob-view-upstream-warehouse" - Since the catalog capabilities are partially missing, when
     /// Blob View is enabled, cpp paimon cannot automatically obtain the upstream table warehouse
     /// path and requires manual configuration by the user. No default value.

src/paimon/common/defs.cpp

@@ -102,6 +102,7 @@ const char Options::BLOB_FIELD[] = "blob-field";
 const char Options::BLOB_DESCRIPTOR_FIELD[] = "blob-descriptor-field";
 const char Options::FALLBACK_BLOB_DESCRIPTOR_FIELD[] = "blob.stored-descriptor-fields";
 const char Options::BLOB_VIEW_FIELD[] = "blob-view-field";
+const char Options::BLOB_VIEW_RESOLVE_ENABLED[] = "blob-view.resolve.enabled";
 const char Options::BLOB_VIEW_UPSTREAM_WAREHOUSE[] = "blob-view-upstream-warehouse";
 const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 const char Options::GLOBAL_INDEX_THREAD_NUM[] = "global-index.thread-num";

src/paimon/core/core_options.cpp

@@ -442,6 +442,7 @@ struct CoreOptions::Impl {
     bool row_tracking_enabled = false;
     bool row_tracking_partition_group_on_commit = true;
     bool data_evolution_enabled = false;
+    bool blob_view_resolve_enabled = true;
     bool legacy_partition_name_enabled = true;
     bool global_index_enabled = true;
     std::optional<int32_t> global_index_thread_num;
@@ -563,6 +564,9 @@ struct CoreOptions::Impl {
         // Parse blob-view-upstream-warehouse - warehouse path for configured blob view fields
         PAIMON_RETURN_NOT_OK(
             parser.Parse(Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, &blob_view_upstream_warehouse));
+        // Parse blob-view.resolve.enabled - whether to resolve blob view fields at read time
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse<bool>(Options::BLOB_VIEW_RESOLVE_ENABLED, &blob_view_resolve_enabled));
         return Status::OK();
     }
 
@@ -1489,6 +1493,10 @@ std::optional<std::string> CoreOptions::GetBlobViewUpstreamWarehouse() const {
     return impl_->blob_view_upstream_warehouse;
 }
 
+bool CoreOptions::BlobViewResolveEnabled() const {
+    return impl_->blob_view_resolve_enabled;
+}
+
 std::vector<std::string> CoreOptions::GetBlobInlineFields() const {
     std::vector<std::string> blob_inline_fields = impl_->blob_descriptor_fields;
     blob_inline_fields.insert(blob_inline_fields.end(), impl_->blob_view_fields.begin(),

src/paimon/core/core_options.h

@@ -197,6 +197,7 @@ class PAIMON_EXPORT CoreOptions {
     const std::vector<std::string>& GetBlobDescriptorFields() const;
     const std::vector<std::string>& GetBlobViewFields() const;
     std::optional<std::string> GetBlobViewUpstreamWarehouse() const;
+    bool BlobViewResolveEnabled() const;
     std::vector<std::string> GetBlobInlineFields() const;
 
     const std::map<std::string, std::string>& ToMap() const;

src/paimon/core/core_options_test.cpp

@@ -124,6 +124,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_TRUE(core_options.GetBlobViewFields().empty());
     ASSERT_TRUE(core_options.GetBlobInlineFields().empty());
     ASSERT_EQ(std::nullopt, core_options.GetBlobViewUpstreamWarehouse());
+    ASSERT_TRUE(core_options.BlobViewResolveEnabled());
     ASSERT_TRUE(core_options.LegacyPartitionNameEnabled());
     ASSERT_TRUE(core_options.GlobalIndexEnabled());
     ASSERT_EQ(std::nullopt, core_options.GetGlobalIndexExternalPath());
@@ -228,6 +229,7 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::BLOB_DESCRIPTOR_FIELD, "blob3,blob4"},
         {Options::BLOB_VIEW_FIELD, "blob5"},
         {Options::BLOB_VIEW_UPSTREAM_WAREHOUSE, "FILE:///tmp/blob_view_upstream_warehouse/"},
+        {Options::BLOB_VIEW_RESOLVE_ENABLED, "false"},
         {Options::PARTITION_GENERATE_LEGACY_NAME, "false"},
         {Options::GLOBAL_INDEX_ENABLED, "false"},
         {Options::GLOBAL_INDEX_THREAD_NUM, "4"},
@@ -368,6 +370,7 @@ TEST(CoreOptionsTest, TestFromMap) {
               std::vector<std::string>({"blob3", "blob4", "blob5"}));
     ASSERT_EQ(core_options.GetBlobViewUpstreamWarehouse(),
               std::optional<std::string>("FILE:///tmp/blob_view_upstream_warehouse/"));
+    ASSERT_FALSE(core_options.BlobViewResolveEnabled());
     ASSERT_FALSE(core_options.LegacyPartitionNameEnabled());
     ASSERT_FALSE(core_options.GlobalIndexEnabled());
     ASSERT_EQ(core_options.GetGlobalIndexThreadNum(), 4);

src/paimon/core/operation/data_evolution_split_read.cpp

@@ -167,6 +167,11 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateReader(
 Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::WrapWithBlobViewResolverIfNeeded(
     const std::shared_ptr<DataSplit>& data_split,
     std::unique_ptr<BatchReader>&& inner_reader) const {
+    if (!options_.BlobViewResolveEnabled()) {
+        // preserve serialized BlobViewStruct bytes, e.g. for forwarding blob view values to
+        // another blob-view table
+        return std::move(inner_reader);
+    }
     std::vector<std::string> read_blob_view_fields = HasBlobViewField(options_, raw_read_schema_);
     if (read_blob_view_fields.empty()) {
         return std::move(inner_reader);

test/inte/blob_table_inte_test.cpp

@@ -2500,6 +2500,101 @@ TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamTable) {
     }
 }
 
+TEST_P(BlobTableInteTest, TestBlobViewFieldResolveDisabled) {
+    auto file_format = GetParam();
+    if (file_format != "orc" && file_format != "parquet") {
+        return;
+    }
+
+    // With blob-view.resolve.enabled=false the read must pass the serialized BlobViewStruct
+    // bytes through without touching the upstream side: the referenced upstream table
+    // deliberately does not exist and no upstream warehouse is configured.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
+                                 BlobUtils::ToArrowField("view", true)};
+    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
+                                                  {Options::FILE_FORMAT, file_format},
+                                                  {Options::BUCKET, "-1"},
+                                                  {Options::ROW_TRACKING_ENABLED, "true"},
+                                                  {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                  {Options::BLOB_VIEW_FIELD, "view"},
+                                                  {Options::BLOB_VIEW_RESOLVE_ENABLED, "false"},
+                                                  {Options::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    Identifier upstream_identifier("upstream_db", "upstream_table");
+    arrow::LargeBinaryBuilder view_builder;
+    for (int32_t i = 0; i < 4; ++i) {
+        if (i == 2) {
+            ASSERT_TRUE(view_builder.AppendNull().ok());
+            continue;
+        }
+        BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6,
+                                   /*row_id=*/static_cast<int64_t>(i));
+        auto serialized = view_struct.Serialize(pool_);
+        ASSERT_TRUE(
+            view_builder
+                .Append(reinterpret_cast<const uint8_t*>(serialized->data()), serialized->size())
+                .ok());
+    }
+    std::shared_ptr<arrow::Array> write_view_array;
+    ASSERT_TRUE(view_builder.Finish(&write_view_array).ok());
+    auto write_f0_array =
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::int32(), R"([100,101,102,103])")
+            .ValueOrDie();
+    auto write_struct = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}),
+                                 std::vector<std::string>({"f0", "view"}))
+            .ValueOrDie());
+
+    // write & commit
+    auto schema = arrow::schema(fields);
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                         WriteArray(table_path, {}, schema->field_names(), {write_struct}));
+    ASSERT_OK(Commit(table_path, commit_msgs));
+
+    // scan & read
+    ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path));
+    ASSERT_OK_AND_ASSIGN(auto result,
+                         ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr));
+    ASSERT_TRUE(result.chunked_array);
+    auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie();
+    auto read_struct = std::dynamic_pointer_cast<arrow::StructArray>(read_concat);
+    ASSERT_EQ(read_struct->length(), 4);
+
+    auto read_f0_array = read_struct->GetFieldByName("f0");
+    ASSERT_TRUE(read_f0_array);
+    ASSERT_TRUE(read_f0_array->Equals(write_f0_array))
+        << "read f0:" << read_f0_array->ToString() << std::endl
+        << "written f0:" << write_f0_array->ToString();
+    auto read_view_array = read_struct->GetFieldByName("view");
+    ASSERT_TRUE(read_view_array);
+    ASSERT_TRUE(read_view_array->Equals(write_view_array))
+        << "read view:" << read_view_array->ToString() << std::endl
+        << "written view:" << write_view_array->ToString();
+
+    // The pass-through bytes still deserialize to the original view struct, so they can be
+    // forwarded to another blob-view table.
+    auto typed_view_array = std::dynamic_pointer_cast<arrow::LargeBinaryArray>(read_view_array);
+    ASSERT_TRUE(typed_view_array);
+    std::string_view first_view = typed_view_array->GetView(0);
+    ASSERT_OK_AND_ASSIGN(bool is_view_struct,
+                         BlobViewStruct::IsBlobViewStruct(first_view.data(), first_view.size()));
+    ASSERT_TRUE(is_view_struct);
+    ASSERT_OK_AND_ASSIGN(auto view_struct,
+                         BlobViewStruct::Deserialize(first_view.data(), first_view.size()));
+    ASSERT_TRUE(view_struct->GetIdentifier() == upstream_identifier);
+    ASSERT_EQ(view_struct->FieldId(), 6);
+    ASSERT_EQ(view_struct->RowId(), 0);
+
+    // Re-enabling resolve makes the same read fail on the missing upstream warehouse.
+    // Without this check the assertions above would also pass if resolution were never
+    // triggered at all.
+    ASSERT_NOK_WITH_MSG(ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr,
+                                  {{Options::BLOB_VIEW_RESOLVE_ENABLED, "true"}}),
+                        "BLOB_VIEW_UPSTREAM_WAREHOUSE");
+}
+
 TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamDescriptorBlob) {
     auto file_format = GetParam();
     if (GetParam() == "lance") {