core_options.cpp

/*
 * Copyright 2024-present Alibaba Inc.
 *
 * Licensed 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.
 */

#include "paimon/core/core_options.h"

#include <cstring>
#include <limits>
#include <memory>
#include <optional>
#include <utility>

#include "fmt/format.h"
#include "paimon/common/fs/resolving_file_system.h"
#include "paimon/common/options/memory_size.h"
#include "paimon/common/options/time_duration.h"
#include "paimon/common/utils/options_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/string_utils.h"
#include "paimon/core/options/expire_config.h"
#include "paimon/core/options/lookup_strategy.h"
#include "paimon/core/options/sort_order.h"
#include "paimon/core/utils/branch_manager.h"
#include "paimon/defs.h"
#include "paimon/format/file_format.h"
#include "paimon/format/file_format_factory.h"
#include "paimon/status.h"

namespace paimon {

// ConfigParser is a helper class for parsing configurations from a map of strings.
class ConfigParser {
 public:
    explicit ConfigParser(const std::map<std::string, std::string>& map) : config_map_(map) {}

    // Parse basic type configurations
    template <typename T>
    Status Parse(const std::string& key, T* value) const {
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            auto result = StringUtils::StringToValue<T>(iter->second);
            if (result) {
                *value = result.value();
                return Status::OK();
            }
            return Status::Invalid(fmt::format("Invalid Config [{}: {}]", key, iter->second));
        }
        return Status::OK();  // Return success even if the configuration does not exist
    }

