refactor(blob-view): resolve blob view dependencies through catalog context

include/paimon/defs.h

@@ -577,12 +577,6 @@ struct PAIMON_EXPORT Options {
     /// 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" - Warehouse root the upstream table a blob-view field points
-    /// at is resolved from. Since the catalog capabilities are partially missing, cpp paimon cannot
-    /// always obtain it automatically, so it may be configured explicitly. When it is not set, the
-    /// read falls back to the warehouse the downstream table itself lives in, which resolves a
-    /// co-located upstream table. No default value.
-    static const char BLOB_VIEW_UPSTREAM_WAREHOUSE[];
     /// "blob-write-null-on-missing-file" - Whether to write NULL for a descriptor BLOB value,
     /// including an ARRAY<BLOB> element or a MAP<..., BLOB> value, when the referenced file does
     /// not exist at write time.

src/paimon/common/defs.cpp

@@ -154,7 +154,6 @@ 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::BLOB_WRITE_NULL_ON_MISSING_FILE[] = "blob-write-null-on-missing-file";
 const char Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE[] = "blob-write-null-on-fetch-failure";
 const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";

src/paimon/core/core_options.cpp

@@ -390,7 +390,6 @@ struct CoreOptions::Impl {
     std::optional<std::string> field_default_func;
     std::optional<std::string> scan_fallback_branch;
     std::optional<std::string> data_file_external_paths;
-    std::optional<std::string> blob_view_upstream_warehouse;
     std::optional<std::string> changelog_file_compression;
     std::optional<std::string> global_index_external_path;
     std::optional<std::string> scan_tag_name;
@@ -582,9 +581,6 @@ struct CoreOptions::Impl {
         PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(Options::BLOB_VIEW_FIELD,
                                                            Options::FIELDS_SEPARATOR,
                                                            &blob_view_fields, /*need_trim=*/true));
-        // 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));
@@ -1966,10 +1962,6 @@ const std::vector<std::string>& CoreOptions::GetBlobViewFields() const {
     return impl_->blob_view_fields;
 }
 
-std::optional<std::string> CoreOptions::GetBlobViewUpstreamWarehouse() const {
-    return impl_->blob_view_upstream_warehouse;
-}
-
 bool CoreOptions::BlobViewResolveEnabled() const {
     return impl_->blob_view_resolve_enabled;
 }

src/paimon/core/core_options.h

@@ -304,7 +304,6 @@ class PAIMON_EXPORT CoreOptions {
     const std::vector<std::string>& GetBlobFields() const;
     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;
 

src/paimon/core/core_options_test.cpp

@@ -175,7 +175,6 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_TRUE(core_options.GetBlobDescriptorFields().empty());
     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());
@@ -303,7 +302,6 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::BLOB_DESCRIPTOR_FIELD, "blob3,blob4"},
         {Options::BLOB_AS_DESCRIPTOR, "true"},
         {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"},
@@ -476,8 +474,6 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(core_options.GetBlobViewFields(), std::vector<std::string>({"blob5"}));
     ASSERT_EQ(core_options.GetBlobInlineFields(),
               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());

src/paimon/core/operation/data_evolution_split_read.cpp

@@ -34,7 +34,6 @@
 #include "arrow/array/array_binary.h"
 #include "arrow/array/array_nested.h"
 #include "arrow/c/bridge.h"
-#include "paimon/common/catalog/catalog_context.h"
 #include "paimon/common/data/blob_defs.h"
 #include "paimon/common/data/blob_utils.h"
 #include "paimon/common/data/blob_view_struct.h"
@@ -49,7 +48,6 @@
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/common/utils/object_utils.h"
-#include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/range_helper.h"
 #include "paimon/common/utils/vector_store_utils.h"
 #include "paimon/core/core_options.h"
@@ -282,33 +280,18 @@ Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::WrapWithBlobViewRes
     if (read_blob_view_fields.empty()) {
         return std::move(inner_reader);
     }
-    // The explicit upstream warehouse is optional. Absent it, fall back to the warehouse this
-    // (downstream) table itself lives in, which is where a co-located upstream table is resolved
-    // from; a metastore-backed catalog resolves the upstream location by identifier anyway, so the
-    // root is only a hint there. Stripping the trailing `<database>.db/<table>` (or
-    // `<database>/<table>`) components off this table's path yields that warehouse.
-    std::optional<std::string> warehouse_path = options_.GetBlobViewUpstreamWarehouse();
-    if (!warehouse_path) {
-        // TODO(yonghao.fyh): refactor warehouse path in the future
-        warehouse_path =
-            PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(context_->GetPath()));
-    }
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BatchReader> pre_reader,
                            CreateBlobViewReader(data_split, read_blob_view_fields, row_ranges));
     PAIMON_ASSIGN_OR_RAISE(std::unordered_set<BlobViewStruct> blob_view_structs,
                            ExtractBlobViewStructs(pre_reader.get()));
