#include <Disks/DiskType.h>

#include <Common/CurrentThread.h>

#include <Storages/PartitionCommands.h>

#include <Storages/MergeTree/MergeTreeData.h>

#include <Access/AccessControl.h>

#include <AggregateFunctions/AggregateFunctionCount.h>

#include <Analyzer/QueryTreeBuilder.h>

#include <Analyzer/Utils.h>

#include <Backups/BackupEntriesCollector.h>

#include <Backups/BackupEntryWrappedWith.h>

#include <Backups/IBackup.h>

#include <Backups/RestorerFromBackup.h>

#include <Columns/ColumnAggregateFunction.h>

#include <Common/Config/ConfigHelper.h>

#include <Common/CurrentMetrics.h>

#include <Common/Increment.h>

#include <Common/ProfileEventsScope.h>

#include <Common/Stopwatch.h>

#include <Common/StringUtils.h>

#include <Common/ThreadFuzzer.h>

#include <Common/ZooKeeper/ZooKeeperCommon.h>

#include <Common/escapeForFileName.h>

#include <Common/noexcept_scope.h>

#include <Common/quoteString.h>

#include <Common/typeid_cast.h>

#include <Common/thread_local_rng.h>

#include <Core/BackgroundSchedulePool.h>

#include <Core/Settings.h>

#include <Core/ServerSettings.h>

#include <Storages/MergeTree/RangesInDataPart.h>

#include <Compression/CompressionFactory.h>

#include <Core/QueryProcessingStage.h>

#include <DataTypes/DataTypeCustomSimpleAggregateFunction.h>

#include <DataTypes/DataTypeEnum.h>

#include <DataTypes/DataTypeLowCardinality.h>

#include <DataTypes/DataTypeTuple.h>

#include <DataTypes/DataTypeUUID.h>

#include <DataTypes/NestedUtils.h>

#include <DataTypes/hasNullable.h>

#include <Disks/SingleDiskVolume.h>

#include <Disks/TemporaryFileOnDisk.h>

#include <Disks/createVolume.h>

#include <IO/Operators.h>

#include <IO/S3Common.h>

#include <IO/SharedThreadPools.h>

#include <IO/WriteBufferFromString.h>

#include <IO/WriteHelpers.h>

#include <Interpreters/Aggregator.h>

#include <Interpreters/Context.h>

#include <Interpreters/convertFieldToType.h>

#include <Interpreters/DatabaseCatalog.h>

#include <Interpreters/evaluateConstantExpression.h>

#include <Interpreters/ExpressionAnalyzer.h>

#include <Interpreters/ExpressionActions.h>

#include <Interpreters/InterpreterSelectQuery.h>

#include <Interpreters/MergeTreeTransaction.h>

#include <Interpreters/PartLog.h>

#include <Interpreters/TransactionLog.h>

#include <Interpreters/TreeRewriter.h>

#include <Interpreters/inplaceBlockConversions.h>

#include <Interpreters/MutationsInterpreter.h>

#include <Parsers/ASTExpressionList.h>

#include <Parsers/ASTIndexDeclaration.h>

#include <Parsers/ASTHelpers.h>

#include <Parsers/ASTFunction.h>

#include <Parsers/ASTLiteral.h>

#include <Parsers/ASTPartition.h>

#include <Parsers/ASTSetQuery.h>

#include <Processors/Formats/IInputFormat.h>

#include <Processors/QueryPlan/QueryIdHolder.h>

#include <Processors/QueryPlan/ReadFromMergeTree.h>

#include <Processors/Transforms/SquashingTransform.h>

#include <Processors/Transforms/DeduplicationTokenTransforms.h>

#include <Storages/AlterCommands.h>

#include <Storages/MergeTree/MergeTreeVirtualColumns.h>

#include <Storages/Freeze.h>

#include <Storages/MergeTree/DataPartStorageOnDiskFull.h>

#include <Storages/MergeTree/MergeTreeDataPartBuilder.h>

#include <Storages/MergeTree/MergeTreeSettings.h>

#include <Storages/MergeTree/PrimaryIndexCache.h>

#include <Storages/Statistics/ConditionSelectivityEstimator.h>

#include <Storages/MergeTree/checkDataPart.h>

#include <Storages/MutationCommands.h>

#include <Storages/MergeTree/ActiveDataPartSet.h>

#include <Storages/StorageReplicatedMergeTree.h>

#include <Storages/VirtualColumnUtils.h>

#include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>

#include <QueryPipeline/QueryPipelineBuilder.h>

#include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>

#include <Storages/MergeTree/MergeTreeDataMergerMutator.h>

#include <Storages/MergeTree/FutureMergedMutatedPart.h>

#include <Storages/MergeTree/Compaction/CompactionStatistics.h>

#include <Storages/MergeTree/Compaction/PartProperties.h>

#include <Storages/MergeTree/Compaction/ConstructFuturePart.h>

#include <Storages/MergeTree/Compaction/MergeSelectorApplier.h>

#include <Storages/MergeTree/PatchParts/PatchPartsUtils.h>

#include <Storages/MergeTree/Compaction/PartsCollectors/Common.h>

#include <boost/algorithm/string/join.hpp>

#include <base/insertAtEnd.h>

#include <base/interpolate.h>

#include <base/isSharedPtrUnique.h>

#include <algorithm>

#include <atomic>

#include <chrono>