    // Parse optional basic type configurations
    template <typename T>
    Status Parse(const std::string& key, std::optional<T>* value) const {
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            auto result = StringUtils::StringToValue<T>(iter->second);
            if (result) {
                *value = result.value();
                return Status::OK();
            }
            return Status::Invalid(fmt::format("Invalid Config [{}: {}]", key, iter->second));
        }
        return Status::OK();  // Return success even if the configuration does not exist
    }

    // Parse list configurations
    template <typename T>
    Status ParseList(const std::string& key, const std::string& delimiter, std::vector<T>* list,
                     bool need_trim = false) const {
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            auto value_str_vec = StringUtils::Split(iter->second, delimiter, /*ignore_empty=*/true);
            for (auto& value_str : value_str_vec) {
                if (need_trim) {
                    StringUtils::Trim(&value_str);
                }
                if constexpr (std::is_same_v<T, std::string>) {
                    list->emplace_back(value_str);
                } else {
                    auto value = StringUtils::StringToValue<T>(value_str);
                    if (!value) {
                        return Status::Invalid(
                            fmt::format("Invalid Config [{}: {}]", key, iter->second));
                    }
                    list->emplace_back(value.value());
                }
            }
        }
        return Status::OK();  // Return success even if the configuration does not exist
    }

    // Parse memory size configurations
    template <typename T>
    Status ParseMemorySize(const std::string& key, T* value) const {
        static_assert(std::is_same_v<T, int64_t> || std::is_same_v<T, std::optional<int64_t>>,
                      "ParseMemorySize only supports int64_t and std::optional<int64_t>");
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            PAIMON_ASSIGN_OR_RAISE(*value, MemorySize::ParseBytes(iter->second));
        }
        return Status::OK();
    }

    // Parse time duration configurations
    template <typename T>
    Status ParseTimeDuration(const std::string& key, T* value) const {
        static_assert(std::is_same_v<T, int64_t> || std::is_same_v<T, std::optional<int64_t>>,
                      "ParseTimeDuration only supports int64_t and std::optional<int64_t>");
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            PAIMON_ASSIGN_OR_RAISE(*value, TimeDuration::Parse(iter->second));
        }
        return Status::OK();
    }

    // Parse object configurations
    template <typename Factory, typename ObjectType>
    Status ParseObject(const std::string& key, const std::string& default_identifier,
                       std::shared_ptr<ObjectType>* value) const {
        auto iter = config_map_.find(key);
        if (iter != config_map_.end()) {
            std::string normalized_value = StringUtils::ToLowerCase(iter->second);
            PAIMON_ASSIGN_OR_RAISE(*value, Factory::Get(normalized_value, config_map_));
        } else {
            PAIMON_ASSIGN_OR_RAISE(
                *value, Factory::Get(StringUtils::ToLowerCase(default_identifier), config_map_));
        }
        return Status::OK();
    }

    // Parse file system
    Status ParseFileSystem(const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
                           const std::shared_ptr<FileSystem>& specified_file_system,
                           std::shared_ptr<FileSystem>* value) const {
        if (specified_file_system) {
            // if exists user specified file system, first use
            *value = specified_file_system;
            return Status::OK();
        }
        std::string default_fs_identifier = "local";
        auto iter = config_map_.find(Options::FILE_SYSTEM);
        if (iter != config_map_.end()) {
            default_fs_identifier = StringUtils::ToLowerCase(iter->second);
        }
        *value = std::make_shared<ResolvingFileSystem>(fs_scheme_to_identifier_map,
                                                       default_fs_identifier, config_map_);
        return Status::OK();
    }

    // Parse SortOrder
    Status ParseSortOrder(SortOrder* sort_order) const {
        auto iter = config_map_.find(Options::SEQUENCE_FIELD_SORT_ORDER);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "ascending") {
                *sort_order = SortOrder::ASCENDING;
            } else if (str == "descending") {
                *sort_order = SortOrder::DESCENDING;
            } else {
                return Status::Invalid(fmt::format("invalid sort order: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse LookupCompactMode
    Status ParseLookupCompactMode(LookupCompactMode* mode) const {
        auto iter = config_map_.find(Options::LOOKUP_COMPACT);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "radical") {
                *mode = LookupCompactMode::RADICAL;
            } else if (str == "gentle") {
                *mode = LookupCompactMode::GENTLE;
            } else {
                return Status::Invalid(fmt::format("invalid lookup mode: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse SortEngine
    Status ParseSortEngine(SortEngine* sort_engine) const {
        auto iter = config_map_.find(Options::SORT_ENGINE);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "min-heap") {
                *sort_engine = SortEngine::MIN_HEAP;
            } else if (str == "loser-tree") {
                *sort_engine = SortEngine::LOSER_TREE;
            } else {
                return Status::Invalid(fmt::format("invalid sort engine: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse MergeEngine
    Status ParseMergeEngine(MergeEngine* merge_engine) const {
        auto iter = config_map_.find(Options::MERGE_ENGINE);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "deduplicate") {
                *merge_engine = MergeEngine::DEDUPLICATE;
            } else if (str == "partial-update") {
                *merge_engine = MergeEngine::PARTIAL_UPDATE;
            } else if (str == "aggregation") {
                *merge_engine = MergeEngine::AGGREGATE;
            } else if (str == "first-row") {
                *merge_engine = MergeEngine::FIRST_ROW;
            } else {
                return Status::Invalid(fmt::format("invalid merge engine: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse ChangelogProducer
    Status ParseChangelogProducer(ChangelogProducer* changelog_producer) const {
        auto iter = config_map_.find(Options::CHANGELOG_PRODUCER);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "none") {
                *changelog_producer = ChangelogProducer::NONE;
            } else if (str == "input") {
                *changelog_producer = ChangelogProducer::INPUT;
            } else if (str == "full-compaction") {
                *changelog_producer = ChangelogProducer::FULL_COMPACTION;
            } else if (str == "lookup") {
                *changelog_producer = ChangelogProducer::LOOKUP;
            } else {
                return Status::Invalid(fmt::format("invalid changelog producer: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse ExternalPathStrategy
    Status ParseExternalPathStrategy(ExternalPathStrategy* external_path_strategy) const {
        auto iter = config_map_.find(Options::DATA_FILE_EXTERNAL_PATHS_STRATEGY);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "none") {
                *external_path_strategy = ExternalPathStrategy::NONE;
            } else if (str == "specific-fs") {
                *external_path_strategy = ExternalPathStrategy::SPECIFIC_FS;
            } else if (str == "round-robin") {
                *external_path_strategy = ExternalPathStrategy::ROUND_ROBIN;
            } else {
                return Status::Invalid(fmt::format("invalid external path strategy: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse BucketFunctionType
    Status ParseBucketFunctionType(BucketFunctionType* bucket_function_type) const {
        auto iter = config_map_.find(Options::BUCKET_FUNCTION_TYPE);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            if (str == "default") {
                *bucket_function_type = BucketFunctionType::DEFAULT;
            } else if (str == "mod") {
                *bucket_function_type = BucketFunctionType::MOD;
            } else if (str == "hive") {
                *bucket_function_type = BucketFunctionType::HIVE;
            } else {
                return Status::Invalid(fmt::format("invalid bucket function type: {}", str));
            }
        }
        return Status::OK();
    }

    // Parse StartupMode
    Status ParseStartupMode(StartupMode* startup_mode) const {
        auto iter = config_map_.find(Options::SCAN_MODE);
        if (iter != config_map_.end()) {
            std::string str = StringUtils::ToLowerCase(iter->second);
            PAIMON_ASSIGN_OR_RAISE(*startup_mode, StartupMode::FromString(str));
        }
        return Status::OK();
    }

    // parse file.format.per.level
    Status ParseFileFormatPerLevel(
        std::map<int32_t, std::shared_ptr<FileFormat>>* file_format_per_level_ptr) const {
        auto& file_format_per_level = *file_format_per_level_ptr;
        std::string file_format_per_level_str;
        PAIMON_RETURN_NOT_OK(Parse(Options::FILE_FORMAT_PER_LEVEL, &file_format_per_level_str));
        auto level2format =
            StringUtils::Split(file_format_per_level_str, std::string(","), std::string(":"));
        for (const auto& single_level : level2format) {
            if (single_level.size() != 2) {
                return Status::Invalid(
                    fmt::format("fail to parse key {}, value {} (usage example: 0:avro,3:parquet)",
                                Options::FILE_FORMAT_PER_LEVEL, file_format_per_level_str));
            }
            auto level = StringUtils::StringToValue<int32_t>(single_level[0]);
            if (!level || level.value() < 0) {
                return Status::Invalid(
                    fmt::format("fail to parse level {} from string to int in {}", single_level[0],
                                Options::FILE_FORMAT_PER_LEVEL));
            }
            std::shared_ptr<FileFormat> file_format;
            PAIMON_RETURN_NOT_OK(ParseObject<FileFormatFactory>(
                "_no_use", /*default_identifier=*/single_level[1], &file_format));
            file_format_per_level[level.value()] = file_format;
        }
        return Status::OK();
    }

    // parse file.compression.per.level
    Status ParseFileCompressionPerLevel(
        std::map<int32_t, std::string>* file_compression_per_level_ptr) const {
        auto& file_compression_per_level = *file_compression_per_level_ptr;
        std::string file_compression_per_level_str;
        PAIMON_RETURN_NOT_OK(
            Parse(Options::FILE_COMPRESSION_PER_LEVEL, &file_compression_per_level_str));
        auto level2compression =
            StringUtils::Split(file_compression_per_level_str, std::string(","), std::string(":"));
        for (const auto& single_level : level2compression) {
            if (single_level.size() != 2) {
                return Status::Invalid(fmt::format(
                    "fail to parse key {}, value {} (usage example: 0:lz4,1:zstd)",
                    Options::FILE_COMPRESSION_PER_LEVEL, file_compression_per_level_str));
            }
            auto level = StringUtils::StringToValue<int32_t>(single_level[0]);
            if (!level || level.value() < 0) {
                return Status::Invalid(
                    fmt::format("fail to parse level {} from string to int in {}", single_level[0],
                                Options::FILE_COMPRESSION_PER_LEVEL));
            }
            file_compression_per_level[level.value()] = single_level[1];
        }
        return Status::OK();
    }

    bool ContainsKey(const std::string& key) const {
        return config_map_.find(key) != config_map_.end();
    }

 private:
    const std::map<std::string, std::string> config_map_;
};

// Impl is a private implementation of CoreOptions,
// storing various configurable fields and their default values.
struct CoreOptions::Impl {
    int64_t page_size = 64 * 1024;
    std::optional<int64_t> target_file_size;
    std::optional<int64_t> blob_target_file_size;
    int64_t source_split_target_size = 128 * 1024 * 1024;
    int64_t source_split_open_file_cost = 4 * 1024 * 1024;
    int64_t manifest_target_file_size = 8 * 1024 * 1024;
    int64_t deletion_vector_target_file_size = 2 * 1024 * 1024;
    int64_t manifest_full_compaction_file_size = 16 * 1024 * 1024;
    int64_t write_buffer_size = 256 * 1024 * 1024;
    int64_t commit_timeout = std::numeric_limits<int64_t>::max();
    int64_t commit_min_retry_wait = 10;
    int64_t commit_max_retry_wait = 10 * 1000;

    std::shared_ptr<FileFormat> file_format;
    std::shared_ptr<FileSystem> file_system;
    std::shared_ptr<FileFormat> manifest_file_format;
    std::shared_ptr<Cache> cache;

    std::optional<int64_t> scan_snapshot_id;
    std::optional<int64_t> scan_timestamp_millis;
    ExpireConfig expire_config;
    std::vector<std::string> sequence_field;
    std::vector<std::string> remove_record_on_sequence_group;
    std::vector<std::string> blob_fields;
    std::vector<std::string> blob_descriptor_fields;
    std::vector<std::string> blob_view_fields;

    std::string partition_default_name = "__DEFAULT_PARTITION__";
    StartupMode startup_mode = StartupMode::Default();
    std::string file_compression = "zstd";
    std::string manifest_compression = "zstd";
    std::string branch = BranchManager::DEFAULT_MAIN_BRANCH;
    std::string data_file_prefix = "data-";
    std::string file_system_scheme_to_identifier_map_str;

    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::map<std::string, std::string> raw_options;

    int32_t bucket = -1;

    int32_t manifest_merge_min_count = 30;
    int32_t scan_manifest_entry_cache_max_snapshots = 0;
    int32_t read_batch_size = 1024;
    int32_t write_batch_size = 1024;
    int32_t local_sort_max_num_file_handles = 128;
    int32_t commit_max_retries = 10;
    int32_t compaction_min_file_num = 5;
    int32_t compaction_max_size_amplification_percent = 200;
    int32_t compaction_size_ratio = 1;
    int32_t num_sorted_runs_compaction_trigger = 5;
    std::optional<int32_t> num_sorted_runs_stop_trigger;
    std::optional<int32_t> num_levels;

    SortOrder sequence_field_sort_order = SortOrder::ASCENDING;
    MergeEngine merge_engine = MergeEngine::DEDUPLICATE;
    SortEngine sort_engine = SortEngine::LOSER_TREE;
    ChangelogProducer changelog_producer = ChangelogProducer::NONE;
    ExternalPathStrategy external_path_strategy = ExternalPathStrategy::NONE;
    LookupCompactMode lookup_compact_mode = LookupCompactMode::RADICAL;
    std::optional<int32_t> lookup_compact_max_interval;
    BucketFunctionType bucket_function_type = BucketFunctionType::DEFAULT;

    int32_t file_compression_zstd_level = 1;
    int64_t write_buffer_spill_max_disk_size = std::numeric_limits<int64_t>::max();

    bool ignore_delete = false;
    bool write_buffer_spillable = true;
    bool write_only = false;
    bool bucket_append_ordered = false;
    CoreOptions::SequenceNumberInitMode write_sequence_number_init_mode =
        CoreOptions::SequenceNumberInitMode::SCAN;
    bool deletion_vectors_enabled = false;
    bool deletion_vectors_bitmap64 = false;
    bool force_lookup = false;
    bool lookup_wait = true;
    bool partial_update_remove_record_on_delete = false;
    bool aggregation_remove_record_on_delete = false;
    bool table_read_sequence_number_enabled = false;
    bool key_value_sequence_number_enabled = false;
    bool file_index_read_enabled = true;
    bool enable_adaptive_prefetch_strategy = true;
    bool index_file_in_data_file_dir = false;
    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 blob_as_descriptor = false;
    std::optional<bool> blob_split_by_file_size;
    bool legacy_partition_name_enabled = true;
    bool global_index_enabled = true;
    std::optional<int32_t> global_index_thread_num;
    bool commit_force_compact = false;
    bool commit_discard_duplicate_files = false;
    bool dynamic_partition_overwrite = true;
    bool overwrite_upgrade = true;
    bool compaction_force_rewrite_all_files = false;
    bool compaction_force_up_level_0 = false;
    std::optional<std::string> global_index_external_path;

    std::optional<std::string> scan_tag_name;
    std::optional<int64_t> optimized_compaction_interval;
    std::optional<int64_t> compaction_total_size_threshold;
    std::optional<int64_t> compaction_incremental_size_threshold;
    int32_t compact_off_peak_start_hour = -1;
    int32_t compact_off_peak_end_hour = -1;
    int32_t compact_off_peak_ratio = 0;
    bool lookup_cache_bloom_filter = true;
    double lookup_cache_bloom_filter_fpp = 0.05;
    bool lookup_remote_file_enabled = false;
    int32_t lookup_remote_level_threshold = INT32_MIN;
    CompressOptions lookup_compress_options{"zstd", 1};
    CompressOptions spill_compress_options{"zstd", 1};
    int64_t cache_page_size = 64 * 1024;  // 64KB
    std::map<int32_t, std::shared_ptr<FileFormat>> file_format_per_level;
    std::map<int32_t, std::string> file_compression_per_level;
    int64_t lookup_cache_max_memory = 256 * 1024 * 1024;
    double lookup_cache_high_prio_pool_ratio = 0.25;
    int64_t lookup_cache_file_retention_ms = 1 * 3600 * 1000;  // 1 hour
    int64_t lookup_cache_max_disk_size = INT64_MAX;

    // Parse basic table options: bucket, partition, file sizes, batch sizes, file system, etc.
    Status ParseBasicOptions(
        const ConfigParser& parser, const std::shared_ptr<FileSystem>& specified_file_system,
        const std::map<std::string, std::string>& fs_scheme_to_identifier_map) {
        // Parse bucket - bucket number, -1 for dynamic bucket mode, >0 for fixed bucket mode
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::BUCKET, &bucket));
        // Parse partition.default-name - default partition name for null/empty partition values
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::PARTITION_DEFAULT_NAME, &partition_default_name));
        // Parse page-size - memory page size, default 64 kb
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::PAGE_SIZE, &page_size));
        // Parse target-file-size - target size of a data file
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::TARGET_FILE_SIZE, &target_file_size));
        // Parse blob.target-file-size - target size of a blob file
        PAIMON_RETURN_NOT_OK(
            parser.ParseMemorySize(Options::BLOB_TARGET_FILE_SIZE, &blob_target_file_size));
        // Parse source.split.target-size - target size of a source split when scanning a bucket
        PAIMON_RETURN_NOT_OK(
            parser.ParseMemorySize(Options::SOURCE_SPLIT_TARGET_SIZE, &source_split_target_size));
        // Parse source.split.open-file-cost - open file cost to avoid reading too many files
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::SOURCE_SPLIT_OPEN_FILE_COST,
                                                    &source_split_open_file_cost));
        // Parse read.batch-size - read batch size for file formats
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::READ_BATCH_SIZE, &read_batch_size));
        // Parse write.batch-size - write batch size for file formats
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::WRITE_BATCH_SIZE, &write_batch_size));
        // Parse write-buffer-size - data to build up in memory before flushing to disk
        PAIMON_RETURN_NOT_OK(
            parser.ParseMemorySize(Options::WRITE_BUFFER_SIZE, &write_buffer_size));
        // Parse write-buffer-spillable - whether write buffer may spill to disk
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::WRITE_BUFFER_SPILLABLE, &write_buffer_spillable));
        // Parse write-buffer-spill.max-disk-size - max disk size for spill files
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::WRITE_BUFFER_SPILL_MAX_DISK_SIZE,
                                                    &write_buffer_spill_max_disk_size));
        // Parse local-sort.max-num-file-handles - spill file handle cap for local merge
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES,
                                          &local_sort_max_num_file_handles));
        // Parse spill-compression - compression codec for spill files
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::SPILL_COMPRESSION, &spill_compress_options.compress));
        // Parse spill-compression.zstd-level - zstd level for spill compression, default 1
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::SPILL_COMPRESSION_ZSTD_LEVEL,
                                                   &(spill_compress_options.zstd_level)));
        // Parse file-system - file system type, default "local"
        PAIMON_RETURN_NOT_OK(parser.ParseFileSystem(fs_scheme_to_identifier_map,
                                                    specified_file_system, &file_system));
        // Parse write-only - if true, compactions and snapshot expiration will be skipped
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::WRITE_ONLY, &write_only));
        // Parse bucket-append-ordered - append writes in fixed-bucket mode are ordered
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::BUCKET_APPEND_ORDERED, &bucket_append_ordered));
        // Parse partition.legacy-name - use legacy ToString for partition names, default true
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::PARTITION_GENERATE_LEGACY_NAME,
                                                &legacy_partition_name_enabled));
        // Only for test, parse enable-adaptive-prefetch-strategy
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>("test.enable-adaptive-prefetch-strategy",
                                                &enable_adaptive_prefetch_strategy));
        // Parse data-file.external-paths - external paths for data files, comma separated
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::DATA_FILE_EXTERNAL_PATHS, &data_file_external_paths));
        // Parse data-file.external-paths.strategy - strategy for selecting external path
        PAIMON_RETURN_NOT_OK(parser.ParseExternalPathStrategy(&external_path_strategy));
        // Parse data-file.prefix - file name prefix of data files, default "data-"
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::DATA_FILE_PREFIX, &data_file_prefix));
        // Parse row-tracking.enabled - whether to enable unique row id for append table
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::ROW_TRACKING_ENABLED, &row_tracking_enabled));
        // Parse row-tracking.partition-group-on-commit - whether to group delta files by partition
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::ROW_TRACKING_PARTITION_GROUP_ON_COMMIT,
                                                &row_tracking_partition_group_on_commit));
        // Parse data-evolution.enabled - whether to enable data evolution for row tracking
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::DATA_EVOLUTION_ENABLED, &data_evolution_enabled));
        // Parse bucket-function - bucket function type, default "DEFAULT"
        PAIMON_RETURN_NOT_OK(parser.ParseBucketFunctionType(&bucket_function_type));
        // Parse blob-field - column names to store as blob type, comma separated
        PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
            Options::BLOB_FIELD, Options::FIELDS_SEPARATOR, &blob_fields, /*need_trim=*/true));
        // Parse blob-descriptor-field - BLOB fields stored inline as serialized descriptors
        PAIMON_RETURN_NOT_OK(
            parser.ParseList<std::string>(Options::BLOB_DESCRIPTOR_FIELD, Options::FIELDS_SEPARATOR,
                                          &blob_descriptor_fields, /*need_trim=*/true));
        if (blob_descriptor_fields.empty()) {
            PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
                Options::FALLBACK_BLOB_DESCRIPTOR_FIELD, Options::FIELDS_SEPARATOR,
                &blob_descriptor_fields, /*need_trim=*/true));
        }
        // Parse blob-view-field - BLOB fields stored inline as serialized view metadata
        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));
        // Parse blob-as-descriptor - read blob field as descriptor rather than blob bytes
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::BLOB_AS_DESCRIPTOR, &blob_as_descriptor));
        // Parse blob.split-by-file-size - whether blob file size counts in scan splitting
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::BLOB_SPLIT_BY_FILE_SIZE, &blob_split_by_file_size));
        return Status::OK();
    }

    // Parse data file format, compression, and per-level format/compression configurations.
    Status ParseFileFormatOptions(const ConfigParser& parser) {
        // Parse file.format - data file format, default "parquet"
        PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
            Options::FILE_FORMAT, /*default_identifier=*/"parquet", &file_format));
        // Parse file.compression - default file compression, default "zstd"
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::FILE_COMPRESSION, &file_compression));
        // Parse file.compression.zstd-level - zstd compression level, default 1
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::FILE_COMPRESSION_ZSTD_LEVEL, &file_compression_zstd_level));
        // Parse file.format.per.level - different file format for different levels
        PAIMON_RETURN_NOT_OK(parser.ParseFileFormatPerLevel(&file_format_per_level));
        // Parse file.compression.per.level - different compression for different levels
        PAIMON_RETURN_NOT_OK(parser.ParseFileCompressionPerLevel(&file_compression_per_level));
        return Status::OK();
    }

    // Parse manifest file configurations: format, compression, merge, and compaction thresholds.
    Status ParseManifestOptions(const ConfigParser& parser) {
        // Parse manifest.format - manifest file format, default "avro"
        PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
            Options::MANIFEST_FORMAT, /*default_identifier=*/"avro", &manifest_file_format));
        // Parse manifest.compression - manifest file compression, default "zstd"
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::MANIFEST_COMPRESSION, &manifest_compression));
        // Parse manifest.target-file-size - suggested manifest file size, default 8MB
        PAIMON_RETURN_NOT_OK(
            parser.ParseMemorySize(Options::MANIFEST_TARGET_FILE_SIZE, &manifest_target_file_size));
        // Parse manifest.merge-min-count - minimum ManifestFileMeta count to trigger merge
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::MANIFEST_MERGE_MIN_COUNT, &manifest_merge_min_count));
        // Parse manifest.full-compaction-threshold-size - size threshold for full compaction
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::MANIFEST_FULL_COMPACTION_FILE_SIZE,
                                                    &manifest_full_compaction_file_size));
        return Status::OK();
    }

    // Parse snapshot expiration and retention configurations.
    Status ParseExpireOptions(const ConfigParser& parser) {
        // Parse snapshot.num-retained.min - minimum completed snapshots to retain, default 10
        int32_t snapshot_num_retain_min = 10;
        // Parse snapshot.num-retained.max - maximum completed snapshots to retain
        int32_t snapshot_num_retain_max = std::numeric_limits<int32_t>::max();
        // Parse snapshot.expire.limit - maximum snapshots allowed to expire at a time, default 50
        int32_t snapshot_expire_limit = 50;
        // Parse snapshot.time-retained - maximum time of completed snapshots to retain
        int64_t snapshot_time_retained = 1 * 3600 * 1000;  // 1 hour
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::SNAPSHOT_NUM_RETAINED_MIN, &snapshot_num_retain_min));
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::SNAPSHOT_NUM_RETAINED_MAX, &snapshot_num_retain_max));
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SNAPSHOT_EXPIRE_LIMIT, &snapshot_expire_limit));
        PAIMON_RETURN_NOT_OK(
            parser.ParseTimeDuration(Options::SNAPSHOT_TIME_RETAINED, &snapshot_time_retained));
        // Parse snapshot.clean-empty-directories - whether to clean empty dirs on expiration
        bool snapshot_clean_empty_directories = false;
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES,
                                                &snapshot_clean_empty_directories));
        expire_config =
            ExpireConfig(snapshot_num_retain_max, snapshot_num_retain_min, snapshot_time_retained,
                         snapshot_expire_limit, snapshot_clean_empty_directories);
        return Status::OK();
    }

    // Parse commit configurations: timeout, retries, and force-compact.
    Status ParseCommitOptions(const ConfigParser& parser) {
        // Parse commit.force-compact - whether to force compaction before commit, default false
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::COMMIT_FORCE_COMPACT, &commit_force_compact));
        // Parse commit.timeout - timeout duration of retry when commit failed
        PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::COMMIT_TIMEOUT, &commit_timeout));
        // Parse commit.max-retries - maximum retries when commit failed, default 10
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::COMMIT_MAX_RETRIES, &commit_max_retries));
        // Parse commit.min-retry-wait - minimum retry wait when commit failed, default 10ms
        PAIMON_RETURN_NOT_OK(
            parser.ParseTimeDuration(Options::COMMIT_MIN_RETRY_WAIT, &commit_min_retry_wait));
        // Parse commit.max-retry-wait - maximum retry wait when commit failed, default 10s
        PAIMON_RETURN_NOT_OK(
            parser.ParseTimeDuration(Options::COMMIT_MAX_RETRY_WAIT, &commit_max_retry_wait));
        // Parse commit.discard-duplicate-files - whether to discard duplicate files on append
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::COMMIT_DISCARD_DUPLICATE_FILES,
                                                &commit_discard_duplicate_files));
        // Parse dynamic-partition-overwrite - whether overwrite only dynamic partitions
        // for partitioned table overwrite.
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::DYNAMIC_PARTITION_OVERWRITE, &dynamic_partition_overwrite));
        // Parse overwrite-upgrade - whether to try upgrading data files after overwrite on
        // primary key table
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::OVERWRITE_UPGRADE, &overwrite_upgrade));
        return Status::OK();
    }

    // Parse merge engine, sort engine, sequence field, changelog, and partial-update options.
    Status ParseMergeAndSequenceOptions(const ConfigParser& parser) {
        // Parse sequence.field - field that generates sequence number for primary key table
        PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
            Options::SEQUENCE_FIELD, Options::FIELDS_SEPARATOR, &sequence_field));
        // Parse sequence.field.sort-order - order of sequence field, default "ascending"
        PAIMON_RETURN_NOT_OK(parser.ParseSortOrder(&sequence_field_sort_order));
        // Parse write.sequence-number-init-mode - sequence init mode for write path
        std::string write_sequence_init_mode_str = "scan";
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::WRITE_SEQUENCE_NUMBER_INIT_MODE, &write_sequence_init_mode_str));
        write_sequence_init_mode_str = StringUtils::ToLowerCase(write_sequence_init_mode_str);
        if (write_sequence_init_mode_str == "scan") {
            write_sequence_number_init_mode = CoreOptions::SequenceNumberInitMode::SCAN;
        } else if (write_sequence_init_mode_str == "snapshot") {
            write_sequence_number_init_mode = CoreOptions::SequenceNumberInitMode::SNAPSHOT;
        } else {
            return Status::Invalid(fmt::format("invalid write sequence number init mode: {}",
                                               write_sequence_init_mode_str));
        }
        // Parse sort-engine - sort engine for primary key table, default "loser-tree"
        PAIMON_RETURN_NOT_OK(parser.ParseSortEngine(&sort_engine));
        // Parse merge-engine - merge engine for primary key table, default "deduplicate"
        PAIMON_RETURN_NOT_OK(parser.ParseMergeEngine(&merge_engine));
        // Parse ignore-delete - whether to ignore delete records, default false
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::IGNORE_DELETE, &ignore_delete));
        // Parse fields.default-aggregate-function - default agg function for partial-update
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::FIELDS_DEFAULT_AGG_FUNC, &field_default_func));
        // Parse changelog-producer - whether to double write to a changelog file, default "none"
        PAIMON_RETURN_NOT_OK(parser.ParseChangelogProducer(&changelog_producer));
        // Parse partial-update.remove-record-on-delete - remove whole row on delete
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE,
                                                &partial_update_remove_record_on_delete));
        // Parse aggregation_remove_record_on_delete
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::AGGREGATION_REMOVE_RECORD_ON_DELETE,
                                                &aggregation_remove_record_on_delete));
        // Parse table-read.sequence-number.enabled - expose sequence number in system tables
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED,
                                                &table_read_sequence_number_enabled));
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED,
                                                &key_value_sequence_number_enabled));
        // Parse partial-update.remove-record-on-sequence-group
        PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
            Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP, Options::FIELDS_SEPARATOR,
            &remove_record_on_sequence_group));
        return Status::OK();
    }

    // Parse deletion vector configurations.
    Status ParseDeletionVectorOptions(const ConfigParser& parser) {
        // Parse deletion-vectors.enabled - whether to enable deletion vectors mode, default false
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::DELETION_VECTORS_ENABLED, &deletion_vectors_enabled));
        // Parse deletion-vector.index-file.target-size - target size of dv index file, default 2MB
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::DELETION_VECTOR_INDEX_FILE_TARGET_SIZE,
                                                    &deletion_vector_target_file_size));
        // Parse deletion-vectors.bitmap64 - enable 64 bit bitmap implementation, default false
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::DELETION_VECTOR_BITMAP64, &deletion_vectors_bitmap64));
        return Status::OK();
    }

    // Parse scan, branch, and tag related configurations.
    Status ParseScanAndBranchOptions(const ConfigParser& parser) {
        // Parse scan.snapshot-id - optional snapshot id for "from-snapshot" scan mode
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_SNAPSHOT_ID, &scan_snapshot_id));
        // Parse scan.timestamp-millis and scan.timestamp
        std::string scan_timestamp_str;
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_TIMESTAMP, &scan_timestamp_str));
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_TIMESTAMP_MILLIS, &scan_timestamp_millis));
        if (scan_timestamp_millis != std::nullopt && !scan_timestamp_str.empty()) {
            return Status::Invalid(
                "scan.timestamp-millis and scan.timestamp cannot be set at the same time");
        }
        if (!scan_timestamp_str.empty()) {
            PAIMON_ASSIGN_OR_RAISE(int64_t millis,
                                   StringUtils::StringToTimestampMillis(scan_timestamp_str));
            scan_timestamp_millis = millis;
        }
        // Parse scan.mode - scanning behavior of the source, default "default"
        PAIMON_RETURN_NOT_OK(parser.ParseStartupMode(&startup_mode));
        // Parse scan.manifest-entry-cache.max-snapshots - cached snapshots per bucket.
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS,
                                          &scan_manifest_entry_cache_max_snapshots));
        if (scan_manifest_entry_cache_max_snapshots < 0) {
            return Status::Invalid(fmt::format("{} must be non-negative",
                                               Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS));
        }
        // Parse scan.fallback-branch - fallback branch when partition not found
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_FALLBACK_BRANCH, &scan_fallback_branch));
        // Parse branch - branch name, default "main"
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::BRANCH, &branch));
        // Parse scan.tag-name - optional tag name for "from-snapshot" scan mode
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_TAG_NAME, &scan_tag_name));
        return Status::OK();
    }

    // Parse index-related configurations: file index, global index.
    Status ParseIndexOptions(const ConfigParser& parser) {
        // Parse file-index.read.enabled - whether to enable reading file index, default true
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::FILE_INDEX_READ_ENABLED, &file_index_read_enabled));
        // Parse index-file-in-data-file-dir - whether index file in data file directory
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::INDEX_FILE_IN_DATA_FILE_DIR, &index_file_in_data_file_dir));
        // Parse global-index.enabled - whether to enable global index for scan, default true
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::GLOBAL_INDEX_ENABLED, &global_index_enabled));
        // Parse global-index.thread-num - the maximum number of concurrent scanner for global
        // index, no default value
        PAIMON_RETURN_NOT_OK(
            parser.Parse<int32_t>(Options::GLOBAL_INDEX_THREAD_NUM, &global_index_thread_num));
        // Parse global-index.external-path - global index root directory
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::GLOBAL_INDEX_EXTERNAL_PATH, &global_index_external_path));
        return Status::OK();
    }

    // Parse compaction configurations: sorted run triggers, size ratios, thresholds, off-peak.
    Status ParseCompactionOptions(const ConfigParser& parser) {
        // Parse compaction.min.file-num - minimum file number to trigger compaction, default 5
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::COMPACTION_MIN_FILE_NUM, &compaction_min_file_num));
        // Parse compaction.max-size-amplification-percent - size amplification percent, default 200
        PAIMON_RETURN_NOT_OK(
            parser.Parse<int32_t>(Options::COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT,
                                  &compaction_max_size_amplification_percent));
        // Parse compaction.size-ratio - percentage flexibility for sorted run comparison, default 1
        PAIMON_RETURN_NOT_OK(
            parser.Parse<int32_t>(Options::COMPACTION_SIZE_RATIO, &compaction_size_ratio));
        // Parse num-sorted-run.compaction-trigger - sorted run number to trigger compaction
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::NUM_SORTED_RUNS_COMPACTION_TRIGGER,
                                                   &num_sorted_runs_compaction_trigger));
        // Parse num-sorted-run.stop-trigger - sorted run number to stop writes
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::NUM_SORTED_RUNS_STOP_TRIGGER,
                                                   &num_sorted_runs_stop_trigger));
        // Parse num-levels - total level number for LSM tree
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::NUM_LEVELS, &num_levels));
        // Parse compaction.force-rewrite-all-files - force pick all files for full compaction
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::COMPACTION_FORCE_REWRITE_ALL_FILES,
                                                &compaction_force_rewrite_all_files));
        // Parse compaction.force-up-level-0 - always include all level 0 files in candidates
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::COMPACTION_FORCE_UP_LEVEL_0, &compaction_force_up_level_0));
        // Parse compaction.optimization-interval - how often to perform optimization compaction
        PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::COMPACTION_OPTIMIZATION_INTERVAL,
                                                      &optimized_compaction_interval));
        // Parse compaction.total-size-threshold - force full compaction when total size is smaller
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::COMPACTION_TOTAL_SIZE_THRESHOLD,
                                                    &compaction_total_size_threshold));
        // Parse compaction.incremental-size-threshold - force full compaction when incremental size
        // is bigger
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::COMPACTION_INCREMENTAL_SIZE_THRESHOLD,
                                                    &compaction_incremental_size_threshold));
        // Parse compaction.offpeak.start.hour - start of off-peak hours (0-23), -1 to disable
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::COMPACT_OFFPEAK_START_HOUR, &compact_off_peak_start_hour));
        // Parse compaction.offpeak.end.hour - end of off-peak hours (0-23), -1 to disable
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::COMPACT_OFFPEAK_END_HOUR, &compact_off_peak_end_hour));
        // Parse compaction.offpeak-ratio - more aggressive ratio during off-peak hours, default 0
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::COMPACTION_OFFPEAK_RATIO, &compact_off_peak_ratio));
        return Status::OK();
    }

    // Parse lookup configurations: compact mode, bloom filter, remote file, cache, compression.
    Status ParseLookupOptions(const ConfigParser& parser) {
        // Parse force-lookup - whether to force lookup for compaction, default false
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::FORCE_LOOKUP, &force_lookup));
        // Parse lookup-wait - commit will wait for compaction by lookup, default true
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::LOOKUP_WAIT, &lookup_wait));
        // Parse lookup-compact - lookup compact mode, default RADICAL
        PAIMON_RETURN_NOT_OK(parser.ParseLookupCompactMode(&lookup_compact_mode));
        // Parse lookup-compact.max-interval - max interval for gentle mode lookup compaction
        PAIMON_RETURN_NOT_OK(
            parser.Parse(Options::LOOKUP_COMPACT_MAX_INTERVAL, &lookup_compact_max_interval));
        // Parse lookup.cache.bloom.filter.enabled - enable bloom filter for lookup cache
        PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::LOOKUP_CACHE_BLOOM_FILTER_ENABLED,
                                                &lookup_cache_bloom_filter));
        // Parse lookup.cache.bloom.filter.fpp - false positive probability, default 0.05
        PAIMON_RETURN_NOT_OK(parser.Parse<double>(Options::LOOKUP_CACHE_BLOOM_FILTER_FPP,
                                                  &lookup_cache_bloom_filter_fpp));
        // Parse lookup.remote-file.enabled - whether to enable remote file for lookup
        PAIMON_RETURN_NOT_OK(
            parser.Parse<bool>(Options::LOOKUP_REMOTE_FILE_ENABLED, &lookup_remote_file_enabled));
        // Parse lookup.remote-file.level-threshold - level threshold for remote lookup files
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::LOOKUP_REMOTE_LEVEL_THRESHOLD,
                                                   &lookup_remote_level_threshold));
        // Parse lookup.cache-spill-compression - spill compression for lookup cache, default "zstd"
        PAIMON_RETURN_NOT_OK(parser.Parse(Options::LOOKUP_CACHE_SPILL_COMPRESSION,
                                          &lookup_compress_options.compress));
        // Parse spill-compression.zstd-level - zstd level for spill compression, default 1
        PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::SPILL_COMPRESSION_ZSTD_LEVEL,
                                                   &(lookup_compress_options.zstd_level)));
        // Parse cache-page-size - memory page size for caching, default 64 kb
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::CACHE_PAGE_SIZE, &cache_page_size));
        // Parse lookup.cache-max-memory-size - max memory size for lookup cache, default 256 mb
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::LOOKUP_CACHE_MAX_MEMORY_SIZE,
                                                    &lookup_cache_max_memory));
        // Parse lookup.cache.high-priority-pool-ratio - fraction for high-priority data, default
        // 0.25
        PAIMON_RETURN_NOT_OK(parser.Parse<double>(Options::LOOKUP_CACHE_HIGH_PRIO_POOL_RATIO,
                                                  &lookup_cache_high_prio_pool_ratio));
        if (lookup_cache_high_prio_pool_ratio < 0.0 || lookup_cache_high_prio_pool_ratio >= 1.0) {
            return Status::Invalid(fmt::format(
                "The high priority pool ratio should in the range [0, 1), while input is {}",
                lookup_cache_high_prio_pool_ratio));
        }
        // Parse lookup.cache-file-retention - cached files retention time, default "1 hour"
        PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::LOOKUP_CACHE_FILE_RETENTION,
                                                      &lookup_cache_file_retention_ms));
        // Parse lookup.cache-max-disk-size - max disk size for lookup cache, default unlimited
        PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::LOOKUP_CACHE_MAX_DISK_SIZE,
                                                    &lookup_cache_max_disk_size));
        return Status::OK();
    }
};

