feat: support specific fs in ReadContext & options in VectorSearch (#57)

* feat: support specific fs in ReadContext & update lumina lib

* feat: support user-defined options in VectorSearch

Author: lxy-9602

include/paimon/defs.h

@@ -105,8 +105,8 @@ struct PAIMON_EXPORT Options {
     /// Default value is local.
     static const char FILE_SYSTEM[];
 
-    /// "target-file-size" - Target size of a file. Default value is 128MB.
-    // TODO(yonghao.fyh): xinyu, change the default value to 128MB for primary key table.
+    /// "target-file-size" - Target size of a file. Default value is 256MB.
+    // TODO(xinyu.lxy): change the default value to 128MB for primary key table.
     static const char TARGET_FILE_SIZE[];
 
     /// "blob.target-file-size" - Target size of a blob file. Default is TARGET_FILE_SIZE.

include/paimon/global_index/io/global_index_file_reader.h

@@ -30,7 +30,7 @@ class PAIMON_EXPORT GlobalIndexFileReader {
 
     /// Opens an input stream for reading the specified global index file.
     virtual Result<std::unique_ptr<InputStream>> GetInputStream(
-        const std::string& file_name) const = 0;
+        const std::string& file_path) const = 0;
 };
 
 }  // namespace paimon

include/paimon/predicate/vector_search.h

@@ -16,7 +16,9 @@
 
 #pragma once
 #include <functional>
+#include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -35,16 +37,24 @@ struct PAIMON_EXPORT VectorSearch {
     /// @note Must be thread-safe.
     using PreFilter = std::function<bool(int64_t)>;
 
+    /// Enumeration of distance or similarity metrics for vector comparison.
+    enum class DistanceType { EUCLIDEAN = 1, INNER_PRODUCT = 2, COSINE = 3, UNKNOWN = 128 };
+
     VectorSearch(const std::string& _field_name, int32_t _limit, const std::vector<float>& _query,
-                 PreFilter _pre_filter, const std::shared_ptr<Predicate>& _predicate)
+                 PreFilter _pre_filter, const std::shared_ptr<Predicate>& _predicate,
+                 const std::optional<DistanceType>& _distance_type,
+                 const std::map<std::string, std::string>& _options)
         : field_name(_field_name),
           limit(_limit),
           query(_query),
           pre_filter(_pre_filter),
-          predicate(_predicate) {}
+          predicate(_predicate),
+          distance_type(_distance_type),
+          options(_options) {}
 
     std::shared_ptr<VectorSearch> ReplacePreFilter(PreFilter _pre_filter) const {
-        return std::make_shared<VectorSearch>(field_name, limit, query, _pre_filter, predicate);
+        return std::make_shared<VectorSearch>(field_name, limit, query, _pre_filter, predicate,
+                                              distance_type, options);
     }
 
     /// Search field name.
@@ -65,5 +75,13 @@ struct PAIMON_EXPORT VectorSearch {
     /// @note All fields referenced in the predicate must have been materialized
     ///       in the index during build to ensure availability.
     std::shared_ptr<Predicate> predicate;
+    /// The distance metric to use for this query, if explicitly specified.
+    /// If set, this value must match the distance type used by the index (e.g., EUCLIDEAN, COSINE).
+    /// A mismatch will result in an error during query execution.
+    /// If not set (std::nullopt), the query will use the distance type configured in the index.
+    std::optional<DistanceType> distance_type;
+    /// A key-value map of query-specific runtime options.
+    /// Such as the size of candidate list in approximate search or parallelism for this query.
+    std::map<std::string, std::string> options;
 };
 }  // namespace paimon

include/paimon/read_context.h