#include <exception>

#include <limits>

#include <optional>

#include <ranges>

#include <set>

#include <thread>

#include <unordered_set>

#include <filesystem>

#include <boost/container_hash/hash.hpp>

#include <fmt/format.h>

#include <Poco/Net/NetException.h>

#if USE_AZURE_BLOB_STORAGE

#endif

template <>

struct fmt::formatter<DB::DataPartPtr> : fmt::formatter<std::string>

{

template <typename FormatCtx>

auto format(const DB::DataPartPtr & part, FormatCtx & ctx) const

{

return fmt::formatter<std::string>::format(part->name, ctx);

}

};

namespace fs = std::filesystem;

namespace ProfileEvents

{

extern const Event RejectedInserts;

extern const Event DelayedInserts;

extern const Event DelayedInsertsMilliseconds;

extern const Event InsertedWideParts;

extern const Event InsertedCompactParts;

extern const Event MergedIntoWideParts;

extern const Event MergedIntoCompactParts;

extern const Event RejectedMutations;

extern const Event DelayedMutations;

extern const Event DelayedMutationsMilliseconds;

extern const Event RejectedLightweightUpdates;

extern const Event PartsLockWaitMicroseconds;

extern const Event PartsLockHoldMicroseconds;

extern const Event PartsLocks;

extern const Event SharedPartsLockWaitMicroseconds;

extern const Event SharedPartsLockHoldMicroseconds;

extern const Event SharedPartsLocks;

extern const Event LoadedDataParts;

extern const Event LoadedDataPartsMicroseconds;

extern const Event RestorePartsSkippedFiles;

extern const Event RestorePartsSkippedBytes;

extern const Event LoadedStatisticsMicroseconds;

}

namespace CurrentMetrics

{

extern const Metric DelayedInserts;

extern const Metric FreezePartThreads;

extern const Metric FreezePartThreadsActive;

extern const Metric FreezePartThreadsScheduled;

extern const Metric ColumnsDescriptionsCacheSize;

}

namespace

{

constexpr UInt64 RESERVATION_MIN_ESTIMATION_SIZE = 1u * 1024u * 1024u; /// 1MB

}

namespace DB