// Parse configurations from a map and return a populated CoreOptions object.
Result<CoreOptions> CoreOptions::FromMap(
    const std::map<std::string, std::string>& options_map,
    const std::shared_ptr<FileSystem>& specified_file_system,
    const std::map<std::string, std::string>& fs_scheme_to_identifier_map) {
    CoreOptions options;
    auto& impl = options.impl_;
    impl->raw_options = options_map;
    ConfigParser parser(options_map);

    PAIMON_RETURN_NOT_OK(
        impl->ParseBasicOptions(parser, specified_file_system, fs_scheme_to_identifier_map));
    PAIMON_RETURN_NOT_OK(impl->ParseFileFormatOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseManifestOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseExpireOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseCommitOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseMergeAndSequenceOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseDeletionVectorOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseScanAndBranchOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseIndexOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseCompactionOptions(parser));
    PAIMON_RETURN_NOT_OK(impl->ParseLookupOptions(parser));
    return options;
}

CoreOptions::CoreOptions() : impl_(std::make_unique<Impl>()) {}

CoreOptions::CoreOptions(const CoreOptions& rhs)
    : impl_(std::make_unique<Impl>(*(rhs.impl_.get()))) {}

CoreOptions& CoreOptions::operator=(const CoreOptions& rhs) {
    if (this != &rhs) {
        impl_ = std::make_unique<Impl>(*(rhs.impl_.get()));
    }
    return *this;
}