@@ -32,6 +32,7 @@ namespace paimon {
 class Executor;
 class MemoryPool;
 class Predicate;
+class FileSystem;
 
 /// `ReadContext` is some configuration for read operations.
 ///
@@ -48,6 +49,7 @@ class PAIMON_EXPORT ReadContext {
                 uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
                 const std::shared_ptr<MemoryPool>& memory_pool,
                 const std::shared_ptr<Executor>& executor,
+                const std::shared_ptr<FileSystem>& specific_file_system,
                 const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
                 const std::map<std::string, std::string>& options);
     ~ReadContext();
@@ -103,6 +105,9 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<Executor> GetExecutor() const {
         return executor_;
     }
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
 
  private:
     std::string path_;
@@ -118,6 +123,7 @@ class PAIMON_EXPORT ReadContext {
     std::optional<std::string> table_schema_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> fs_scheme_to_identifier_map_;
     std::map<std::string, std::string> options_;
 };
@@ -258,15 +264,29 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @note Default branch is "main" if not specified.
     ReadContextBuilder& WithBranch(const std::string& branch);
 
-    /// Set the file system scheme to identifier mapping for custom file system configurations.
-    /// This allows using different file system implementations for different URI schemes.
+    /// Sets a mapping from URI schemes (e.g., "file", "oss") to registered file system
+    /// identifiers. This allows selecting different pre-registered file system implementations
+    /// based on the URI scheme at runtime.
     ///
-    /// @param fs_scheme_to_identifier_map Map from URI scheme to file system identifier.
+    /// @param fs_scheme_to_identifier_map Map from URI scheme (like "oss") to the corresponding
+    /// file system identifier.
     /// @return Reference to this builder for method chaining.
-    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`).
+    /// @note
+    ///   - This method is intended for environments where multiple file systems are pre-registered.
+    ///   - The specified identifiers must correspond to file systems that have been registered at
+    ///   compile time or initialization.
+    ///   - Cannot be used together with `WithFileSystem()`.
+    ///   - If not set, use default file system (configured in `Options::FILE_SYSTEM`).
+    /// Example:
+    ///   builder.WithFileSystemSchemeToIdentifierMap({{"oss", "jindo"}, {"file", "local"}});
+    ///
     ReadContextBuilder& WithFileSystemSchemeToIdentifierMap(
         const std::map<std::string, std::string>& fs_scheme_to_identifier_map);
 
+    /// Sets a custom file system instance to be used for all file operations in this read context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    ReadContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
+
     /// Build and return a `ReadContext` instance with input validation.
     /// @return Result containing the constructed `ReadContext` or an error status.
     Result<std::unique_ptr<ReadContext>> Finish();

src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp

@@ -219,7 +219,8 @@ TEST_F(BitmapGlobalIndexTest, TestStringType) {
 
         // test visit vector search
         ASSERT_NOK_WITH_MSG(reader->VisitVectorSearch(std::make_shared<VectorSearch>(
-                                "f0", 10, std::vector<float>({1.0f, 2.0f}), nullptr, nullptr)),
+                                "f0", 10, std::vector<float>({1.0f, 2.0f}), nullptr, nullptr,
+                                std::nullopt, std::map<std::string, std::string>())),
                             "FileIndexReaderWrapper is not supposed to handle vector search query");
     };
 

src/paimon/core/operation/internal_read_context.cpp

@@ -34,7 +34,8 @@ Result<std::unique_ptr<InternalReadContext>> InternalReadContext::Create(
     const std::map<std::string, std::string>& options) {
     PAIMON_ASSIGN_OR_RAISE(
         CoreOptions core_options,
-        CoreOptions::FromMap(options, context->GetFileSystemSchemeToIdentifierMap()));
+        CoreOptions::FromMap(options, context->GetFileSystemSchemeToIdentifierMap(),
+                             context->GetSpecificFileSystem()));
     // prepare read schema
     std::vector<DataField> read_data_fields;
     read_data_fields.reserve(context->GetReadSchema().size());

src/paimon/core/operation/read_context.cpp

@@ -36,6 +36,7 @@ ReadContext::ReadContext(const std::string& path, const std::string& branch,
                          const std::optional<std::string>& table_schema,
                          const std::shared_ptr<MemoryPool>& memory_pool,
                          const std::shared_ptr<Executor>& executor,
+                         const std::shared_ptr<FileSystem>& specific_file_system,
                          const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
                          const std::map<std::string, std::string>& options)
     : path_(path),
@@ -51,6 +52,7 @@ ReadContext::ReadContext(const std::string& path, const std::string& branch,
       table_schema_(table_schema),
       memory_pool_(memory_pool),
       executor_(executor),
+      specific_file_system_(specific_file_system),
       fs_scheme_to_identifier_map_(fs_scheme_to_identifier_map),
       options_(options) {}
 
@@ -74,6 +76,7 @@ class ReadContextBuilder::Impl {
         table_schema_ = std::nullopt;
         memory_pool_ = GetDefaultPool();
         executor_.reset();
+        specific_file_system_.reset();
     }
 
  private:
@@ -92,6 +95,7 @@ class ReadContextBuilder::Impl {
     std::optional<std::string> table_schema_;
     std::shared_ptr<MemoryPool> memory_pool_ = GetDefaultPool();
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
 };
 
 ReadContextBuilder::ReadContextBuilder(const std::string& path)
@@ -180,6 +184,12 @@ ReadContextBuilder& ReadContextBuilder::WithFileSystemSchemeToIdentifierMap(
     return *this;
 }
 
+ReadContextBuilder& ReadContextBuilder::WithFileSystem(
+    const std::shared_ptr<FileSystem>& file_system) {
+    impl_->specific_file_system_ = file_system;
+    return *this;
+}
+
 Result<std::unique_ptr<ReadContext>> ReadContextBuilder::Finish() {
     PAIMON_ASSIGN_OR_RAISE(impl_->path_, PathUtil::NormalizePath(impl_->path_));
     if (impl_->path_.empty()) {
@@ -207,7 +217,8 @@ Result<std::unique_ptr<ReadContext>> ReadContextBuilder::Finish() {
         impl_->enable_predicate_filter_, impl_->enable_prefetch_, impl_->prefetch_batch_count_,
         impl_->prefetch_max_parallel_num_, impl_->enable_multi_thread_row_to_batch_,
         impl_->row_to_batch_thread_number_, impl_->table_schema_, impl_->memory_pool_,
-        impl_->executor_, impl_->fs_scheme_to_identifier_map_, impl_->options_);
+        impl_->executor_, impl_->specific_file_system_, impl_->fs_scheme_to_identifier_map_,
+        impl_->options_);
     impl_->Reset();
     return ctx;
 }

src/paimon/core/operation/read_context_test.cpp

@@ -22,6 +22,7 @@
 #include "paimon/defs.h"
 #include "paimon/predicate/predicate_builder.h"
 #include "paimon/status.h"
+#include "paimon/testing/mock/mock_file_system.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -42,6 +43,7 @@ TEST(ReadContextTest, TestSimple) {
     ASSERT_EQ(1, ctx->GetRowToBatchThreadNumber());
     ASSERT_EQ("main", ctx->GetBranch());
     ASSERT_TRUE(ctx->GetFileSystemSchemeToIdentifierMap().empty());
+    ASSERT_FALSE(ctx->GetSpecificFileSystem());
 }
 
 TEST(ReadContextTest, TestSetContent) {
@@ -59,6 +61,8 @@ TEST(ReadContextTest, TestSetContent) {
     builder.SetRowToBatchThreadNumber(9);
     builder.WithBranch("rt");
     builder.WithFileSystemSchemeToIdentifierMap({{"file", "local"}});
+    auto fs = std::make_shared<MockFileSystem>();
+    builder.WithFileSystem(fs);
     ASSERT_OK_AND_ASSIGN(auto ctx, builder.Finish());
 
     // test result
@@ -78,6 +82,7 @@ TEST(ReadContextTest, TestSetContent) {
     ASSERT_EQ(expected_fs_map, ctx->GetFileSystemSchemeToIdentifierMap());
     std::map<std::string, std::string> expected_options = {{"key", "value"}};
     ASSERT_EQ(expected_options, ctx->GetOptions());
+    ASSERT_EQ(ctx->GetSpecificFileSystem(), fs);
 }
 
 }  // namespace paimon::test

src/paimon/core/operation/scan_context_test.cpp

@@ -50,7 +50,9 @@ TEST(ScanContextTest, TestSetFilter) {
     builder.SetPredicate(predicate);
     std::vector<float> query = {1.0, 2.0};
     VectorSearch::PreFilter pre_filter = [](int64_t id) -> bool { return id % 2; };
-    builder.SetVectorSearch(std::make_shared<VectorSearch>("f0", 10, query, pre_filter, nullptr));
+    builder.SetVectorSearch(std::make_shared<VectorSearch>(
+        "f0", 10, query, pre_filter, nullptr, VectorSearch::DistanceType::INNER_PRODUCT,
+        std::map<std::string, std::string>()));
     std::vector<Range> row_ranges = {Range(1, 2), Range(4, 5)};
     auto global_index_result = BitmapGlobalIndexResult::FromRanges(row_ranges);
     builder.SetGlobalIndexResult(global_index_result);
@@ -67,6 +69,8 @@ TEST(ScanContextTest, TestSetFilter) {
     auto result_vector_search = ctx->GetScanFilters()->GetVectorSearch();
     ASSERT_TRUE(result_vector_search);
     ASSERT_EQ(query, result_vector_search->query);
+    ASSERT_EQ(VectorSearch::DistanceType::INNER_PRODUCT,
+              result_vector_search->distance_type.value());
     ASSERT_EQ(partition_filters, ctx->GetScanFilters()->GetPartitionFilters());
     ASSERT_EQ("{1,2,4,5}", ctx->GetGlobalIndexResult()->ToString());
     std::map<std::string, std::string> expected_options = {{"key", "value"}};

src/paimon/core/table/source/table_read.cpp

@@ -57,7 +57,8 @@ Result<std::unique_ptr<InternalReadContext>> CreateInternalReadContext(
     } else {
         PAIMON_ASSIGN_OR_RAISE(
             CoreOptions tmp_core_options,
-            CoreOptions::FromMap(tmp_options, context->GetFileSystemSchemeToIdentifierMap()));
+            CoreOptions::FromMap(tmp_options, context->GetFileSystemSchemeToIdentifierMap(),
+                                 context->GetSpecificFileSystem()));
         SchemaManager schema_manager(tmp_core_options.GetFileSystem(), context->GetPath(), branch);
         PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
                                schema_manager.Latest());