{

namespace Setting

{

extern const SettingsBool allow_drop_detached;

extern const SettingsBool allow_experimental_analyzer;

extern const SettingsBool enable_full_text_index;

extern const SettingsBool allow_non_metadata_alters;

extern const SettingsBool allow_suspicious_indices;

extern const SettingsBool alter_move_to_space_execute_async;

extern const SettingsBool alter_partition_verbose_result;

extern const SettingsBool apply_mutations_on_fly;

extern const SettingsBool fsync_metadata;

extern const SettingsSeconds lock_acquire_timeout;

extern const SettingsBool optimize_dry_run_check_part;

extern const SettingsBool materialize_ttl_after_modify;

extern const SettingsUInt64 max_partition_size_to_drop;

extern const SettingsMaxThreads max_threads;

extern const SettingsUInt64 number_of_mutations_to_delay;

extern const SettingsUInt64 number_of_mutations_to_throw;

extern const SettingsBool parallel_replicas_for_non_replicated_merge_tree;

extern const SettingsUInt64 parts_to_delay_insert;

extern const SettingsUInt64 parts_to_throw_insert;

extern const SettingsBool enable_shared_storage_snapshot_in_query;

extern const SettingsUInt64 merge_tree_storage_snapshot_sleep_ms;

extern const SettingsUInt64 min_insert_block_size_rows;

extern const SettingsUInt64 min_insert_block_size_bytes;

extern const SettingsBool apply_patch_parts;

extern const SettingsUInt64 max_table_size_to_drop;

extern const SettingsBool use_statistics;

extern const SettingsBool use_statistics_cache;

extern const SettingsBool use_partition_pruning;

}

namespace MergeTreeSetting

{

extern const MergeTreeSettingsBool allow_experimental_reverse_key;

extern const MergeTreeSettingsBool allow_nullable_key;

extern const MergeTreeSettingsBool allow_remote_fs_zero_copy_replication;

extern const MergeTreeSettingsBool allow_suspicious_indices;

extern const MergeTreeSettingsBool allow_summing_columns_in_partition_or_order_key;

extern const MergeTreeSettingsBool allow_coalescing_columns_in_partition_or_order_key;

extern const MergeTreeSettingsBool assign_part_uuids;

extern const MergeTreeSettingsBool async_insert;

extern const MergeTreeSettingsBool check_sample_column_is_correct;

extern const MergeTreeSettingsBool compatibility_allow_sampling_expression_not_in_primary_key;

extern const MergeTreeSettingsAlterColumnSecondaryIndexMode alter_column_secondary_index_mode;

extern const MergeTreeSettingsUInt64 concurrent_part_removal_threshold;

extern const MergeTreeSettingsDeduplicateMergeProjectionMode deduplicate_merge_projection_mode;

extern const MergeTreeSettingsBool disable_freeze_partition_for_zero_copy_replication;

extern const MergeTreeSettingsString disk;

extern const MergeTreeSettingsBool table_disk;

extern const MergeTreeSettingsBool enable_mixed_granularity_parts;

extern const MergeTreeSettingsBool escape_index_filenames;

extern const MergeTreeSettingsBool fsync_after_insert;

extern const MergeTreeSettingsBool fsync_part_directory;

extern const MergeTreeSettingsUInt64 inactive_parts_to_delay_insert;

extern const MergeTreeSettingsUInt64 inactive_parts_to_throw_insert;

extern const MergeTreeSettingsUInt64 index_granularity;

extern const MergeTreeSettingsUInt64 index_granularity_bytes;

extern const MergeTreeSettingsSeconds lock_acquire_timeout_for_background_operations;

extern const MergeTreeSettingsUInt64 max_avg_part_size_for_too_many_parts;

extern const MergeTreeSettingsUInt64 max_delay_to_insert;

extern const MergeTreeSettingsUInt64 max_delay_to_mutate_ms;

extern const MergeTreeSettingsUInt64 max_file_name_length;

extern const MergeTreeSettingsUInt64 max_parts_in_total;

extern const MergeTreeSettingsUInt64 max_projections;

extern const MergeTreeSettingsUInt64 max_suspicious_broken_parts_bytes;

extern const MergeTreeSettingsUInt64 max_suspicious_broken_parts;

extern const MergeTreeSettingsUInt64 min_bytes_for_wide_part;

extern const MergeTreeSettingsUInt64 min_bytes_to_rebalance_partition_over_jbod;

extern const MergeTreeSettingsUInt64 min_delay_to_insert_ms;

extern const MergeTreeSettingsUInt64 min_delay_to_mutate_ms;

extern const MergeTreeSettingsUInt64 min_rows_for_wide_part;

extern const MergeTreeSettingsUInt64 number_of_mutations_to_delay;

extern const MergeTreeSettingsUInt64 number_of_mutations_to_throw;

extern const MergeTreeSettingsSeconds old_parts_lifetime;

extern const MergeTreeSettingsUInt64 part_moves_between_shards_enable;

extern const MergeTreeSettingsUInt64 parts_to_delay_insert;

extern const MergeTreeSettingsUInt64 parts_to_throw_insert;

extern const MergeTreeSettingsFloat ratio_of_defaults_for_sparse_serialization;

extern const MergeTreeSettingsBool remove_empty_parts;

extern const MergeTreeSettingsBool remove_rolled_back_parts_immediately;

extern const MergeTreeSettingsBool replace_long_file_name_to_hash;

extern const MergeTreeSettingsUInt64 simultaneous_parts_removal_limit;

extern const MergeTreeSettingsUInt64 sleep_before_loading_outdated_parts_ms;

extern const MergeTreeSettingsString storage_policy;

extern const MergeTreeSettingsFloat zero_copy_concurrent_part_removal_max_postpone_ratio;

extern const MergeTreeSettingsUInt64 zero_copy_concurrent_part_removal_max_split_times;

extern const MergeTreeSettingsBool table_readonly;

extern const MergeTreeSettingsBool use_primary_key_cache;

extern const MergeTreeSettingsBool prewarm_primary_key_cache;

extern const MergeTreeSettingsBool prewarm_mark_cache;

extern const MergeTreeSettingsBool primary_key_lazy_load;

extern const MergeTreeSettingsBool apply_patches_on_merge;

extern const MergeTreeSettingsBool enforce_index_structure_match_on_partition_manipulation;

extern const MergeTreeSettingsUInt64 min_bytes_to_prewarm_caches;

extern const MergeTreeSettingsBool enable_block_number_column;

extern const MergeTreeSettingsBool enable_block_offset_column;

extern const MergeTreeSettingsBool allow_commit_order_projection;

extern const MergeTreeSettingsBool columns_and_secondary_indices_sizes_lazy_calculation;

extern const MergeTreeSettingsSeconds refresh_parts_interval;

extern const MergeTreeSettingsSeconds refresh_statistics_interval;

extern const MergeTreeSettingsBool remove_unused_patch_parts;

extern const MergeTreeSettingsSearchOrphanedPartsDisks search_orphaned_parts_disks;

extern const MergeTreeSettingsBool allow_part_offset_column_in_projections;

extern const MergeTreeSettingsUInt64 max_uncompressed_bytes_in_patches;

extern const MergeTreeSettingsString auto_statistics_types;

extern const MergeTreeSettingsMergeTreeSerializationInfoVersion serialization_info_version;

extern const MergeTreeSettingsMergeTreeStringSerializationVersion string_serialization_version;

extern const MergeTreeSettingsMergeTreeNullableSerializationVersion nullable_serialization_version;

extern const MergeTreeSettingsMergeTreeMapSerializationVersion map_serialization_version;

extern const MergeTreeSettingsUInt32 min_level_for_wide_part;

extern const MergeTreeSettingsBool propagate_types_serialization_versions_to_nested_types;

}

namespace ServerSetting

{

extern const ServerSettingsDouble mark_cache_prewarm_ratio;

extern const ServerSettingsDouble primary_index_cache_prewarm_ratio;

extern const ServerSettingsDouble index_mark_cache_prewarm_ratio;

}

namespace ErrorCodes

{

extern const int NO_SUCH_DATA_PART;

extern const int NOT_IMPLEMENTED;

extern const int DIRECTORY_ALREADY_EXISTS;

extern const int TOO_MANY_UNEXPECTED_DATA_PARTS;

extern const int DUPLICATE_DATA_PART;

extern const int NO_SUCH_COLUMN_IN_TABLE;

extern const int LOGICAL_ERROR;

extern const int ILLEGAL_COLUMN;

extern const int ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER;

extern const int CORRUPTED_DATA;

extern const int BAD_TYPE_OF_FIELD;

extern const int BAD_ARGUMENTS;

extern const int INVALID_PARTITION_VALUE;

extern const int METADATA_MISMATCH;

extern const int PART_IS_TEMPORARILY_LOCKED;

extern const int TOO_MANY_PARTS;

extern const int INCOMPATIBLE_COLUMNS;

extern const int BAD_TTL_EXPRESSION;

extern const int INCORRECT_FILE_NAME;

extern const int BAD_DATA_PART_NAME;

extern const int READONLY_SETTING;

extern const int ABORTED;

extern const int UNKNOWN_DISK;

extern const int NOT_ENOUGH_SPACE;

extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;

extern const int SUPPORT_IS_DISABLED;

extern const int TOO_MANY_SIMULTANEOUS_QUERIES;

extern const int INCORRECT_QUERY;

extern const int INVALID_SETTING_VALUE;

extern const int CANNOT_RESTORE_TABLE;

extern const int NOT_INITIALIZED;

extern const int SERIALIZATION_ERROR;

extern const int TOO_MANY_MUTATIONS;

extern const int CANNOT_SCHEDULE_TASK;

extern const int LIMIT_EXCEEDED;

extern const int CANNOT_FORGET_PARTITION;

extern const int DATA_TYPE_CANNOT_BE_USED_IN_KEY;

extern const int TOO_LARGE_LIGHTWEIGHT_UPDATES;

}

static String getPartNameFromAST(const ASTPtr & partition)

{

const auto * literal = partition->as<ASTLiteral>();

if (!literal)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected a string literal for part name, got: {}", partition->formatForErrorMessage());

return literal->value.safeGet<String>();

}

static void checkSuspiciousIndices(const ASTFunction * index_function)

{

std::unordered_set<UInt64> unique_index_expression_hashes;

for (const auto & child : index_function->arguments->children)

{

const IASTHash hash = child->getTreeHash(/*ignore_aliases=*/ true);

const auto & first_half_of_hash = hash.low64;

if (!unique_index_expression_hashes.emplace(first_half_of_hash).second)

throw Exception(ErrorCodes::BAD_ARGUMENTS,

"Primary key or secondary index contains a duplicate expression. To suppress this exception, rerun the command with setting 'allow_suspicious_indices = 1'");

}

}

static void checkSampleExpression(const StorageInMemoryMetadata & metadata, bool allow_sampling_expression_not_in_primary_key, bool check_sample_column_is_correct)

{

if (metadata.sampling_key.column_names.empty())

throw Exception(ErrorCodes::INCORRECT_QUERY, "There are no columns in sampling expression");

const auto & pk_sample_block = metadata.getPrimaryKey().sample_block;

if (!pk_sample_block.has(metadata.sampling_key.column_names[0]) && !allow_sampling_expression_not_in_primary_key)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Sampling expression must be present in the primary key");

if (!check_sample_column_is_correct)

return;

const auto & sampling_key = metadata.getSamplingKey();

DataTypePtr sampling_column_type = sampling_key.data_types[0];

bool is_correct_sample_condition = false;

if (sampling_key.data_types.size() == 1)

{

if (typeid_cast<const DataTypeUInt64 *>(sampling_column_type.get()))

is_correct_sample_condition = true;

else if (typeid_cast<const DataTypeUInt32 *>(sampling_column_type.get()))

is_correct_sample_condition = true;

else if (typeid_cast<const DataTypeUInt16 *>(sampling_column_type.get()))

is_correct_sample_condition = true;

else if (typeid_cast<const DataTypeUInt8 *>(sampling_column_type.get()))

is_correct_sample_condition = true;

}

if (!is_correct_sample_condition)

throw Exception(ErrorCodes::ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER,

"Invalid sampling column type in storage parameters: {}. Must be one unsigned integer type",

sampling_column_type->getName());

}

static bool hasColumnsWithDynamicSubcolumns(const Block & block)

{

for (const auto & column : block.getColumnsWithTypeAndName())

{

if (column.type->hasDynamicSubcolumns())

return true;

}

return false;

}

void MergeTreeData::initializeDirectoriesAndFormatVersion(const std::string & relative_data_path_, bool attach, const std::string & date_column_name, bool need_create_directories)

{

auto settings = getSettings();

if ((*settings)[MergeTreeSetting::table_disk])

{

LOG_TRACE(log, "Table is located directly on disk (without database layer)");

}

else

{

relative_data_path = relative_data_path_;

if (relative_data_path.empty())

throw Exception(ErrorCodes::INCORRECT_FILE_NAME, "MergeTree storages require data path");

}

MergeTreeDataFormatVersion min_format_version(0);

if (date_column_name.empty())

min_format_version = MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING;

const auto format_version_path = fs::path(relative_data_path) / MergeTreeData::FORMAT_VERSION_FILE_NAME;

std::optional<UInt32> read_format_version;

for (const auto & disk : getDisks())

{

if (disk->isBroken())

continue;

if (need_create_directories && !disk->isReadOnly())

{

disk->createDirectories(relative_data_path);

disk->createDirectories(fs::path(relative_data_path) / DETACHED_DIR_NAME);

}

if (auto buf = disk->readFileIfExists(format_version_path, getReadSettings()))

{

UInt32 current_format_version{0};

readIntText(current_format_version, *buf);

if (!buf->eof())

throw Exception(ErrorCodes::CORRUPTED_DATA, "Bad version file: {}", fullPath(disk, format_version_path));

if (!read_format_version.has_value())

read_format_version = current_format_version;

else if (*read_format_version != current_format_version)

throw Exception(ErrorCodes::CORRUPTED_DATA,

"Version file on {} contains version {} expected version is {}.",

fullPath(disk, format_version_path), current_format_version, *read_format_version);

}

}

/// When data path or file not exists, ignore the format_version check

if (!attach || !read_format_version)

{

format_version = min_format_version;

/// Try to write to first non-readonly disk

for (const auto & disk : getStoragePolicy()->getDisks())

{

if (disk->isBroken())

continue;

/// Write once disk is almost the same as read-only for MergeTree,

/// since it does not support move, that is required for any

/// operation over MergeTree, so avoid writing format_version.txt

/// into it as well, to avoid leaving it after DROP.

if (!disk->isReadOnly() && !disk->isWriteOnce())

{

auto buf = disk->writeFile(format_version_path, 16, WriteMode::Rewrite, getContext()->getWriteSettings());

writeIntText(format_version.toUnderType(), *buf);

buf->finalize();

if (getContext()->getSettingsRef()[Setting::fsync_metadata])

buf->sync();

}

break;

}

}

else

{

format_version = *read_format_version;

}

if (format_version < min_format_version)

{

if (min_format_version == MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING.toUnderType())

throw Exception(ErrorCodes::METADATA_MISMATCH, "MergeTree data format version on disk doesn't support custom partitioning");

}

}

DataPartsLock::DataPartsLock(SharedMutex & data_parts_mutex_, const MergeTreeData * data_)

: wait_watch(Stopwatch(CLOCK_MONOTONIC))

, lock(data_parts_mutex_)

, lock_watch(Stopwatch(CLOCK_MONOTONIC))

, data(data_)

{

ProfileEvents::increment(ProfileEvents::PartsLockWaitMicroseconds, wait_watch->elapsedMicroseconds());

ProfileEvents::increment(ProfileEvents::PartsLocks);

}

DataPartsLock::~DataPartsLock()

{

if (data)

{

data->shared_parts_list.reset();

data->shared_ranges_in_parts.reset();

}

if (lock_watch.has_value())

ProfileEvents::increment(ProfileEvents::PartsLockHoldMicroseconds, lock_watch->elapsedMicroseconds());

}

DataPartsSharedLock::DataPartsSharedLock(DB::SharedMutex & data_parts_mutex_)

: wait_watch(Stopwatch(CLOCK_MONOTONIC))

, lock(data_parts_mutex_)

, lock_watch(Stopwatch(CLOCK_MONOTONIC))

{

ProfileEvents::increment(ProfileEvents::SharedPartsLockWaitMicroseconds, wait_watch->elapsedMicroseconds());

ProfileEvents::increment(ProfileEvents::SharedPartsLocks);

}

DataPartsSharedLock::~DataPartsSharedLock()

{

if (lock_watch.has_value())

ProfileEvents::increment(ProfileEvents::SharedPartsLockHoldMicroseconds, lock_watch->elapsedMicroseconds());

}

static Int64 extractVersion(const auto & versions, const String & partition_id, Int64 default_value)

{

if (!versions)

return default_value;

auto it = versions->find(partition_id);

if (it == versions->end())

return default_value;

return it->second;

}

Int64 MergeTreeData::IMutationsSnapshot::getMinPartDataVersionForPartition(const Params & params, const String & partition_id)

{

return extractVersion(params.min_part_data_versions, partition_id, std::numeric_limits<Int64>::min());

}

Int64 MergeTreeData::IMutationsSnapshot::getMaxMutationVersionForPartition(const Params & params, const String & partition_id)

{

return extractVersion(params.max_mutation_versions, partition_id, std::numeric_limits<Int64>::max());

}

bool MergeTreeData::IMutationsSnapshot::needIncludeMutationToSnapshot(const Params & params, const MutationCommands & commands)

{

for (const auto & command : commands)

{

if (params.need_data_mutations && AlterConversions::isSupportedDataMutation(command.type))

return true;

if (params.need_alter_mutations && AlterConversions::isSupportedAlterMutation(command.type))

return true;

/// Metadata mutations must be included into the snapshot regardless of the parameters.

if (AlterConversions::isSupportedMetadataMutation(command.type))

return true;

}

return false;

}

MergeTreeData::MutationsSnapshotBase::MutationsSnapshotBase(Params params_, MutationCounters counters_, DataPartsVector patches_)

: params(std::move(params_))

, counters(std::move(counters_))

, patches_by_partition(getPatchPartsByPartition(patches_, params.max_mutation_versions))

{

}

void MergeTreeData::MutationsSnapshotBase::addPatches(DataPartsVector patches_)

{

if (patches_.empty())

return;

patches_by_partition = getPatchPartsByPartition(patches_, params.max_mutation_versions);

params.need_patch_parts = true;

}

NameSet MergeTreeData::MutationsSnapshotBase::getColumnsUpdatedInPatches() const

{

if (!params.need_patch_parts)

return {};

NameSet res;

for (const auto & [_, patches] : patches_by_partition)

{

for (const auto & patch : patches)

{

const auto & columns = patch->getColumns();

auto metadata_snapshot = patch->storage.getInMemoryMetadataPtr();

for (const auto & column : columns)

{

if (!isPatchPartSystemColumn(column.name))

res.insert(column.name);

}

}

}

return res;

}

void MergeTreeData::MutationsSnapshotBase::addSupportedCommands(const MutationCommands & commands, UInt64 mutation_version, MutationCommands & result_commands) const

{

for (const auto & command : commands | std::views::reverse)

{

bool is_supported = AlterConversions::isSupportedMetadataMutation(command.type)

|| (params.need_data_mutations && AlterConversions::isSupportedDataMutation(command.type))

|| (params.need_alter_mutations && AlterConversions::isSupportedAlterMutation(command.type));

if (is_supported)

{

auto & result_command = result_commands.emplace_back(command);

result_command.mutation_version = mutation_version;

}

}

}

PatchParts MergeTreeData::MutationsSnapshotBase::getPatchesForPart(const DataPartPtr & part) const

{

if (!params.need_patch_parts)

return {};

auto in_partition = patches_by_partition.find(part->info.getPartitionId());

if (in_partition == patches_by_partition.end())

return {};

PatchParts res;

for (const auto & patch_part : in_partition->second)

{

auto patch_infos = DB::getPatchesForPart(part->info, patch_part);

std::move(patch_infos.begin(), patch_infos.end(), std::back_inserter(res));

}

return res;

}

MergeTreeData::MergeTreeData(

const StorageID & table_id_,

const StorageInMemoryMetadata & metadata_,

ContextMutablePtr context_,

const String & date_column_name,

const MergingParams & merging_params_,

std::unique_ptr<MergeTreeSettings> storage_settings_,

bool require_part_metadata_,

LoadingStrictnessLevel mode,

BrokenPartCallback broken_part_callback_)

: WithMutableContext(context_->getGlobalContext())

, IStorage(table_id_)

, format_version(date_column_name.empty() ? MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING : MERGE_TREE_DATA_OLD_FORMAT_VERSION)

, merging_params(merging_params_)

, require_part_metadata(require_part_metadata_)

, columns_descriptions_metric_handle(CurrentMetrics::ColumnsDescriptionsCacheSize)

, broken_part_callback(broken_part_callback_)

, log(table_id_.getNameForLogs())

, storage_settings(std::move(storage_settings_))

, pinned_part_uuids(std::make_shared<PinnedPartUUIDs>())

, data_parts_by_info(data_parts_indexes.get<TagByInfo>())

, data_parts_by_state_and_info(data_parts_indexes.get<TagByStateAndInfo>())

, parts_mover(this)

, background_operations_assignee(*this, table_id_, BackgroundJobsAssignee::Type::DataProcessing, getContext())

, background_moves_assignee(*this, table_id_, BackgroundJobsAssignee::Type::Moving, getContext())

{

context_->getGlobalContext()->initializeBackgroundExecutorsIfNeeded();

const auto settings = getSettings();

bool sanity_checks = mode <= LoadingStrictnessLevel::CREATE;

allow_nullable_key = !sanity_checks || (*settings)[MergeTreeSetting::allow_nullable_key];

allow_reverse_key = !sanity_checks || (*settings)[MergeTreeSetting::allow_experimental_reverse_key];

/// Check sanity of MergeTreeSettings. Only when table is created.

if (sanity_checks)

{

const auto & ac = getContext()->getAccessControl();

bool allow_experimental = ac.getAllowExperimentalTierSettings();

bool allow_beta = ac.getAllowBetaTierSettings();

settings->sanityCheck(getContext()->getMergeMutateExecutor()->getMaxTasksCount(), allow_experimental, allow_beta);

}

if (!date_column_name.empty())

{

try

{

checkPartitionKeyAndInitMinMax(metadata_.partition_key);

setProperties(metadata_, metadata_, !sanity_checks);

if (minmax_idx_date_column_pos == -1)

throw Exception(ErrorCodes::BAD_TYPE_OF_FIELD, "Could not find Date column");

}

catch (Exception & e)

{

/// Better error message.

e.addMessage("(while initializing MergeTree partition key from date column " + backQuote(date_column_name) + ")");

throw;

}

}

else

{

is_custom_partitioned = true;

checkPartitionKeyAndInitMinMax(metadata_.partition_key);

}

setProperties(metadata_, metadata_, !sanity_checks);

/// NOTE: using the same columns list as is read when performing actual merges.

merging_params.check(*settings, metadata_);

if (metadata_.sampling_key.definition_ast != nullptr)

{

/// This is for backward compatibility.

checkSampleExpression(metadata_, !sanity_checks || (*settings)[MergeTreeSetting::compatibility_allow_sampling_expression_not_in_primary_key],

(*settings)[MergeTreeSetting::check_sample_column_is_correct] && sanity_checks);

}

checkColumnFilenamesForCollision(metadata_.getColumns(), *settings, sanity_checks);

checkTTLExpressions(metadata_, metadata_);

String reason;

if (!canUsePolymorphicParts(*settings, reason) && !reason.empty())

LOG_WARNING(log, "{} Settings 'min_rows_for_wide_part'and 'min_bytes_for_wide_part' will be ignored.", reason);

common_assignee_trigger = [this] (bool delay) noexcept

{

if (delay)

background_operations_assignee.postpone();

else

background_operations_assignee.trigger();

};

moves_assignee_trigger = [this] (bool delay) noexcept

{

if (delay)

background_moves_assignee.postpone();

else

background_moves_assignee.trigger();

};

}

VirtualColumnsDescription MergeTreeData::createVirtuals(const StorageInMemoryMetadata & metadata)

{

VirtualColumnsDescription desc;

desc.addEphemeral("_part", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Name of part");

desc.addEphemeral("_part_index", std::make_shared<DataTypeUInt64>(), "Sequential index of the part in the query result");

desc.addEphemeral("_part_starting_offset", std::make_shared<DataTypeUInt64>(), "Cumulative starting row of the part in the query result");

desc.addEphemeral("_part_uuid", std::make_shared<DataTypeUUID>(), "Unique part identifier (if enabled MergeTree setting assign_part_uuids)");

desc.addEphemeral("_partition_id", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Name of partition");

desc.addEphemeral("_sample_factor", std::make_shared<DataTypeFloat64>(), "Sample factor (from the query)");

desc.addEphemeral("_part_offset", std::make_shared<DataTypeUInt64>(), "Number of row in the part");

desc.addEphemeral("_part_granule_offset", std::make_shared<DataTypeUInt64>(), "Number of granule in the part");

desc.addEphemeral(PartDataVersionColumn::name, std::make_shared<DataTypeUInt64>(), "Data version of part (either min block number or mutation version)");

desc.addEphemeral("_disk_name", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Disk name");

desc.addEphemeral("_distance", std::make_shared<DataTypeFloat32>(), "Pre-computed distance for vector search queries");

if (metadata.hasPartitionKey())

{

auto partition_types = metadata.partition_key.sample_block.getDataTypes();

desc.addEphemeral("_partition_value", std::make_shared<DataTypeTuple>(std::move(partition_types)), "Value (a tuple) of a PARTITION BY expression");

}

desc.addPersistent(RowExistsColumn::name, RowExistsColumn::type, nullptr, "Persisted mask created by lightweight delete that show whether row exists or is deleted");

desc.addPersistent(BlockNumberColumn::name, BlockNumberColumn::type, BlockNumberColumn::codec, "Persisted original number of block that was assigned at insert");

desc.addPersistent(BlockOffsetColumn::name, BlockOffsetColumn::type, BlockOffsetColumn::codec, "Persisted original number of row in block that was assigned at insert");

return desc;

}

VirtualColumnsDescription MergeTreeData::createProjectionVirtuals(const StorageInMemoryMetadata & metadata)

{

VirtualColumnsDescription desc;

desc.addEphemeral("_part", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Name of part");

desc.addEphemeral("_part_index", std::make_shared<DataTypeUInt64>(), "Sequential index of the part in the query result");

desc.addEphemeral("_part_starting_offset", std::make_shared<DataTypeUInt64>(), "Cumulative starting row of the part in the query result");

desc.addEphemeral("_part_uuid", std::make_shared<DataTypeUUID>(), "Unique part identifier (if enabled MergeTree setting assign_part_uuids)");

desc.addEphemeral("_partition_id", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Name of partition");

desc.addEphemeral("_part_data_version", std::make_shared<DataTypeUInt64>(), "Data version of part (either min block number or mutation version)");

desc.addEphemeral("_disk_name", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>()), "Disk name");

if (metadata.hasPartitionKey())

{

auto partition_types = metadata.partition_key.sample_block.getDataTypes();

desc.addEphemeral("_partition_value", std::make_shared<DataTypeTuple>(std::move(partition_types)), "Value (a tuple) of a PARTITION BY expression");

}

return desc;

}

StoragePolicyPtr MergeTreeData::getStoragePolicy() const

{

auto settings = getSettings();

const auto & context = getContext();

StoragePolicyPtr storage_policy;

if ((*settings)[MergeTreeSetting::disk].changed)

storage_policy = context->getStoragePolicyFromDisk((*settings)[MergeTreeSetting::disk]);

else

storage_policy = context->getStoragePolicy((*settings)[MergeTreeSetting::storage_policy]);

return storage_policy;

}

std::map<std::string, DiskPtr> MergeTreeData::getDistinctDisksForParts(const DataPartsVector & parts_list) const

{

if (parts_list.empty())

return {};

std::map<std::string, DiskPtr> results;

auto storage_policy = getStoragePolicy();

for (const auto & part : parts_list)

{

auto disk_name = part->getDataPartStorage().getDiskName();

DiskPtr disk = storage_policy->tryGetDiskByName(disk_name);

if (disk == nullptr)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Couldn't find disk '{}' for part {}",

disk_name, part->name);

results.try_emplace(disk_name, disk);

}

return results;

}

ConditionSelectivityEstimatorPtr MergeTreeData::getConditionSelectivityEstimator(

const RangesInDataParts & parts, const Names & required_columns, ContextPtr local_context) const

{

if (!local_context->getSettingsRef()[Setting::use_statistics])

return nullptr;

if (parts.empty())

return {};

{

std::lock_guard<std::mutex> lock(stats_mutex);

if (local_context->getSettingsRef()[Setting::use_statistics_cache]

&& cached_estimator)

return cached_estimator;

}

LOG_DEBUG(log, "Loading statistics");

ConditionSelectivityEstimatorBuilder estimator_builder(local_context);

ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::LoadedStatisticsMicroseconds);

for (const auto & part : parts)

{

try

{

auto parts_lock = readLockParts();

auto stats = part.data_part->loadStatistics(required_columns);

estimator_builder.markDataPart(part.data_part);

for (const auto & [column_name, stat] : stats)

estimator_builder.addStatistics(column_name, stat);

}

catch (...)

{

tryLogCurrentException(log, fmt::format("while loading statistics on part {}", part.data_part->info.getPartNameV1()));

}

}

return estimator_builder.getEstimator();

}

bool MergeTreeData::supportsFinal() const

{

return merging_params.mode == MergingParams::Collapsing

|| merging_params.mode == MergingParams::Summing

|| merging_params.mode == MergingParams::Aggregating

|| merging_params.mode == MergingParams::Replacing

|| merging_params.mode == MergingParams::Coalescing

|| merging_params.mode == MergingParams::Graphite

|| merging_params.mode == MergingParams::VersionedCollapsing;

}

static void checkKeyExpression(const ExpressionActions & expr, const Block & sample_block, const String & key_name, bool allow_nullable_key)

{

if (expr.hasArrayJoin())

throw Exception(ErrorCodes::ILLEGAL_COLUMN, "{} key cannot contain array joins", key_name);

try

{

expr.assertDeterministic();

}

catch (Exception & e)

{

e.addMessage(fmt::format("for {} key", key_name));

throw;

}

for (const ColumnWithTypeAndName & element : sample_block)

{

const ColumnPtr & column = element.column;

if (column && (isColumnConst(*column) || column->isDummy()))

throw Exception(ErrorCodes::ILLEGAL_COLUMN, "{} key cannot contain constants", key_name);

if (!allow_nullable_key && hasNullable(element.type))

throw Exception(

ErrorCodes::ILLEGAL_COLUMN,

"{} key contains nullable columns, "

"but merge tree setting `allow_nullable_key` is disabled", key_name);

}

}

void MergeTreeData::checkProperties(

const StorageInMemoryMetadata & new_metadata,

const StorageInMemoryMetadata & old_metadata,

bool attach,

bool allow_empty_sorting_key,

bool allow_reverse_sorting_key,

bool allow_nullable_key_,

ContextPtr local_context) const

{

if (!new_metadata.sorting_key.definition_ast && !allow_empty_sorting_key)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "ORDER BY cannot be empty");

if (!allow_reverse_sorting_key)

{

size_t num_sorting_keys = new_metadata.sorting_key.column_names.size();

for (size_t i = 0; i < num_sorting_keys; ++i)

{

if (!new_metadata.sorting_key.reverse_flags.empty() && new_metadata.sorting_key.reverse_flags[i])

{

throw Exception(

ErrorCodes::ILLEGAL_COLUMN,

"Sorting key {} is reversed, but merge tree setting `allow_experimental_reverse_key` is disabled",

new_metadata.sorting_key.column_names[i]);

}

}

}

KeyDescription new_sorting_key = new_metadata.sorting_key;

KeyDescription new_primary_key = new_metadata.primary_key;

size_t sorting_key_size = new_sorting_key.column_names.size();

size_t primary_key_size = new_primary_key.column_names.size();

if (primary_key_size > sorting_key_size)

throw Exception(ErrorCodes::BAD_ARGUMENTS, "Primary key must be a prefix of the sorting key, but its length: "

"{} is greater than the sorting key length: {}", primary_key_size, sorting_key_size);

bool allow_suspicious_indices = (*getSettings())[MergeTreeSetting::allow_suspicious_indices];

if (local_context)

allow_suspicious_indices = local_context->getSettingsRef()[Setting::allow_suspicious_indices];

if (!allow_suspicious_indices && !attach)

if (const auto * index_function = typeid_cast<ASTFunction *>(new_sorting_key.definition_ast.get()))

checkSuspiciousIndices(index_function);

for (size_t i = 0; i < sorting_key_size; ++i)

{

const String & sorting_key_column = new_sorting_key.column_names[i];

if (i < primary_key_size)

{

const String & pk_column = new_primary_key.column_names[i];

if (pk_column != sorting_key_column)

throw Exception(ErrorCodes::BAD_ARGUMENTS,

"Primary key must be a prefix of the sorting key, "

"but the column in the position {} is {}", i, sorting_key_column +", not " + pk_column);

}

}

auto all_columns = new_metadata.columns.getAllPhysical();

/// This is ALTER, not CREATE/ATTACH TABLE. Let us check that all new columns used in the sorting key

/// expression have just been added (so that the sorting order is guaranteed to be valid with the new key).

Names new_primary_key_columns = new_primary_key.column_names;

Names new_sorting_key_columns = new_sorting_key.column_names;

ASTPtr added_key_column_expr_list = make_intrusive<ASTExpressionList>();

const auto & old_sorting_key_columns = old_metadata.getSortingKeyColumns();

for (size_t new_i = 0, old_i = 0; new_i < sorting_key_size; ++new_i)

{

if (old_i < old_sorting_key_columns.size())

{

if (new_sorting_key_columns[new_i] != old_sorting_key_columns[old_i])

added_key_column_expr_list->children.push_back(new_sorting_key.expression_list_ast->children[new_i]);

else

++old_i;

}

else

added_key_column_expr_list->children.push_back(new_sorting_key.expression_list_ast->children[new_i]);

}

if (!added_key_column_expr_list->children.empty())