Google Git Google Git
Sign in
apache/paimon-cpp/40a7d05ed9402c3cf2a206b995c64addbbfa2d9e/./src/paimon/core/core_options.cpp
blob: 92afe6ead8091eb04f6aed88882917c0e121bffd [file]
/*
* 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.
*/
#include "paimon/core/core_options.h"
#include <cstring>
#include <initializer_list>
#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/statistics_mode.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 {
PAIMON_ASSIGN_OR_RAISE(std::optional<T> parsed_value, GetOptionalValue<T>(key));
if (parsed_value) {
*value = parsed_value.value();
}
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 {
PAIMON_ASSIGN_OR_RAISE(std::optional<T> parsed_value, GetOptionalValue<T>(key));
if (parsed_value) {
*value = parsed_value.value();
}
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 {
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> config_value,
GetOptionalValue<std::string>(key));
if (!config_value) {
return Status::OK();
}
auto value_str_vec =
StringUtils::Split(config_value.value(), 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, config_value.value()));
}
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>");
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> config_value,
GetOptionalValue<std::string>(key));
if (config_value) {
PAIMON_ASSIGN_OR_RAISE(*value, MemorySize::ParseBytes(config_value.value()));
}
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>");
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> config_value,
GetOptionalValue<std::string>(key));
if (config_value) {
PAIMON_ASSIGN_OR_RAISE(*value, TimeDuration::Parse(config_value.value()));
}
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 {
PAIMON_ASSIGN_OR_RAISE(std::string identifier, OptionsUtils::GetValueFromMap<std::string>(
config_map_, key, default_identifier));
PAIMON_ASSIGN_OR_RAISE(*value,
Factory::Get(StringUtils::ToLowerCase(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();
}
PAIMON_ASSIGN_OR_RAISE(
std::string default_fs_identifier,
OptionsUtils::GetValueFromMap<std::string>(config_map_, Options::FILE_SYSTEM, "local"));
default_fs_identifier = StringUtils::ToLowerCase(default_fs_identifier);
*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 {
return ParseEnum(
Options::SEQUENCE_FIELD_SORT_ORDER,
{{"ascending", SortOrder::ASCENDING}, {"descending", SortOrder::DESCENDING}},
"sort order", sort_order);
}
// Parse LookupCompactMode
Status ParseLookupCompactMode(LookupCompactMode* mode) const {
return ParseEnum(
Options::LOOKUP_COMPACT,
{{"radical", LookupCompactMode::RADICAL}, {"gentle", LookupCompactMode::GENTLE}},
"lookup mode", mode);
}
// Parse SortEngine
Status ParseSortEngine(SortEngine* sort_engine) const {
return ParseEnum(
Options::SORT_ENGINE,
{{"min-heap", SortEngine::MIN_HEAP}, {"loser-tree", SortEngine::LOSER_TREE}},
"sort engine", sort_engine);
}
// Parse MergeEngine
Status ParseMergeEngine(MergeEngine* merge_engine) const {
return ParseEnum(Options::MERGE_ENGINE,
{{"deduplicate", MergeEngine::DEDUPLICATE},
{"partial-update", MergeEngine::PARTIAL_UPDATE},
{"aggregation", MergeEngine::AGGREGATE},
{"first-row", MergeEngine::FIRST_ROW}},
"merge engine", merge_engine);
}
// Parse VariantShreddingInferenceMode
Status ParseVariantShreddingInferenceMode(VariantShreddingInferenceMode* inference_mode) const {
return ParseEnum(Options::VARIANT_SHREDDING_INFERENCE_MODE,
{{"per-file", VariantShreddingInferenceMode::PER_FILE},
{"adaptive", VariantShreddingInferenceMode::ADAPTIVE}},
"variant shredding inference mode", inference_mode);
}
// Parse ChangelogProducer
Status ParseChangelogProducer(ChangelogProducer* changelog_producer) const {
return ParseEnum(Options::CHANGELOG_PRODUCER,
{{"none", ChangelogProducer::NONE},
{"input", ChangelogProducer::INPUT},
{"full-compaction", ChangelogProducer::FULL_COMPACTION},
{"lookup", ChangelogProducer::LOOKUP}},
"changelog producer", changelog_producer);
}
// Parse ExternalPathStrategy
Status ParseExternalPathStrategy(ExternalPathStrategy* external_path_strategy) const {
return ParseEnum(Options::DATA_FILE_EXTERNAL_PATHS_STRATEGY,
{{"none", ExternalPathStrategy::NONE},
{"specific-fs", ExternalPathStrategy::SPECIFIC_FS},
{"round-robin", ExternalPathStrategy::ROUND_ROBIN}},
"external path strategy", external_path_strategy);
}
// Parse BucketFunctionType
Status ParseBucketFunctionType(BucketFunctionType* bucket_function_type) const {
return ParseEnum(Options::BUCKET_FUNCTION_TYPE,
{{"default", BucketFunctionType::DEFAULT},
{"mod", BucketFunctionType::MOD},
{"hive", BucketFunctionType::HIVE}},
"bucket function type", bucket_function_type);
}
// Parse StartupMode
Status ParseStartupMode(StartupMode* startup_mode) const {
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> value,
GetOptionalValue<std::string>(Options::SCAN_MODE));
if (value) {
PAIMON_ASSIGN_OR_RAISE(
*startup_mode, StartupMode::FromString(StringUtils::ToLowerCase(value.value())));
}
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:
template <typename T>
Status ParseEnum(const std::string& key,
std::initializer_list<std::pair<const char*, T>> candidates,
const std::string& error_name, T* value) const {
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> config_value,
GetOptionalValue<std::string>(key));
if (!config_value) {
return Status::OK();
}
std::string normalized_value = StringUtils::ToLowerCase(config_value.value());
for (const auto& [candidate, candidate_value] : candidates) {
if (normalized_value == candidate) {
*value = candidate_value;
return Status::OK();
}
}
return Status::Invalid(fmt::format("invalid {}: {}", error_name, normalized_value));
}
template <typename T>
Result<std::optional<T>> GetOptionalValue(const std::string& key) const {
Result<std::optional<T>> result =
OptionsUtils::GetOptionalValueFromMap<T>(config_map_, key);
if (!result.ok()) {
return Status::Invalid(
fmt::format("Invalid Config [{}: {}]", key, config_map_.at(key)));
}
return result.value();
}
const std::map<std::string, std::string>& config_map_;
};
// Impl is a private implementation of CoreOptions,
// storing various configurable fields and their default values.
// The fields are grouped by the option each one parses, so that a new option lands beside the code
// that reads it. Reordering them by size would save a few dozen bytes in the one instance a table
// holds, at the cost of that grouping.
// NOLINTNEXTLINE(clang-analyzer-optin.performance.Padding)
struct CoreOptions::Impl {
int64_t page_size = 64 * 1024;
int64_t target_file_row_num = std::numeric_limits<int64_t>::max();
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 file_index_in_manifest_threshold = 500;
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;
int64_t realtime_read_view_ttl_millis = 5 * 60 * 1000;
int64_t write_buffer_spill_max_disk_size = std::numeric_limits<int64_t>::max();
double variant_shredding_min_field_cardinality_ratio = 0.1;
double variant_shredding_adaptive_retention_ratio = 0.05;
double lookup_cache_bloom_filter_fpp = 0.05;
int64_t cache_page_size = 64 * 1024; // 64KB
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;
std::optional<int64_t> target_file_size;
std::optional<int64_t> blob_target_file_size;
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;
std::optional<int64_t> optimized_compaction_interval;
std::optional<int64_t> compaction_total_size_threshold;
std::optional<int64_t> compaction_incremental_size_threshold;
std::shared_ptr<FileFormat> changelog_file_format;
ExpireConfig expire_config;
std::vector<std::string> sequence_field;
std::vector<std::string> remove_record_on_sequence_group;
std::vector<std::string> changelog_row_deduplicate_ignore_fields;
std::vector<std::string> blob_fields;
std::vector<std::string> blob_descriptor_fields;
std::vector<std::string> blob_view_fields;
std::vector<std::string> realtime_deduplicate_key_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 changelog_file_prefix = "changelog-";
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::optional<std::string> changelog_file_compression;
std::optional<std::string> global_index_external_path;
std::optional<std::string> scan_tag_name;
CompressOptions lookup_compress_options{"zstd", 1};
CompressOptions spill_compress_options{"zstd", 1};
std::map<std::string, std::string> raw_options;
std::map<int32_t, std::shared_ptr<FileFormat>> file_format_per_level;
std::map<int32_t, std::string> file_compression_per_level;
StatisticsMode realtime_store_statistics_mode = StatisticsMode::NONE;
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;
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;
BucketFunctionType bucket_function_type = BucketFunctionType::DEFAULT;
int32_t file_compression_zstd_level = 1;
CoreOptions::SequenceNumberInitMode write_sequence_number_init_mode =
CoreOptions::SequenceNumberInitMode::SCAN;
VariantShreddingInferenceMode variant_shredding_inference_mode =
VariantShreddingInferenceMode::PER_FILE;
int32_t variant_shredding_max_schema_width = 300;
int32_t variant_shredding_max_schema_depth = 50;
int32_t variant_shredding_max_infer_buffer_row = 4096;
int32_t variant_shredding_adaptive_max_infer_buffer_row = 256;
int32_t compact_off_peak_start_hour = -1;
int32_t compact_off_peak_end_hour = -1;
int32_t compact_off_peak_ratio = 0;
int32_t lookup_remote_level_threshold = INT32_MIN;
std::optional<int32_t> num_sorted_runs_stop_trigger;
std::optional<int32_t> num_levels;
std::optional<int32_t> lookup_compact_max_interval;
std::optional<int32_t> global_index_thread_num;
bool realtime_enabled = false;
bool realtime_spill_enabled = false;
bool scan_manifest_entry_lazy_decode_enabled = true;
bool ignore_delete = false;
bool manifest_delete_file_drop_stats = false;
bool write_buffer_spillable = true;
bool write_only = false;
bool bucket_append_ordered = false;
bool deletion_vectors_enabled = false;
bool pk_clustering_override = false;
bool deletion_vectors_bitmap64 = false;
bool force_lookup = false;
bool lookup_wait = true;
bool changelog_row_deduplicate = false;
bool input_changelog_parallel_write_enabled = 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 prefetch_io_metrics_enabled = false;
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 variant_infer_shredding_schema = false;
bool blob_view_resolve_enabled = true;
bool blob_as_descriptor = false;
bool legacy_partition_name_enabled = true;
bool file_suffix_include_compression = false;
bool format_table_partition_only_value_in_path = false;
bool metastore_partitioned_table = false;
bool global_index_enabled = true;
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;
bool lookup_cache_bloom_filter = true;
bool lookup_remote_file_enabled = false;
std::optional<bool> blob_split_by_file_size;
// 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 target-file-row-num - target rows of a newly written data file
PAIMON_RETURN_NOT_OK(
parser.Parse<int64_t>(Options::TARGET_FILE_ROW_NUM, &target_file_row_num));
if (target_file_row_num <= 0) {
return Status::Invalid(
fmt::format("{} should be at least 1", Options::TARGET_FILE_ROW_NUM));
}
// 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 changelog-file.prefix - file name prefix of changelog files, default "changelog-"
PAIMON_RETURN_NOT_OK(parser.Parse(Options::CHANGELOG_FILE_PREFIX, &changelog_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 changelog-file.format - no default value
if (parser.ContainsKey(Options::CHANGELOG_FILE_FORMAT)) {
PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
Options::CHANGELOG_FILE_FORMAT, file_format->Identifier(), &changelog_file_format));
}
// Parse changelog-file.compression - no default value
PAIMON_RETURN_NOT_OK(
parser.Parse(Options::CHANGELOG_FILE_COMPRESSION, &changelog_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));
// Parse file.suffix.include.compression - carry the compression in a data file's name
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::FILE_SUFFIX_INCLUDE_COMPRESSION,
&file_suffix_include_compression));
// Parse format-table.partition-path-only-value - name a partition directory by its value
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::FORMAT_TABLE_PARTITION_PATH_ONLY_VALUE,
&format_table_partition_only_value_in_path));
// Parse metastore.partitioned-table - partitions are registered with the catalog
PAIMON_RETURN_NOT_OK(
parser.Parse<bool>(Options::METASTORE_PARTITIONED_TABLE, &metastore_partitioned_table));
return Status::OK();
}
// Parse manifest file configurations: format, compression, merge, and compaction thresholds.
Status ParseManifestOptions(const ConfigParser& parser) {
// Parse the legacy manifest.format option for reading existing tables.
PAIMON_RETURN_NOT_OK(parser.ParseObject<FileFormatFactory>(
"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));
// Parse manifest.delete-file-drop-stats - drop stats from DELETE entries, default false
PAIMON_RETURN_NOT_OK(parser.Parse(Options::MANIFEST_DELETE_FILE_DROP_STATS,
&manifest_delete_file_drop_stats));
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.
// Java CoreOptions declares first-row.ignore-delete, deduplicate.ignore-delete
// and partial-update.ignore-delete as fallback keys, checked in that order only
// when ignore-delete itself is absent.
std::optional<bool> ignore_delete_value;
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::IGNORE_DELETE, &ignore_delete_value));
for (const char* fallback_key : {Options::FALLBACK_FIRST_ROW_IGNORE_DELETE,
Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE,
Options::FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE}) {
if (ignore_delete_value.has_value()) {
break;
}
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(fallback_key, &ignore_delete_value));
}
ignore_delete = ignore_delete_value.value_or(false);
// 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 changelog-producer.row-deduplicate - skip unchanged row changelogs
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE,
&changelog_row_deduplicate));
// Parse changelog-producer.row-deduplicate-ignore-fields - ignored comparison fields
PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS, Options::FIELDS_SEPARATOR,
&changelog_row_deduplicate_ignore_fields, /*need_trim=*/true));
// Parse changelog-producer.input.parallel-write - write data and input changelog
// files in parallel, default true.
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::CHANGELOG_PRODUCER_INPUT_PARALLEL_WRITE,
&input_changelog_parallel_write_enabled));
// 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));
PAIMON_RETURN_NOT_OK(
parser.Parse<bool>(Options::PK_CLUSTERING_OVERRIDE, &pk_clustering_override));
// 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));
}
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
&scan_manifest_entry_lazy_decode_enabled));
PAIMON_RETURN_NOT_OK(
parser.Parse<bool>(Options::PREFETCH_IO_METRICS_ENABLED, &prefetch_io_metrics_enabled));
// Parse scan.fallback-branch - fallback branch when partition not found
PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_FALLBACK_BRANCH, &scan_fallback_branch));
if (scan_fallback_branch) {
PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(scan_fallback_branch.value()));
}
// Parse branch - branch name, default "main"
PAIMON_RETURN_NOT_OK(parser.Parse(Options::BRANCH, &branch));
// Both branches name a directory under the table root, so they must stay a single path
// component.
PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(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 statistics collected by real-time stores.
Status ParseRealtimeStoreStatisticsMode(const ConfigParser& parser) {
std::string statistics_mode = "none";
PAIMON_RETURN_NOT_OK(parser.Parse(Options::REALTIME_STORE_STATS_MODE, &statistics_mode));
statistics_mode = StringUtils::ToLowerCase(statistics_mode);
if (statistics_mode == "none") {
realtime_store_statistics_mode = StatisticsMode::NONE;
} else if (statistics_mode == "full") {
realtime_store_statistics_mode = StatisticsMode::FULL;
} else {
return Status::Invalid(
fmt::format("{} must be 'none' or 'full'", Options::REALTIME_STORE_STATS_MODE));
}
return Status::OK();
}
// Parse real-time write and read configurations.
Status ParseRealtimeOptions(const ConfigParser& parser) {
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::REALTIME_ENABLED, &realtime_enabled));
PAIMON_RETURN_NOT_OK(
parser.Parse<bool>(Options::REALTIME_SPILL_ENABLED, &realtime_spill_enabled));
PAIMON_RETURN_NOT_OK(parser.ParseList<std::string>(
Options::REALTIME_DEDUPLICATE_KEY_FIELDS, Options::FIELDS_SEPARATOR,
&realtime_deduplicate_key_fields, /*need_trim=*/true));
PAIMON_RETURN_NOT_OK(parser.ParseTimeDuration(Options::REALTIME_READ_VIEW_TTL,
&realtime_read_view_ttl_millis));
if (realtime_read_view_ttl_millis <= 0) {
return Status::Invalid(
fmt::format("{} must be positive", Options::REALTIME_READ_VIEW_TTL));
}
return ParseRealtimeStoreStatisticsMode(parser);
}
// Parse index-related configurations: file index, global index.
Status ParseIndexOptions(const ConfigParser& parser) {
// Parse file-index.in-manifest-threshold - max inline file index size, default 500B
PAIMON_RETURN_NOT_OK(parser.ParseMemorySize(Options::FILE_INDEX_IN_MANIFEST_THRESHOLD,
&file_index_in_manifest_threshold));
// 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 ParseVariantOptions(const ConfigParser& parser) {
// Parse variant.inferShreddingSchema - infer the shredding schema from sampled rows
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::VARIANT_INFER_SHREDDING_SCHEMA,
&variant_infer_shredding_schema));
PAIMON_RETURN_NOT_OK(
parser.ParseVariantShreddingInferenceMode(&variant_shredding_inference_mode));
// Parse variant.shredding.maxSchemaWidth - max number of shredded fields, default 300
PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::VARIANT_SHREDDING_MAX_SCHEMA_WIDTH,
&variant_shredding_max_schema_width));
// Parse variant.shredding.maxSchemaDepth - max shredded nesting depth, default 50
PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::VARIANT_SHREDDING_MAX_SCHEMA_DEPTH,
&variant_shredding_max_schema_depth));
// Parse variant.shredding.minFieldCardinalityRatio - min occurrence ratio for a field to
// be shredded, default 0.1
PAIMON_RETURN_NOT_OK(
parser.Parse<double>(Options::VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO,
&variant_shredding_min_field_cardinality_ratio));
// Parse variant.shredding.maxInferBufferRow - rows buffered per file for inference,
// default 4096
PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(Options::VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW,
&variant_shredding_max_infer_buffer_row));
PAIMON_RETURN_NOT_OK(
parser.Parse<int32_t>(Options::VARIANT_SHREDDING_ADAPTIVE_MAX_INFER_BUFFER_ROW,
&variant_shredding_adaptive_max_infer_buffer_row));
PAIMON_RETURN_NOT_OK(
parser.Parse<double>(Options::VARIANT_SHREDDING_ADAPTIVE_RETENTION_RATIO,
&variant_shredding_adaptive_retention_ratio));
if (variant_shredding_max_schema_width <= 0) {
return Status::Invalid(fmt::format(
"The option '{}' should be positive, while input is {}",
Options::VARIANT_SHREDDING_MAX_SCHEMA_WIDTH, variant_shredding_max_schema_width));
}
if (variant_shredding_max_schema_depth <= 0) {
return Status::Invalid(fmt::format(
"The option '{}' should be positive, while input is {}",
Options::VARIANT_SHREDDING_MAX_SCHEMA_DEPTH, variant_shredding_max_schema_depth));
}
if (variant_shredding_min_field_cardinality_ratio < 0.0 ||
variant_shredding_min_field_cardinality_ratio > 1.0) {
return Status::Invalid(
fmt::format("The option '{}' should be in the range [0, 1], while input is {}",
Options::VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO,
variant_shredding_min_field_cardinality_ratio));
}
if (variant_shredding_max_infer_buffer_row <= 0) {
return Status::Invalid(
fmt::format("The option '{}' should be positive, while input is {}",
Options::VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW,
variant_shredding_max_infer_buffer_row));
}
if (variant_shredding_inference_mode == VariantShreddingInferenceMode::ADAPTIVE) {
if (variant_shredding_adaptive_max_infer_buffer_row <= 0) {
return Status::Invalid(
fmt::format("The option '{}' should be positive, while input is {}",
Options::VARIANT_SHREDDING_ADAPTIVE_MAX_INFER_BUFFER_ROW,
variant_shredding_adaptive_max_infer_buffer_row));
}
if (variant_shredding_adaptive_retention_ratio < 0.0 ||
variant_shredding_adaptive_retention_ratio >
variant_shredding_min_field_cardinality_ratio) {
return Status::Invalid(
fmt::format("The option '{}' should be in the range [0, {}], while input is {}",
Options::VARIANT_SHREDDING_ADAPTIVE_RETENTION_RATIO,
Options::VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO,
variant_shredding_adaptive_retention_ratio));
}
}
return Status::OK();
}
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->ParseRealtimeOptions(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));
PAIMON_RETURN_NOT_OK(impl->ParseVariantOptions(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::GetTargetFileRowNum() const {
return impl_->target_file_row_num;
}
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;
}
bool CoreOptions::RealtimeEnabled() const {
return impl_->realtime_enabled;
}
bool CoreOptions::RealtimeSpillEnabled() const {
return impl_->realtime_spill_enabled;
}
int64_t CoreOptions::GetRealtimeReadViewTtlMillis() const {
return impl_->realtime_read_view_ttl_millis;
}
StatisticsMode CoreOptions::GetRealtimeStoreStatisticsMode() const {
return impl_->realtime_store_statistics_mode;
}
const std::vector<std::string>& CoreOptions::GetRealtimeDeduplicateKeyFields() const {
return impl_->realtime_deduplicate_key_fields;
}
int32_t CoreOptions::GetScanManifestEntryCacheMaxSnapshots() const {
return impl_->scan_manifest_entry_cache_max_snapshots;
}
bool CoreOptions::ScanManifestEntryLazyDecodeEnabled() const {
return impl_->scan_manifest_entry_lazy_decode_enabled;
}
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;
}
bool CoreOptions::ManifestDeleteFileDropStats() const {
return impl_->manifest_delete_file_drop_stats;
}
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;
}
bool CoreOptions::PrefetchIoMetricsEnabled() const {
return impl_->prefetch_io_metrics_enabled;
}
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::vector<std::string>> CoreOptions::FieldNestedUpdateAggNestedKey(
const std::string& field_name) const {
ConfigParser parser(impl_->raw_options);
std::vector<std::string> nested_key;
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
std::string(Options::NESTED_KEY);
PAIMON_RETURN_NOT_OK(
parser.ParseList<std::string>(key, Options::FIELDS_SEPARATOR, &nested_key, true));
return nested_key;
}
Result<CoreOptions::NestedKeyNullStrategy> CoreOptions::FieldNestedUpdateAggNestedKeyNullStrategy(
const std::string& field_name) const {
ConfigParser parser(impl_->raw_options);
std::string strategy = "merge";
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
std::string(Options::NESTED_KEY_NULL_STRATEGY);
PAIMON_RETURN_NOT_OK(parser.Parse(key, &strategy));
std::string lower = StringUtils::ToLowerCase(strategy);
if (lower == "merge") {
return NestedKeyNullStrategy::MERGE;
}
if (lower == "ignore") {
return NestedKeyNullStrategy::IGNORE;
}
if (lower == "error") {
return NestedKeyNullStrategy::ERROR;
}
return Status::Invalid(fmt::format(
"Invalid Config [{}: {}], supported values are merge, ignore and error", key, strategy));
}
Result<std::vector<std::string>> CoreOptions::FieldNestedUpdateAggNestedSequenceField(
const std::string& field_name) const {
ConfigParser parser(impl_->raw_options);
std::vector<std::string> sequence_fields;
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
std::string(Options::NESTED_SEQUENCE_FIELD);
PAIMON_RETURN_NOT_OK(
parser.ParseList<std::string>(key, Options::FIELDS_SEPARATOR, &sequence_fields, true));
return sequence_fields;
}
Result<int32_t> CoreOptions::FieldNestedUpdateAggCountLimit(const std::string& field_name) const {
ConfigParser parser(impl_->raw_options);
int32_t count_limit = std::numeric_limits<int32_t>::max();
std::string key = std::string(Options::FIELDS_PREFIX) + "." + field_name + "." +
std::string(Options::COUNT_LIMIT);
PAIMON_RETURN_NOT_OK(parser.Parse<int32_t>(key, &count_limit));
if (count_limit < 0) {
return Status::Invalid(
fmt::format("Invalid Config [{}: {}], must not be negative", key, count_limit));
}
return count_limit;
}
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;
}
std::optional<std::string> CoreOptions::GetVariantShreddingSchema() const {
auto it = impl_->raw_options.find(Options::VARIANT_SHREDDING_SCHEMA);
if (it == impl_->raw_options.end()) {
it = impl_->raw_options.find(Options::PARQUET_VARIANT_SHREDDING_SCHEMA);
}
if (it == impl_->raw_options.end() || it->second.empty()) {
return std::nullopt;
}
return it->second;
}
bool CoreOptions::VariantInferShreddingSchemaEnabled() const {
return impl_->variant_infer_shredding_schema;
}
VariantShreddingInferenceMode CoreOptions::GetVariantShreddingInferenceMode() const {
return impl_->variant_shredding_inference_mode;
}
int32_t CoreOptions::GetVariantShreddingMaxSchemaWidth() const {
return impl_->variant_shredding_max_schema_width;
}
int32_t CoreOptions::GetVariantShreddingMaxSchemaDepth() const {
return impl_->variant_shredding_max_schema_depth;
}
double CoreOptions::GetVariantShreddingMinFieldCardinalityRatio() const {
return impl_->variant_shredding_min_field_cardinality_ratio;
}
int32_t CoreOptions::GetVariantShreddingMaxInferBufferRow() const {
return impl_->variant_shredding_max_infer_buffer_row;
}
int32_t CoreOptions::GetVariantShreddingAdaptiveMaxInferBufferRow() const {
return impl_->variant_shredding_adaptive_max_infer_buffer_row;
}
double CoreOptions::GetVariantShreddingAdaptiveRetentionRatio() const {
return impl_->variant_shredding_adaptive_retention_ratio;
}
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::PkClusteringOverrideEnabled() const {
return impl_->pk_clustering_override;
}
std::vector<std::string> CoreOptions::GetPrimaryKeyBTreeIndexColumns() const {
auto iter = ToMap().find(Options::PK_BTREE_INDEX_COLUMNS);
if (iter == ToMap().end()) {
return {};
}
std::vector<std::string> columns = StringUtils::Split(iter->second, ",", false);
for (std::string& column : columns) {
StringUtils::Trim(&column);
}
return columns;
}
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;
}
bool CoreOptions::ChangelogRowDeduplicate() const {
return impl_->changelog_row_deduplicate;
}
const std::vector<std::string>& CoreOptions::GetChangelogRowDeduplicateIgnoreFields() const {
return impl_->changelog_row_deduplicate_ignore_fields;
}
bool CoreOptions::InputChangelogParallelWriteEnabled() const {
return impl_->input_changelog_parallel_write_enabled;
}
std::string CoreOptions::ChangelogFilePrefix() const {
return impl_->changelog_file_prefix;
}
std::shared_ptr<FileFormat> CoreOptions::GetChangelogFileFormat() const {
return impl_->changelog_file_format;
}
std::optional<std::string> CoreOptions::GetChangelogFileCompression() const {
return impl_->changelog_file_compression;
}
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;
}
int64_t CoreOptions::FileIndexInManifestThreshold() const {
return impl_->file_index_in_manifest_threshold;
}
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::FileSuffixIncludeCompression() const {
return impl_->file_suffix_include_compression;
}
bool CoreOptions::FormatTablePartitionOnlyValueInPath() const {
return impl_->format_table_partition_only_value_in_path;
}
bool CoreOptions::MetastorePartitionedTable() const {
return impl_->metastore_partitioned_table;
}
std::string CoreOptions::FormatTableFileCompression() const {
// The resolution order the rest of the paimon ecosystem follows; both the compression suffix
// in a file's name and its contents derive from it, so they cannot disagree. `compression` is
// not a paimon option of its own but the key an engine's own writer reads: paimon-spark copies
// `format-table.file.compression` onto it, so a table written that way carries only that key.
const char* const keys[] = {Options::FILE_COMPRESSION, Options::FORMAT_TABLE_FILE_COMPRESSION,
"compression"};
for (const char* key : keys) {
auto iter = impl_->raw_options.find(key);
if (iter != impl_->raw_options.end()) {
return iter->second;
}
}
// What the format writes by default, when no option names a compression.
return impl_->file_format != nullptr && impl_->file_format->Identifier() == "parquet" ? "snappy"
: "zstd";
}
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