CoreOptions::~CoreOptions() = default;

int32_t CoreOptions::GetBucket() const {
    return impl_->bucket;
}

std::shared_ptr<FileFormat> CoreOptions::GetWriteFileFormat(int32_t level) const {
    auto iter = impl_->file_format_per_level.find(level);
    if (iter != impl_->file_format_per_level.end()) {
        return iter->second;
    }
    return impl_->file_format;
}

std::shared_ptr<FileFormat> CoreOptions::GetFileFormat() const {
    return impl_->file_format;
}

std::shared_ptr<FileSystem> CoreOptions::GetFileSystem() const {
    return impl_->file_system;
}

const std::string& CoreOptions::GetFileCompression() const {
    return impl_->file_compression;
}

const std::string& CoreOptions::GetWriteFileCompression(int32_t level) const {
    auto iter = impl_->file_compression_per_level.find(level);
    if (iter != impl_->file_compression_per_level.end()) {
        return iter->second;
    }
    return impl_->file_compression;
}

int32_t CoreOptions::GetFileCompressionZstdLevel() const {
    return impl_->file_compression_zstd_level;
}

int64_t CoreOptions::GetPageSize() const {
    return impl_->page_size;
}

int64_t CoreOptions::GetTargetFileSize(bool has_primary_key) const {
    if (impl_->target_file_size == std::nullopt) {
        return has_primary_key ? 128 * 1024 * 1024 : 256 * 1024 * 1024;
    }
    return impl_->target_file_size.value();
}

