fix

Author: lucasfang

docs/source/user_guide/catalog.rst

@@ -169,6 +169,23 @@ it; writers also allow ``WithFileSystemSchemeToIdentifierMap`` to override
 file-system selection. For format tables, use ``WriteContextBuilder(FormatTable)``
 instead.
 
+A file system passed to ``WithFileSystem`` is used exactly as it is, so a custom
+one that signs each request itself stays authenticated by drawing the table's
+temporary credentials from ``Catalog::GetTableCredentialProvider``: hold the provider
+and call ``CredentialProvider::GetCredentials`` at each access, which returns
+credentials that are still valid and reloads them before they expire. A catalog
+that issues no per-table credentials returns a null provider, and the custom file
+system's own configuration authenticates the access.
+
+.. code-block:: cpp
+
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::CredentialProvider> provider,
+                          shared_catalog->GetTableCredentialProvider(paimon::Identifier("db", "tbl")));
+   // MyFileSystem merges provider->GetCredentials() over its own options per access.
+   auto my_fs = std::make_shared<MyFileSystem>(provider);
+   write_builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"))
+       .WithFileSystem(my_fs);
+
 The snapshot being committed carries a uuid generated on the client, and the
 commit names the snapshot it is based on by that snapshot's uuid, so the server
 can tell a commit that raced another. A commit which lost such a race is
@@ -235,6 +252,52 @@ so its restored file references are visible to the expiration operation.
    A branch the catalog holds without that directory, or one created while
    expiration runs, is not found: do not expire a table with such a branch, and
    serialize branch creation and expiration through the upstream coordinator.
+=======
+Reading through the catalog
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Pass the catalog and table identifier to ``ReadContextBuilder::WithCatalog`` or
+``ScanContextBuilder::WithCatalog`` to read a native table the same way, in one
+call: the table's schema comes from the catalog, its data is read through the
+table's file system, and a scan reads the latest snapshot from the catalog:
+
+.. code-block:: cpp
+
+   // Readers and scanners share ownership of the catalog.
+   std::shared_ptr<paimon::Catalog> shared_catalog(std::move(catalog));
+
+   paimon::ScanContextBuilder scan_builder(table_path);
+   scan_builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"));
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ScanContext> scan_context,
+                          scan_builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableScan> scan,
+                          paimon::TableScan::Create(std::move(scan_context)));
+
+   paimon::ReadContextBuilder read_builder(table_path);
+   read_builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"));
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ReadContext> read_context,
+                          read_builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableRead> read,
+                          paimon::TableRead::Create(std::move(read_context)));
+
+A scan takes only its latest snapshot from the catalog; the manifests that
+snapshot points at and any historical snapshots still come from the table path.
+Both builders read the table's schema from the catalog when ``Finish()`` builds
+the context, which is one catalog request; call ``SetTableSchema`` with a schema
+you already hold to skip it. The table's file system, including the per-table
+temporary credentials a catalog serves through ``Catalog::GetTableFileSystem``,
+is used for both. ``WithFileSystem`` overrides it, and a custom file system
+that signs its own requests can draw the table's credentials from
+``Catalog::GetTableCredentialProvider`` as shown above; the read builder also allows
+``WithFileSystemSchemeToIdentifierMap``.
+
+Only the main branch is supported. Both builders reject other branches in the
+identifier or ``branch`` option; the read builder also checks ``WithBranch``.
+Explicit ``tbl$branch_main`` and ``branch=main`` are accepted. For a format
+table, use ``ReadContextBuilder(FormatTable)`` or ``ScanContextBuilder(FormatTable)``
+instead; ``WithCatalog`` is rejected. A system table path such as ``tbl$snapshots``
+reads snapshots from under the table path, which a version-managed catalog
+publishes nowhere but in itself, so it is rejected with ``NotImplemented``; scan
+such a table with ``WithFileSystem(Catalog::GetTableFileSystem(...))`` instead.
 
 The C++ REST catalog covers the database, table, snapshot and commit operations
 of the ``Catalog`` API. The parts of the Java REST catalog that have no C++
@@ -298,3 +361,66 @@ identifier. The request body carries the table id but no table name and no
 branch, so the branch has to appear in the URL the caller sends that body to:
 the commit endpoint of ``tbl$branch_dev``, not the one of ``tbl``. A body sent
 to the bare table's URL publishes the branch's snapshot on the main branch.
