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

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,166 @@ TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamTable) {
     }
 }
 
+TEST_P(BlobTableInteTest, TestForwardBlobViewReference) {
+    auto file_format = GetParam();
+    if (file_format != "orc" && file_format != "parquet") {
+        return;
+    }
+
+    // Forward blob view references between two blob-view tables: read the source table with
+    // resolve dynamically disabled, write the preserved BlobViewStruct bytes into the target
+    // table, then verify the target still stores the original upstream references and a
+    // default read resolves them to the actual upstream blob values.
+    const std::string upstream_db_name = "append_table_with_multi_blob";
+    const std::string upstream_table_name = "append_table_with_multi_blob";
+    std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name +
+                              ".db/" + upstream_table_name;
+    std::string dst_db_path =
+        PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name);
+    ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path));
+
+    // The source table has no upstream warehouse configured: only a read with resolve
+    // dynamically disabled can succeed on it.
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32()),
+                                 BlobUtils::ToArrowField("view", true)};
+    std::map<std::string, std::string> source_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::FILE_SYSTEM, "local"}};
+    CreateTable(fields, /*partition_keys=*/{}, source_options);
+    std::string source_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+    // The target table configures the upstream warehouse for resolving forwarded references.
+    std::map<std::string, std::string> target_options = source_options;
+    target_options[Options::BLOB_VIEW_UPSTREAM_WAREHOUSE] = dir_->Str();
+    auto schema = arrow::schema(fields);
+    ::ArrowSchema c_target_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*schema, &c_target_schema).ok());
+    ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(dir_->Str(), {}));
+    ASSERT_OK(catalog->CreateTable(Identifier("foo", "bar_forward"), &c_target_schema,
+                                   /*partition_keys=*/{}, /*primary_keys=*/{}, target_options,
+                                   /*ignore_if_exists=*/false));
+    std::string target_table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar_forward");
+
+    // src array
+    Identifier upstream_identifier(upstream_db_name, upstream_table_name);
+    arrow::LargeBinaryBuilder view_builder;
+    for (int32_t i = 0; i < 8; ++i) {
+        if (i < 6) {
+            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());
+        } else {
+            ASSERT_TRUE(view_builder.AppendNull().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,104,105,106,107])")
+                              .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 into the source table
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs,
+                         WriteArray(source_table_path, {}, schema->field_names(), {write_struct}));
+    ASSERT_OK(Commit(source_table_path, commit_msgs));
+
+    // A default read fails on the missing upstream warehouse: the pass-through below is
+    // enabled by the dynamic option alone.
+    ASSERT_OK_AND_ASSIGN(auto source_plan, ScanTable(source_table_path));
+    ASSERT_NOK_WITH_MSG(
+        ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr),
+        "BLOB_VIEW_UPSTREAM_WAREHOUSE");
+
+    ASSERT_OK_AND_ASSIGN(
+        auto source_result,
+        ReadTable(source_table_path, schema->field_names(), source_plan, /*predicate=*/nullptr,
+                  {{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
+    ASSERT_TRUE(source_result.chunked_array);
+    auto source_concat = arrow::Concatenate(source_result.chunked_array->chunks()).ValueOrDie();
+    auto source_struct = std::dynamic_pointer_cast<arrow::StructArray>(source_concat);
+    ASSERT_EQ(source_struct->length(), 8);
+    auto forward_f0_array = source_struct->GetFieldByName("f0");
+    ASSERT_TRUE(forward_f0_array);
+    auto forward_view_array = source_struct->GetFieldByName("view");
+    ASSERT_TRUE(forward_view_array);
+    ASSERT_TRUE(forward_view_array->Equals(write_view_array))
+        << "source view:" << forward_view_array->ToString() << std::endl
+        << "written view:" << write_view_array->ToString();
+
+    // Forward the preserved references into the target blob-view table.
+    auto forward_struct = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::StructArray::Make(arrow::ArrayVector({forward_f0_array, forward_view_array}),
+                                 std::vector<std::string>({"f0", "view"}))
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(
+        auto forward_commit_msgs,
+        WriteArray(target_table_path, {}, schema->field_names(), {forward_struct}));
+    ASSERT_OK(Commit(target_table_path, forward_commit_msgs));
+
+    // The target table stores the original upstream references byte-identically.
+    ASSERT_OK_AND_ASSIGN(auto target_plan, ScanTable(target_table_path));
+    ASSERT_OK_AND_ASSIGN(
+        auto raw_target_result,
+        ReadTable(target_table_path, schema->field_names(), target_plan, /*predicate=*/nullptr,
+                  {{Options::BLOB_VIEW_RESOLVE_ENABLED, "false"}}));
+    ASSERT_TRUE(raw_target_result.chunked_array);
+    auto raw_target_concat =
+        arrow::Concatenate(raw_target_result.chunked_array->chunks()).ValueOrDie();
+    auto raw_target_struct = std::dynamic_pointer_cast<arrow::StructArray>(raw_target_concat);
+    ASSERT_EQ(raw_target_struct->length(), 8);
+    auto raw_target_view_array = raw_target_struct->GetFieldByName("view");
+    ASSERT_TRUE(raw_target_view_array);
+    ASSERT_TRUE(raw_target_view_array->Equals(write_view_array))
+        << "target view:" << raw_target_view_array->ToString() << std::endl
+        << "written view:" << write_view_array->ToString();
+
+    // A default read of the target table resolves to the actual upstream blob values.
+    ASSERT_OK_AND_ASSIGN(auto resolved_result,
+                         ReadTable(target_table_path, schema->field_names(), target_plan,
+                                   /*predicate=*/nullptr));
+    ASSERT_TRUE(resolved_result.chunked_array);
+    auto resolved_concat = arrow::Concatenate(resolved_result.chunked_array->chunks()).ValueOrDie();
+    auto resolved_struct = std::dynamic_pointer_cast<arrow::StructArray>(resolved_concat);
+    ASSERT_EQ(resolved_struct->length(), 8);
+    ASSERT_OK_AND_ASSIGN(auto resolved, ConvertDescriptorToRawBlob(resolved_struct, {"view"}));
+
+    std::string padding_b(2048, 'b');
+    std::string padding_d(2048, 'd');
+    std::string padding_e(2048, 'e');
+    std::string padding_f(2048, 'f');
+    // clang-format off
+    std::string expected_json = R"([
+[100, null],
+[101, ")" + padding_b + R"("],
+[102, null],
+[103, ")" + padding_d + R"("],
+[104, ")" + padding_e + R"("],
+[105, ")" + padding_f + R"("],
+[106, null],
+[107, null]
+])";
+    // clang-format on
+    auto expected_struct = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json)
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct));
+    ASSERT_TRUE(resolved->Equals(expected_with_rk))
+        << "resolved:" << resolved->ToString() << std::endl
+        << "expected:" << expected_with_rk->ToString();
+}
+
 TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamDescriptorBlob) {
     auto file_format = GetParam();
     if (GetParam() == "lance") {