int64_t CoreOptions::GetBlobTargetFileSize() const {
    if (impl_->blob_target_file_size == std::nullopt) {
        return GetTargetFileSize(/*has_primary_key=*/false);
    }
    return impl_->blob_target_file_size.value();
}

bool CoreOptions::BlobSplitByFileSize() const {
    return impl_->blob_split_by_file_size.value_or(!impl_->blob_as_descriptor);
}

int64_t CoreOptions::GetCompactionFileSize(bool has_primary_key) const {
    // file size to join the compaction, we don't process on middle file size to avoid
    // compact a same file twice (the compression is not calculate so accurately. the output
    // file maybe be less than target file generated by rolling file write).
    return GetTargetFileSize(has_primary_key) / 10 * 7;
}

std::string CoreOptions::GetPartitionDefaultName() const {
    return impl_->partition_default_name;
}

std::shared_ptr<FileFormat> CoreOptions::GetManifestFormat() const {
    return impl_->manifest_file_format;
}

int64_t CoreOptions::GetSourceSplitTargetSize() const {
    return impl_->source_split_target_size;
}
int64_t CoreOptions::GetSourceSplitOpenFileCost() const {
    return impl_->source_split_open_file_cost;
}
std::optional<int64_t> CoreOptions::GetScanSnapshotId() const {
    return impl_->scan_snapshot_id;
}
std::optional<int64_t> CoreOptions::GetScanTimestampMillis() const {
    return impl_->scan_timestamp_millis;
}