+=======
+tags, branch management and consumers — are not supported.
+
+Authenticating with credentials of your own
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Credentials that a catalog does not issue — from a token service of your own, or
+from a catalog Paimon has no client for — reach the built-in file systems through
+a ``CredentialProvider``. Implement it together with a
+``CredentialProviderFactory`` that builds it, register the factory with
+``REGISTER_PAIMON_FACTORY``, and name it in the ``fs.credential.provider``
+option: the file system then builds the provider once and asks it for credentials
+as it signs, so credentials are reloaded as they expire without the file system
+being rebuilt.
+
+.. code-block:: cpp
+
+   class MyCredentialProvider : public paimon::CredentialProvider {
+    public:
+       // Called as each access is signed; reload the credentials before they expire.
+       paimon::Result<std::map<std::string, std::string>> GetCredentials() const override {
+           return std::map<std::string, std::string>{
+               {"fs.oss.accessKeyId", ...}, {"fs.oss.accessKeySecret", ...},
+               {"fs.oss.securityToken", ...}};
+       }
+   };
+
+   class MyCredentialProviderFactory : public paimon::CredentialProviderFactory {
+    public:
+       const char* Identifier() const override {
+           return "my-token-service";
+       }
+
+       paimon::Result<std::shared_ptr<paimon::CredentialProvider>> Create(
+           const std::string& path,
+           const std::map<std::string, std::string>& options) const override {
+           return std::make_shared<MyCredentialProvider>(path, options);
+       }
+   };
+
+   REGISTER_PAIMON_FACTORY(MyCredentialProviderFactory);
+
+The credentials are file-system options, e.g. ``fs.oss.securityToken``, and are
+merged over the options the file system was configured with, so a provider serves
+only what rotates and the endpoint and the rest stay in the options. The options
+the factory is called with are those of the accesses to authenticate, which is
+where an implementation reads its own configuration, such as the address of the
+token service, from.
+
+.. code-block:: none
+
+   fs.credential.provider=my-token-service
+
+Only the OSS file system consults a provider today; the others report
+``NotImplemented`` when the option names one, rather than signing with the
+credentials of the moment they were built and failing once those expire. All
+factories share one identifier space, so an identifier a file system factory
+already takes — ``oss``, ``s3``, ``local``, ``jindo`` — would replace it; name the
+provider after where its credentials come from instead.
+
+This is the other of the two ways a provider is reached. A custom file system
+passed to ``WithFileSystem`` signs its own requests and pulls from the provider
+itself, which is what ``Catalog::GetTableCredentialProvider`` serves; a built-in
+file system signs on your behalf and is pointed at a provider by this option.

include/paimon/catalog/catalog.h