-    // The dependency table is read through a catalog of its own, told how to resolve a scheme but
-    // not handed this read's file system: that one authenticates as this table, and the table about
-    // to be read is a different one whose credentials the metastore issues only for it.
-    auto catalog_context = std::make_shared<CatalogContext>(
-        warehouse_path.value(), options_.ToMap(), context_->GetFileSystemSchemeToIdentifierMap());
     // use global thread number
     uint32_t cpu_count = std::thread::hardware_concurrency();
     uint32_t thread_num = cpu_count > 0 ? cpu_count : 1;
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Executor> executor, CreateDefaultExecutor(thread_num));
     PAIMON_ASSIGN_OR_RAISE(
         BlobViewResolver resolver,
-        BlobViewLookup::CreateResolver(blob_view_structs, catalog_context, pool_, executor));
+        BlobViewLookup::CreateResolver(
+            blob_view_structs, context_->CreateDependencyCatalogContext(), pool_, executor));
     return std::make_unique<BlobViewResolvingBatchReader>(std::move(inner_reader),
                                                           std::move(read_blob_view_fields),
                                                           std::move(resolver), arrow_pool_);

src/paimon/core/operation/internal_read_context.cpp

@@ -27,6 +27,9 @@
 #include "arrow/c/abi.h"
 #include "arrow/c/bridge.h"
 #include "fmt/format.h"
+#include "paimon/catalog/catalog.h"
+#include "paimon/catalog_options.h"
+#include "paimon/common/catalog/catalog_context.h"
 #include "paimon/common/data/variant/variant_access_utils.h"
 #include "paimon/common/data/variant/variant_type_utils.h"
 #include "paimon/common/predicate/full_text_match_predicate.h"
@@ -36,16 +39,76 @@
 #include "paimon/common/utils/arrow/mem_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/common/utils/url_utils.h"
 #include "paimon/core/options/map_storage_layout.h"
 #include "paimon/core/schema/arrow_schema_validator.h"
 #include "paimon/core/utils/nested_projection_utils.h"
 #include "paimon/predicate/compound_predicate.h"
 #include "paimon/predicate/full_text_search.h"
 #include "paimon/predicate/predicate_utils.h"
 #include "paimon/predicate/vector_search.h"
+#include "paimon/rest/rest_api.h"
 #include "paimon/status.h"
+#include "rapidjson/document.h"
+#include "rapidjson/stringbuffer.h"
+#include "rapidjson/writer.h"
 
 namespace paimon {
+namespace {
+
+std::string IdentifierToFlatJson(const Identifier& identifier) {
+    rapidjson::Document document(rapidjson::kObjectType);
+    auto& allocator = document.GetAllocator();
+    document.AddMember(rapidjson::StringRef("database"),
+                       rapidjson::Value(identifier.GetDatabaseName(), allocator), allocator);
+    // Keep branch and system-table suffixes in the reported object name.
+    document.AddMember(rapidjson::StringRef("object"),
+                       rapidjson::Value(identifier.GetObjectName(), allocator), allocator);
+    rapidjson::StringBuffer buffer;
+    rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+    document.Accept(writer);
+    return buffer.GetString();
+}
+
+}  // namespace
+
+std::shared_ptr<CatalogContext> InternalReadContext::CreateDependencyCatalogContext() const {
+    const auto& catalog = read_context_->GetCatalog();
+    if (!catalog) {
+        // Preserve path-only filesystem reads using <warehouse>/<database>.db/<table>.
+        return std::make_shared<CatalogContext>(
+            PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(GetPath())), options_.ToMap(),
+            GetFileSystemSchemeToIdentifierMap());
+    }
+
+    // Preserve the dependency catalog's effective settings: table options, then catalog
+    // options, then explicit read overrides. The upstream reader loads its own table options;
+    // ReadContextBuilder must not copy these catalog settings back into that read.
+    auto dependency_options = options_.ToMap();
+    const auto& catalog_options = catalog->GetOptions();
+    for (const auto& [key, value] : catalog_options) {
+        dependency_options[key] = value;
+    }
+    const auto& read_options = read_context_->GetOptions();
+    for (const auto& [key, value] : read_options) {
+        dependency_options[key] = value;
+    }
+    const std::string read_via_option =
+        std::string(RestApi::kHeaderOptionPrefix) + RestApi::kReadViaHeader;
+    auto metastore = dependency_options.find(CatalogOptions::METASTORE);
+    if (metastore != dependency_options.end() &&
+        StringUtils::ToLowerCase(metastore->second) == "rest" && read_context_->GetIdentifier() &&
+        dependency_options.count(read_via_option) == 0) {
+        // Only fill in a missing header when the merged options select a REST dependency catalog.
+        dependency_options[read_via_option] =
+            UrlUtils::EncodeString(IdentifierToFlatJson(*read_context_->GetIdentifier()));
+    }
+    // A REST warehouse identifies the catalog, independent of the table's storage location.
+    return std::make_shared<CatalogContext>(catalog->GetRootPath(), dependency_options,
+                                            GetFileSystemSchemeToIdentifierMap());
+}
 
 Result<std::shared_ptr<arrow::Field>> InternalReadContext::AlignReadFieldWithTableFieldIds(
     const std::shared_ptr<arrow::Field>& read_field,

src/paimon/core/operation/internal_read_context.h

@@ -35,6 +35,7 @@ class Schema;
 }  // namespace arrow
 
 namespace paimon {
+struct CatalogContext;
 class Executor;
 class MemoryPool;
 class Predicate;
@@ -80,6 +81,9 @@ class InternalReadContext {
     const std::string& GetPath() const {
         return read_context_->GetPath();
     }
+
+    std::shared_ptr<CatalogContext> CreateDependencyCatalogContext() const;
+
     const std::vector<std::string>& GetPartitionKeys() const {
         return table_schema_->PartitionKeys();
     }