int32_t CoreOptions::GetScanManifestEntryCacheMaxSnapshots() const {
    return impl_->scan_manifest_entry_cache_max_snapshots;
}

int64_t CoreOptions::GetManifestTargetFileSize() const {
    return impl_->manifest_target_file_size;
}

std::shared_ptr<Cache> CoreOptions::GetCache() const {
    return impl_->cache;
}

CoreOptions& CoreOptions::WithCache(const std::shared_ptr<Cache>& cache) {
    impl_->cache = cache;
    return *this;
}

int32_t CoreOptions::GetManifestMergeMinCount() const {
    return impl_->manifest_merge_min_count;
}

int64_t CoreOptions::GetManifestFullCompactionThresholdSize() const {
    return impl_->manifest_full_compaction_file_size;
}

const std::string& CoreOptions::GetManifestCompression() const {
    return impl_->manifest_compression;
}

StartupMode CoreOptions::GetStartupMode() const {
    if (impl_->startup_mode == StartupMode::Default()) {
        if (GetScanSnapshotId() != std::nullopt || GetScanTagName() != std::nullopt) {
            return StartupMode::FromSnapshot();
        }
        if (GetScanTimestampMillis() != std::nullopt) {
            return StartupMode::FromTimestamp();
        }
        return StartupMode::LatestFull();
    }
    return impl_->startup_mode;
}

int32_t CoreOptions::GetReadBatchSize() const {
    return impl_->read_batch_size;
}

int32_t CoreOptions::GetWriteBatchSize() const {
    return impl_->write_batch_size;
}

int64_t CoreOptions::GetWriteBufferSize() const {
    return impl_->write_buffer_size;
}

bool CoreOptions::GetWriteBufferSpillable() const {
    return impl_->write_buffer_spillable;
}

int64_t CoreOptions::GetWriteBufferSpillMaxDiskSize() const {
    return impl_->write_buffer_spill_max_disk_size;
}

int32_t CoreOptions::GetLocalSortMaxNumFileHandles() const {
    return impl_->local_sort_max_num_file_handles;
}

const CompressOptions& CoreOptions::GetSpillCompressOptions() const {
    return impl_->spill_compress_options;
}

bool CoreOptions::CommitForceCompact() const {
    return impl_->commit_force_compact;
}

int64_t CoreOptions::GetCommitTimeout() const {
    return impl_->commit_timeout;
}

int32_t CoreOptions::GetCommitMaxRetries() const {
    return impl_->commit_max_retries;
}

int64_t CoreOptions::GetCommitMinRetryWait() const {
    return impl_->commit_min_retry_wait;
}

int64_t CoreOptions::GetCommitMaxRetryWait() const {
    return impl_->commit_max_retry_wait;
}

bool CoreOptions::CommitDiscardDuplicateFiles() const {
    return impl_->commit_discard_duplicate_files;
}

bool CoreOptions::DynamicPartitionOverwrite() const {
    return impl_->dynamic_partition_overwrite;
}

bool CoreOptions::OverwriteUpgrade() const {
    return impl_->overwrite_upgrade;
}