@@ -192,7 +192,9 @@ class PAIMON_EXPORT Catalog {
     /// @note A catalog that hands out per-table temporary credentials returns a file
     ///       system that refreshes them, so the returned instance must be used for the
     ///       table it was requested for. Pass it to `ReadContextBuilder::WithFileSystem`,
-    ///       `ScanContextBuilder::WithFileSystem` or `WriteContextBuilder::WithFileSystem`.
+    ///       `ScanContextBuilder::WithFileSystem` or `WriteContextBuilder::WithFileSystem`,
+    ///       or let a builder request it for you through `ReadContextBuilder::WithCatalog`,
+    ///       `ScanContextBuilder::WithCatalog` or `WriteContextBuilder::WithCatalog`.
     ///
     /// @param identifier The identifier (database and table name) of the table.
     /// @return A shared pointer to the file system instance; the catalog-level file system

include/paimon/fs/credential_provider.h

@@ -0,0 +1,54 @@
+/*
+ * 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 <map>
+#include <string>
+
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// The source of the credentials a file system authenticates its accesses with.
+///
+/// Implement this to hand out credentials that expire: the provider is asked again at
+/// every access, so it can reload them before they expire without the file system being
+/// torn down and built again.
+///
+/// A provider is for a caller that brings its own `FileSystem`. Build one with
+/// `CredentialProviderFactory::Get`, consult it from every access of your file system, and
+/// pass that file system in through `Catalog::Create`, `ReadContextBuilder::WithFileSystem`,
+/// `ScanContextBuilder::WithFileSystem` or `WriteContextBuilder::WithFileSystem`. A file
+/// system passed this way is used as-is; the built-in file systems are not involved and
+/// authenticate with the static credentials of their own options.
+///
+/// An implementation is consulted from every thread that accesses the file system and
+/// has to be thread-safe.
+class PAIMON_EXPORT CredentialProvider {
+ public:
+    virtual ~CredentialProvider() = default;
+
+    /// Returns the credentials to sign an access with, reloading them when they are about
+    /// to expire. The keys are file system options, e.g. "fs.oss.accessKeyId" or
+    /// "fs.oss.securityToken", so a caller merges them over its own file system options.
+    virtual Result<std::map<std::string, std::string>> GetCredentials() const = 0;
+};
+
+}  // namespace paimon

include/paimon/fs/credential_provider_factory.h

@@ -0,0 +1,65 @@
+/*
+ * 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 <map>
+#include <memory>
+#include <string>
+
+#include "paimon/factories/factory.h"
+#include "paimon/fs/credential_provider.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A factory for creating `CredentialProvider` instances.
+///
+/// Register an implementation with `REGISTER_PAIMON_FACTORY` to authenticate accesses
+/// with credentials of your own making: the built-in file systems build the provider
+/// named by their `fs.credential.provider` option and consult it whenever they sign.
+///
+/// @note All factories share one identifier space, so an identifier that a file system
+/// factory already takes - "oss", "s3", "local", "jindo" - would replace it. Name the
+/// provider after where its credentials come from instead.
+class PAIMON_EXPORT CredentialProviderFactory : public Factory {
+ public:
+    /// The option naming the factory a file system builds its provider with.
+    static const char CREDENTIAL_PROVIDER_OPTION[];
+
+    /// Create a `CredentialProvider` of current factory for the accesses below a path.
+    ///
+    /// The options are the file system options of the accesses to authenticate, so an
+    /// implementation reads its own configuration out of them.
+    virtual Result<std::shared_ptr<CredentialProvider>> Create(
+        const std::string& path, const std::map<std::string, std::string>& options) const = 0;
+
+    /// Get a `CredentialProvider` corresponding to identifier for the accesses below a path.
+    /// @pre Factory is already registered.
+    static Result<std::shared_ptr<CredentialProvider>> Get(
+        const std::string& identifier, const std::string& path,
+        const std::map<std::string, std::string>& fs_options);
+
+    /// Get the `CredentialProvider` the options ask for, or a null provider when they name
+    /// none, in which case the file system options themselves authenticate the accesses.
+    static Result<std::shared_ptr<CredentialProvider>> GetIfConfigured(
+        const std::string& path, const std::map<std::string, std::string>& fs_options);
+};
+
+}  // namespace paimon

include/paimon/read_context.h

@@ -27,13 +27,15 @@
 
 #include "arrow/c/abi.h"
 #include "paimon/cache/cache.h"
+#include "paimon/catalog/identifier.h"
 #include "paimon/predicate/predicate.h"
 #include "paimon/result.h"
 #include "paimon/type_fwd.h"
 #include "paimon/utils/prefetch_cache_config.h"
 #include "paimon/visibility.h"
 
 namespace paimon {
+class Catalog;
 class Executor;
 class FormatTable;
 class MemoryPool;
@@ -62,7 +64,9 @@ class PAIMON_EXPORT ReadContext {
                 const std::shared_ptr<RealtimeContext>& realtime_context,
                 const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
                 const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
-                const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level);
+                const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level,
+                const std::shared_ptr<Catalog>& catalog,
+                const std::optional<Identifier>& identifier);
 
     ~ReadContext();
 
@@ -115,6 +119,8 @@ class PAIMON_EXPORT ReadContext {
     uint32_t GetRowToBatchThreadNumber() const {
         return row_to_batch_thread_number_;
     }
+    /// The schema this context reads with, or null when it has to be read from the table path.
+    /// `WithCatalog()` resolves it to the catalog's schema when `SetTableSchema()` did not answer.
     const std::optional<std::string>& GetSpecificTableSchema() const {
         return table_schema_;
     }
@@ -155,6 +161,16 @@ class PAIMON_EXPORT ReadContext {
         return warmup_level_;
     }
 
+    /// Returns the catalog supplying table metadata, or null if unset.
+    const std::shared_ptr<Catalog>& GetCatalog() const {
+        return catalog_;
+    }
+
+    /// Returns the table identifier supplied with the catalog.
+    const std::optional<Identifier>& GetIdentifier() const {
+        return identifier_;
+    }
+
     /// Whether a read schema (C ArrowSchema) for nested column pruning was provided.
     bool HasReadSchema() const {
         return read_schema_ != nullptr && read_schema_->release != nullptr;
@@ -196,6 +212,8 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<Cache> cache_;
     std::shared_ptr<FormatTable> format_table_;
     WarmupLevel warmup_level_;
+    std::shared_ptr<Catalog> catalog_;
+    std::optional<Identifier> identifier_;
     // Owns schema resources and releases ArrowSchema::release in destructor.
     std::unique_ptr<ArrowSchema> read_schema_;
 };
@@ -210,8 +228,8 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// Constructs a `ReadContextBuilder` for a format table that is already loaded: the only way
     /// to read one whose schema lives in a metastore rather than under its location, such as a
     /// table a REST catalog serves. The table carries what such a location does not say, so
-    /// `SetTableSchema()`, `WithFileSystem()`, `WithFileSystemSchemeToIdentifierMap()` and a
-    /// branch are refused here rather than ignored.
+    /// `SetTableSchema()`, `WithFileSystem()`, `WithFileSystemSchemeToIdentifierMap()`,
+    /// `WithCatalog()` and a branch are refused here rather than ignored.
     ///
     /// @param table The format table to read, as `Catalog::GetFormatTable()` hands it back.
     explicit ReadContextBuilder(const std::shared_ptr<FormatTable>& table);
@@ -478,6 +496,19 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
     ReadContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
+    /// Loads a native table's schema from the catalog and reads its data through that table's file
+    /// system - including the per-table temporary credentials a catalog that issues them hands out
+    /// through `Catalog::GetTableFileSystem`.
+    ///
+    /// Only the main branch is supported. `SetTableSchema()` keeps the caller's schema and spares
+    /// the catalog request; `WithFileSystem()` and `WithFileSystemSchemeToIdentifierMap()` override
+    /// the file system.
+    /// @param catalog Non-null catalog, read when `Finish()` builds the context.
+    /// @param identifier The native table to read.
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& WithCatalog(const std::shared_ptr<Catalog>& catalog,
+                                    const Identifier& identifier);
+
     /// Inject a cache for read operations. Passing nullptr disables cache.
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& WithCache(const std::shared_ptr<Cache>& cache);