int32_t CoreOptions::GetCompactionMinFileNum() const {
    return impl_->compaction_min_file_num;
}

int32_t CoreOptions::GetCompactionMaxSizeAmplificationPercent() const {
    return impl_->compaction_max_size_amplification_percent;
}

int32_t CoreOptions::GetCompactionSizeRatio() const {
    return impl_->compaction_size_ratio;
}

int32_t CoreOptions::GetNumSortedRunsCompactionTrigger() const {
    return impl_->num_sorted_runs_compaction_trigger;
}

int32_t CoreOptions::GetNumSortedRunsStopTrigger() const {
    int32_t compact_trigger = GetNumSortedRunsCompactionTrigger();
    int32_t stop_trigger = 0;
    if (impl_->num_sorted_runs_stop_trigger.has_value()) {
        stop_trigger = impl_->num_sorted_runs_stop_trigger.value();
    } else {
        int64_t computed = static_cast<int64_t>(compact_trigger) + 3;
        if (computed > std::numeric_limits<int32_t>::max()) {
            computed = std::numeric_limits<int32_t>::max();
        }
        stop_trigger = static_cast<int32_t>(computed);
    }
    return std::max(compact_trigger, stop_trigger);
}

int32_t CoreOptions::GetNumLevels() const {
    // By default, this ensures that the compaction does not fall to level 0, but at least to
    // level 1
    if (impl_->num_levels.has_value()) {
        return impl_->num_levels.value();
    }

    int64_t incremented = static_cast<int64_t>(GetNumSortedRunsCompactionTrigger()) + 1;
    if (incremented > std::numeric_limits<int32_t>::max()) {
        incremented = std::numeric_limits<int32_t>::max();
    }
    return static_cast<int32_t>(incremented);
}

LookupCompactMode CoreOptions::GetLookupCompactMode() const {
    return impl_->lookup_compact_mode;
}

int32_t CoreOptions::GetLookupCompactMaxInterval() const {
    int32_t compact_trigger = GetNumSortedRunsCompactionTrigger();
    int32_t max_interval;
    if (impl_->lookup_compact_max_interval.has_value()) {
        max_interval = impl_->lookup_compact_max_interval.value();
    } else {
        int64_t doubled = static_cast<int64_t>(compact_trigger) * 2;
        if (doubled > std::numeric_limits<int32_t>::max()) {
            doubled = std::numeric_limits<int32_t>::max();
        }
        max_interval = static_cast<int32_t>(doubled);
    }

    if (max_interval < compact_trigger) {
        max_interval = compact_trigger;
    }
    return max_interval;
}

const ExpireConfig& CoreOptions::GetExpireConfig() const {
    return impl_->expire_config;
}

const std::vector<std::string>& CoreOptions::GetSequenceField() const {
    return impl_->sequence_field;
}

bool CoreOptions::SequenceFieldSortOrderIsAscending() const {
    return impl_->sequence_field_sort_order == SortOrder::ASCENDING;
}

MergeEngine CoreOptions::GetMergeEngine() const {
    return impl_->merge_engine;
}

SortEngine CoreOptions::GetSortEngine() const {
    return impl_->sort_engine;
}

bool CoreOptions::IgnoreDelete() const {
    return impl_->ignore_delete;
}

bool CoreOptions::WriteOnly() const {
    return impl_->write_only;
}

bool CoreOptions::BucketAppendOrdered() const {
    return impl_->bucket_append_ordered;
}

CoreOptions::SequenceNumberInitMode CoreOptions::WriteSequenceNumberInitMode() const {
    return impl_->write_sequence_number_init_mode;
}

std::optional<std::string> CoreOptions::GetFieldsDefaultFunc() const {
    return impl_->field_default_func;
}

bool CoreOptions::EnableAdaptivePrefetchStrategy() const {
    return impl_->enable_adaptive_prefetch_strategy;
}

Result<std::optional<std::string>> CoreOptions::GetFieldAggFunc(
    const std::string& field_name) const {
    ConfigParser parser(impl_->raw_options);
    std::optional<std::string> field_agg_func;
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::AGG_FUNCTION);
    PAIMON_RETURN_NOT_OK(parser.Parse(key, &field_agg_func));
    return field_agg_func;
}

Result<bool> CoreOptions::FieldAggIgnoreRetract(const std::string& field_name) const {
    ConfigParser parser(impl_->raw_options);
    bool field_agg_ignore_retract = false;
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::IGNORE_RETRACT);
    PAIMON_RETURN_NOT_OK(parser.Parse<bool>(key, &field_agg_ignore_retract));
    return field_agg_ignore_retract;
}

Result<std::string> CoreOptions::FieldListAggDelimiter(const std::string& field_name) const {
    ConfigParser parser(impl_->raw_options);
    std::string delimiter = ",";
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::LIST_AGG_DELIMITER);
    PAIMON_RETURN_NOT_OK(parser.Parse(key, &delimiter));
    return delimiter;
}

Result<bool> CoreOptions::FieldCollectAggDistinct(const std::string& field_name) const {
    ConfigParser parser(impl_->raw_options);
    bool distinct = false;
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::DISTINCT);
    PAIMON_RETURN_NOT_OK(parser.Parse<bool>(key, &distinct));
    return distinct;
}

Result<MapStorageLayout> CoreOptions::GetMapStorageLayout(const std::string& field_name) const {
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::MAP_STORAGE_LAYOUT);
    PAIMON_ASSIGN_OR_RAISE(std::string layout_str, OptionsUtils::GetValueFromMap<std::string>(
                                                       impl_->raw_options, key, "default"));
    std::string lower = StringUtils::ToLowerCase(layout_str);
    if (lower == "shared-shredding") {
        return MapStorageLayout::SHARED_SHREDDING;
    } else if (lower == "default") {
        return MapStorageLayout::DEFAULT;
    }
    return Status::Invalid(fmt::format("invalid map.storage-layout: {}", layout_str));
}

Result<int32_t> CoreOptions::GetMapSharedShreddingMaxColumns(const std::string& field_name) const {
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::MAP_SHARED_SHREDDING_MAX_COLUMNS);
    PAIMON_ASSIGN_OR_RAISE(int32_t max_columns,
                           OptionsUtils::GetValueFromMap<int32_t>(impl_->raw_options, key, 256));
    if (max_columns <= 0) {
        return Status::Invalid(fmt::format("options {} must > 0",
                                           std::string(Options::MAP_SHARED_SHREDDING_MAX_COLUMNS)));
    }
    return max_columns;
}

Result<MapSharedShreddingColumnPlacementPolicy>
CoreOptions::GetMapSharedShreddingColumnPlacementPolicy(const std::string& field_name) const {
    std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
                      std::string(Options::MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY);
    PAIMON_ASSIGN_OR_RAISE(std::string policy_str, OptionsUtils::GetValueFromMap<std::string>(
                                                       impl_->raw_options, key, "lru"));
    std::string lower = StringUtils::ToLowerCase(policy_str);
    if (lower == "plain") {
        return MapSharedShreddingColumnPlacementPolicy::PLAIN;
    } else if (lower == "sequential") {
        return MapSharedShreddingColumnPlacementPolicy::SEQUENTIAL;
    } else if (lower == "lru") {
        return MapSharedShreddingColumnPlacementPolicy::LRU;
    }
    return Status::Invalid(
        fmt::format("invalid map.shared-shredding.column-placement-policy: {}", policy_str));
}

bool CoreOptions::DeletionVectorsEnabled() const {
    return impl_->deletion_vectors_enabled;
}

bool CoreOptions::DeletionVectorsBitmap64() const {
    return impl_->deletion_vectors_bitmap64;
}
int64_t CoreOptions::DeletionVectorTargetFileSize() const {
    return impl_->deletion_vector_target_file_size;
}

ChangelogProducer CoreOptions::GetChangelogProducer() const {
    return impl_->changelog_producer;
}

LookupStrategy CoreOptions::GetLookupStrategy() const {
    return LookupStrategy::From(
        /*is_first_row=*/GetMergeEngine() == MergeEngine::FIRST_ROW,
        /*produce_changelog=*/GetChangelogProducer() == ChangelogProducer::LOOKUP,
        /*deletion_vector=*/DeletionVectorsEnabled(),
        /*force_lookup=*/impl_->force_lookup);
}

const std::map<std::string, std::string>& CoreOptions::ToMap() const {
    return impl_->raw_options;
}

bool CoreOptions::NeedLookup() const {
    return GetLookupStrategy().need_lookup;
}

bool CoreOptions::PrepareCommitWaitCompaction() const {
    if (!NeedLookup()) {
        return false;
    }
    return impl_->lookup_wait;
}

bool CoreOptions::CompactionForceRewriteAllFiles() const {
    return impl_->compaction_force_rewrite_all_files;
}

bool CoreOptions::CompactionForceUpLevel0() const {
    return impl_->compaction_force_up_level_0;
}

std::map<std::string, std::string> CoreOptions::GetFieldsSequenceGroups() const {
    auto raw_options = impl_->raw_options;
    std::map<std::string, std::string> sequence_groups;
    for (const auto& [key, value] : raw_options) {
        if (StringUtils::StartsWith(key, Options::FIELDS_PREFIX, /*start_pos=*/0) &&
            StringUtils::EndsWith(key, Options::SEQUENCE_GROUP)) {
            std::string seq_fields_str =
                key.substr(std::strlen(Options::FIELDS_PREFIX) + 1,
                           key.size() - std::strlen(Options::FIELDS_PREFIX) -
                               std::strlen(Options::SEQUENCE_GROUP) - 2);
            sequence_groups[seq_fields_str] = value;
        }
    }
    return sequence_groups;
}

bool CoreOptions::PartialUpdateRemoveRecordOnDelete() const {
    return impl_->partial_update_remove_record_on_delete;
}

bool CoreOptions::AggregationRemoveRecordOnDelete() const {
    return impl_->aggregation_remove_record_on_delete;
}

bool CoreOptions::TableReadSequenceNumberEnabled() const {
    return impl_->table_read_sequence_number_enabled;
}

bool CoreOptions::KeyValueSequenceNumberEnabled() const {
    return impl_->key_value_sequence_number_enabled;
}

std::vector<std::string> CoreOptions::GetPartialUpdateRemoveRecordOnSequenceGroup() const {
    return impl_->remove_record_on_sequence_group;
}

std::optional<std::string> CoreOptions::GetScanFallbackBranch() const {
    return impl_->scan_fallback_branch;
}

std::string CoreOptions::GetBranch() const {
    return impl_->branch;
}

bool CoreOptions::FileIndexReadEnabled() const {
    return impl_->file_index_read_enabled;
}

std::optional<std::string> CoreOptions::GetDataFileExternalPaths() const {
    return impl_->data_file_external_paths;
}

ExternalPathStrategy CoreOptions::GetExternalPathStrategy() const {
    return impl_->external_path_strategy;
}

Result<std::vector<std::string>> CoreOptions::CreateExternalPaths() const {
    std::vector<std::string> external_paths;
    std::optional<std::string> data_file_external_paths = GetDataFileExternalPaths();
    ExternalPathStrategy strategy = GetExternalPathStrategy();
    if (strategy == ExternalPathStrategy::SPECIFIC_FS) {
        return Status::NotImplemented("do not support specific-fs external path strategy for now");
    }
    if (data_file_external_paths == std::nullopt || data_file_external_paths->empty() ||
        strategy == ExternalPathStrategy::NONE) {
        return external_paths;
    }
    for (const auto& p : StringUtils::Split(data_file_external_paths.value(), ",")) {
        std::string tmp_path = p;
        StringUtils::Trim(&tmp_path);
        PAIMON_ASSIGN_OR_RAISE(Path path, PathUtil::ToPath(tmp_path));
        if (path.scheme.empty()) {
            return Status::Invalid(fmt::format("scheme is null, path is {}", p));
        }
        external_paths.push_back(path.ToString());
    }
    if (external_paths.empty()) {
        return Status::Invalid("external paths is empty");
    }
    return external_paths;
}

std::string CoreOptions::DataFilePrefix() const {
    return impl_->data_file_prefix;
}

bool CoreOptions::IndexFileInDataFileDir() const {
    return impl_->index_file_in_data_file_dir;
}

bool CoreOptions::RowTrackingEnabled() const {
    return impl_->row_tracking_enabled;
}

bool CoreOptions::RowTrackingPartitionGroupOnCommit() const {
    return impl_->row_tracking_partition_group_on_commit;
}

bool CoreOptions::DataEvolutionEnabled() const {
    return impl_->data_evolution_enabled;
}

bool CoreOptions::LegacyPartitionNameEnabled() const {
    return impl_->legacy_partition_name_enabled;
}

bool CoreOptions::GlobalIndexEnabled() const {
    return impl_->global_index_enabled;
}

std::optional<int32_t> CoreOptions::GetGlobalIndexThreadNum() const {
    return impl_->global_index_thread_num;
}

std::optional<std::string> CoreOptions::GetGlobalIndexExternalPath() const {
    return impl_->global_index_external_path;
}

Result<std::optional<std::string>> CoreOptions::CreateGlobalIndexExternalPath() const {
    std::optional<std::string> global_index_external_path = GetGlobalIndexExternalPath();
    if (global_index_external_path == std::nullopt || global_index_external_path->empty()) {
        return std::optional<std::string>();
    }
    std::string tmp_path = global_index_external_path.value();
    StringUtils::Trim(&tmp_path);
    PAIMON_ASSIGN_OR_RAISE(Path path, PathUtil::ToPath(tmp_path));
    if (path.scheme.empty()) {
        return Status::Invalid(fmt::format("scheme is null, path is {}", tmp_path));
    }
    return std::optional<std::string>(path.ToString());
}

std::optional<std::string> CoreOptions::GetScanTagName() const {
    return impl_->scan_tag_name;
}

std::optional<int64_t> CoreOptions::GetOptimizedCompactionInterval() const {
    return impl_->optimized_compaction_interval;
}
std::optional<int64_t> CoreOptions::GetCompactionTotalSizeThreshold() const {
    return impl_->compaction_total_size_threshold;
}
std::optional<int64_t> CoreOptions::GetCompactionIncrementalSizeThreshold() const {
    return impl_->compaction_incremental_size_threshold;
}

int32_t CoreOptions::GetCompactOffPeakStartHour() const {
    return impl_->compact_off_peak_start_hour;
}
int32_t CoreOptions::GetCompactOffPeakEndHour() const {
    return impl_->compact_off_peak_end_hour;
}
int32_t CoreOptions::GetCompactOffPeakRatio() const {
    return impl_->compact_off_peak_ratio;
}

bool CoreOptions::LookupCacheBloomFilterEnabled() const {
    return impl_->lookup_cache_bloom_filter;
}

double CoreOptions::GetLookupCacheBloomFilterFpp() const {
    return impl_->lookup_cache_bloom_filter_fpp;
}

const CompressOptions& CoreOptions::GetLookupCompressOptions() const {
    return impl_->lookup_compress_options;
}

bool CoreOptions::LookupRemoteFileEnabled() const {
    return impl_->lookup_remote_file_enabled;
}

int32_t CoreOptions::GetLookupRemoteLevelThreshold() const {
    return impl_->lookup_remote_level_threshold;
}

int32_t CoreOptions::GetCachePageSize() const {
    return static_cast<int32_t>(impl_->cache_page_size);
}

int64_t CoreOptions::GetLookupCacheMaxMemory() const {
    return impl_->lookup_cache_max_memory;
}

double CoreOptions::GetLookupCacheHighPrioPoolRatio() const {
    return impl_->lookup_cache_high_prio_pool_ratio;
}

BucketFunctionType CoreOptions::GetBucketFunctionType() const {
    return impl_->bucket_function_type;
}

const std::vector<std::string>& CoreOptions::GetBlobFields() const {
    return impl_->blob_fields;
}

const std::vector<std::string>& CoreOptions::GetBlobDescriptorFields() const {
    return impl_->blob_descriptor_fields;
}

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;
}

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(),
                              impl_->blob_view_fields.end());
    return blob_inline_fields;
}

int64_t CoreOptions::GetLookupCacheFileRetentionMs() const {
    return impl_->lookup_cache_file_retention_ms;
}

int64_t CoreOptions::GetLookupCacheMaxDiskSize() const {
    return impl_->lookup_cache_max_disk_size;
}

}  // namespace paimon