#include <Storages/ColumnSize.h>

#include <Storages/MergeTree/IMergeTreeDataPart.h>

#include <Storages/MergeTree/IDataPartStorage.h>

#include <Storages/MergeTree/DataPartStorageOnDiskBase.h>

#include <Columns/ColumnNullable.h>

#include <Common/DateLUTImpl.h>

#include <Common/SipHash.h>

#include <Common/UniqueLock.h>

#include <Common/ZooKeeper/ZooKeeperCommon.h>

#include <Common/quoteString.h>

#include <Compression/CompressedReadBuffer.h>

#include <Compression/CompressionFactory.h>

#include <Compression/getCompressionCodecForFile.h>

#include <Core/Defines.h>

#include <Core/NamesAndTypes.h>

#include <Core/Settings.h>

#include <DataTypes/DataTypeAggregateFunction.h>

#include <DataTypes/NestedUtils.h>

#include <IO/HashingWriteBuffer.h>

#include <IO/ReadBufferFromString.h>

#include <IO/ReadHelpers.h>

#include <IO/WriteHelpers.h>

#include <IO/PackedFilesReader.h>

#include <Interpreters/MergeTreeTransaction.h>

#include <Interpreters/TransactionLog.h>

#include <Interpreters/Context.h>

#include <Parsers/ExpressionElementParsers.h>

#include <Parsers/parseQuery.h>

#include <Storages/ColumnsDescription.h>

#include <Storages/MergeTree/MergeTreeData.h>

#include <Storages/MergeTree/MergeTreeSettings.h>

#include <Storages/MergeTree/checkDataPart.h>

#include <Storages/MergeTree/Backup.h>

#include <Storages/StorageReplicatedMergeTree.h>

#include <Storages/MergeTree/MergeTreeIndexGranularityAdaptive.h>

#include <Storages/MergeTree/MergeTreeIndexGranularityConstant.h>

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

#include <Storages/MergeTree/PrimaryIndexCache.h>

#include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>

#include <Storages/MergeTree/MergeTreeMarksLoader.h>

#include <Storages/MergeTree/MergeTreeIndices.h>

#include <base/JSON.h>

#include <Common/CurrentMetrics.h>

#include <Common/Exception.h>

#include <Common/ErrnoException.h>

#include <Common/FieldAccurateComparison.h>

#include <Common/FailPoint.h>

#include <Common/MemoryTrackerBlockerInThread.h>

#include <Common/StringUtils.h>

#include <Common/thread_local_rng.h>

#include <Common/escapeForFileName.h>

#include <Common/logger_useful.h>

#include <Common/DimensionalMetrics.h>

#include <Disks/IO/CachedOnDiskReadBufferFromFile.h>

#include <atomic>

#include <exception>

#include <mutex>

#include <optional>

#include <string_view>

namespace CurrentMetrics

{

extern const Metric PartsTemporary;

extern const Metric PartsPreCommitted;

extern const Metric PartsCommitted;

extern const Metric PartsPreActive;

extern const Metric PartsActive;

extern const Metric PartsOutdated;

extern const Metric PartsDeleting;

extern const Metric PartsDeleteOnDestroy;

extern const Metric PartsWide;

extern const Metric PartsCompact;

}

namespace ProfileEvents

{

extern const Event LoadedPrimaryIndexFiles;

extern const Event LoadedPrimaryIndexRows;

extern const Event LoadedPrimaryIndexBytes;

}

namespace DimensionalMetrics

{

extern MetricFamily & MergeTreeParts;

}

namespace DB

{

namespace MergeTreeSetting

{

extern const MergeTreeSettingsBool allow_remote_fs_zero_copy_replication;

extern const MergeTreeSettingsBool exclude_deleted_rows_for_part_size_in_merge;

extern const MergeTreeSettingsBool fsync_part_directory;

extern const MergeTreeSettingsBool load_existing_rows_count_for_old_parts;

extern const MergeTreeSettingsBool primary_key_lazy_load;

extern const MergeTreeSettingsFloat primary_key_ratio_of_unique_prefix_values_to_skip_suffix_columns;

extern const MergeTreeSettingsFloat ratio_of_defaults_for_sparse_serialization;

extern const MergeTreeSettingsBool columns_and_secondary_indices_sizes_lazy_calculation;

extern const MergeTreeSettingsMergeTreeSerializationInfoVersion serialization_info_version;

}

namespace Setting

{

extern const SettingsBool merge_tree_use_prefixes_deserialization_thread_pool;

}

namespace ErrorCodes

{

extern const int CANNOT_READ_ALL_DATA;

extern const int LOGICAL_ERROR;

extern const int NO_FILE_IN_DATA_PART;

extern const int EXPECTED_END_OF_FILE;

extern const int CORRUPTED_DATA;

extern const int NOT_FOUND_EXPECTED_DATA_PART;

extern const int BAD_SIZE_OF_FILE_IN_DATA_PART;

extern const int BAD_TTL_FILE;

extern const int NOT_IMPLEMENTED;

extern const int NO_SUCH_COLUMN_IN_TABLE;

}

namespace FailPoints

{

extern const char remove_merge_tree_part_delay[];

}

namespace

{

String getIndexExtensionFromFilesystem(const IDataPartStorage & data_part_storage)

{

if (data_part_storage.existsFile("primary" + getIndexExtension(true)))

return getIndexExtension(true);

else

return {".idx"};

}

}

void IMergeTreeDataPart::MinMaxIndex::load(const IMergeTreeDataPart & part)

{

auto metadata_snapshot = part.getMetadataSnapshot();

const auto & partition_key = metadata_snapshot->getPartitionKey();

auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(partition_key);

auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(partition_key);

size_t minmax_idx_size = minmax_column_types.size();

hyperrectangle.reserve(minmax_idx_size);

FormatSettings format_settings;

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

{

String file_name = "minmax_" + getFileColumnName(minmax_column_names[i], part.checksums) + ".idx";

auto file = part.readFile(file_name);

auto serialization = minmax_column_types[i]->getDefaultSerialization();

Field min_val;

serialization->deserializeBinary(min_val, *file, format_settings);

Field max_val;

serialization->deserializeBinary(max_val, *file, format_settings);

// NULL_LAST

if (min_val.isNull())

min_val = POSITIVE_INFINITY;

if (max_val.isNull())

max_val = POSITIVE_INFINITY;

hyperrectangle.emplace_back(min_val, true, max_val, true);

}

initialized = true;

}

IMergeTreeDataPart::MinMaxIndex::WrittenFiles IMergeTreeDataPart::MinMaxIndex::store(

StorageMetadataPtr metadata_snapshot, IDataPartStorage & part_storage, Checksums & out_checksums, const MergeTreeSettingsPtr & storage_settings) const

{

const auto & partition_key = metadata_snapshot->getPartitionKey();

auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(partition_key);

auto minmax_column_types = MergeTreeData::getMinMaxColumnsTypes(partition_key);

return store(minmax_column_names, minmax_column_types, part_storage, out_checksums, storage_settings);

}

IMergeTreeDataPart::MinMaxIndex::WrittenFiles IMergeTreeDataPart::MinMaxIndex::store(

const Names & column_names,

const DataTypes & data_types,

IDataPartStorage & part_storage,

Checksums & out_checksums,

const MergeTreeSettingsPtr & storage_settings) const

{

if (!initialized)

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Attempt to store uninitialized MinMax index for part {}. This is a bug",

part_storage.getFullPath());

WrittenFiles written_files;

for (size_t i = 0; i < column_names.size(); ++i)

{

String file_name = "minmax_" + getFileColumnName(column_names[i], storage_settings, part_storage) + ".idx";

auto serialization = data_types.at(i)->getDefaultSerialization();

auto out = part_storage.writeFile(file_name, 4096, {});

HashingWriteBuffer out_hashing(*out);

serialization->serializeBinary(hyperrectangle[i].left, out_hashing, {});

serialization->serializeBinary(hyperrectangle[i].right, out_hashing, {});

out_hashing.finalize();

out_checksums.files[file_name].file_size = out_hashing.count();

out_checksums.files[file_name].file_hash = out_hashing.getHash();

out->preFinalize();

written_files.emplace_back(std::move(out));

}

return written_files;

}

void IMergeTreeDataPart::MinMaxIndex::update(const Block & block, const Names & column_names)

{

if (block.rows() == 0)

return;

if (!initialized)

hyperrectangle.reserve(column_names.size());

for (size_t i = 0; i < column_names.size(); ++i)

{

FieldRef min_value;

FieldRef max_value;

const ColumnWithTypeAndName & column = block.getColumnOrSubcolumnByName(column_names[i]);

if (const auto * column_nullable = typeid_cast<const ColumnNullable *>(column.column.get()))

column_nullable->getExtremesNullLast(min_value, max_value, 0, column.column->size());

else

column.column->getExtremes(min_value, max_value, 0, column.column->size());

if (!initialized)

hyperrectangle.emplace_back(min_value, true, max_value, true);

else

{

hyperrectangle[i].left = accurateLess(hyperrectangle[i].left, min_value) ? hyperrectangle[i].left : min_value;

hyperrectangle[i].right = accurateLess(hyperrectangle[i].right, max_value) ? max_value : hyperrectangle[i].right;

}

}

initialized = true;

}

void IMergeTreeDataPart::MinMaxIndex::merge(const MinMaxIndex & other)

{

if (!other.initialized)

return;

if (!initialized)

{

hyperrectangle = other.hyperrectangle;

initialized = true;

}

else

{

for (size_t i = 0; i < hyperrectangle.size(); ++i)

{

hyperrectangle[i].left = accurateLess(hyperrectangle[i].left, other.hyperrectangle[i].left)

? hyperrectangle[i].left

: other.hyperrectangle[i].left;

hyperrectangle[i].right = accurateLess(hyperrectangle[i].right, other.hyperrectangle[i].right)

? other.hyperrectangle[i].right

: hyperrectangle[i].right;

}

}

}

void IMergeTreeDataPart::MinMaxIndex::appendFiles(const MergeTreeData & data, Strings & files, const IDataPartStorage & data_part_storage)

{

auto metadata_snapshot = data.getInMemoryMetadataPtr();

const auto & partition_key = metadata_snapshot->getPartitionKey();

auto minmax_column_names = MergeTreeData::getMinMaxColumnsNames(partition_key);

size_t minmax_idx_size = minmax_column_names.size();

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

{

String file_name = "minmax_" + getFileColumnName(minmax_column_names[i], data.getSettings(), data_part_storage) + ".idx";

files.push_back(file_name);

}

}

String IMergeTreeDataPart::MinMaxIndex::getFileColumnName(const String & column_name, const MergeTreeSettingsPtr & storage_settings_, const IDataPartStorage & data_part_storage)

{

return replaceFileNameToHashIfNeeded(escapeForFileName(column_name), *storage_settings_, &data_part_storage);

}

String IMergeTreeDataPart::MinMaxIndex::getFileColumnName(const String & column_name, const Checksums & checksums_)

{

String stream_name = escapeForFileName(column_name);

if (checksums_.files.contains("minmax_" + stream_name + ".idx"))

return stream_name;

auto hash = sipHash128String(stream_name);

if (checksums_.files.contains("minmax_" + hash + ".idx"))

return hash;

return stream_name;

}

void IMergeTreeDataPart::incrementStateMetric(MergeTreeDataPartState state_) const

{

switch (state_)

{

case MergeTreeDataPartState::Temporary:

CurrentMetrics::add(CurrentMetrics::PartsTemporary);

return;

case MergeTreeDataPartState::PreActive:

CurrentMetrics::add(CurrentMetrics::PartsPreActive);

CurrentMetrics::add(CurrentMetrics::PartsPreCommitted);

return;

case MergeTreeDataPartState::Active:

CurrentMetrics::add(CurrentMetrics::PartsActive);

CurrentMetrics::add(CurrentMetrics::PartsCommitted);

return;

case MergeTreeDataPartState::Outdated:

storage.total_outdated_parts_count.fetch_add(1, std::memory_order_relaxed);

CurrentMetrics::add(CurrentMetrics::PartsOutdated);

return;

case MergeTreeDataPartState::Deleting:

CurrentMetrics::add(CurrentMetrics::PartsDeleting);

return;

case MergeTreeDataPartState::DeleteOnDestroy:

CurrentMetrics::add(CurrentMetrics::PartsDeleteOnDestroy);

return;

}

}

void IMergeTreeDataPart::decrementStateMetric(MergeTreeDataPartState state_) const

{

switch (state_)

{

case MergeTreeDataPartState::Temporary:

CurrentMetrics::sub(CurrentMetrics::PartsTemporary);

return;

case MergeTreeDataPartState::PreActive:

CurrentMetrics::sub(CurrentMetrics::PartsPreActive);

CurrentMetrics::sub(CurrentMetrics::PartsPreCommitted);

return;

case MergeTreeDataPartState::Active:

CurrentMetrics::sub(CurrentMetrics::PartsActive);

CurrentMetrics::sub(CurrentMetrics::PartsCommitted);

return;

case MergeTreeDataPartState::Outdated:

storage.total_outdated_parts_count.fetch_sub(1, std::memory_order_relaxed);

CurrentMetrics::sub(CurrentMetrics::PartsOutdated);

return;

case MergeTreeDataPartState::Deleting:

CurrentMetrics::sub(CurrentMetrics::PartsDeleting);

return;

case MergeTreeDataPartState::DeleteOnDestroy:

CurrentMetrics::sub(CurrentMetrics::PartsDeleteOnDestroy);

return;

}

}

static void incrementTypeMetric(MergeTreeDataPartType type)

{

switch (type.getValue())

{

case MergeTreeDataPartType::Wide:

CurrentMetrics::add(CurrentMetrics::PartsWide);

return;

case MergeTreeDataPartType::Compact:

CurrentMetrics::add(CurrentMetrics::PartsCompact);

return;

case MergeTreeDataPartType::Unknown:

return;

}

}

static void decrementTypeMetric(MergeTreeDataPartType type)

{

switch (type.getValue())

{

case MergeTreeDataPartType::Wide:

CurrentMetrics::sub(CurrentMetrics::PartsWide);

return;

case MergeTreeDataPartType::Compact:

CurrentMetrics::sub(CurrentMetrics::PartsCompact);

return;

case MergeTreeDataPartType::Unknown:

return;

}

}

IMergeTreeDataPart::IMergeTreeDataPart(

const MergeTreeData & storage_,

const MergeTreeSettings & storage_settings,

const String & name_,

const MergeTreePartInfo & info_,

const MutableDataPartStoragePtr & data_part_storage_,

Type part_type_,

const IMergeTreeDataPart * parent_part_)

: DataPartStorageHolder(data_part_storage_)

, storage(storage_)

, name(mutable_name)

, info(info_)

, index_granularity_info(storage_, storage_settings, part_type_)

, columns_sizes(std::make_shared<ColumnSizeByName>())

, secondary_index_sizes(std::make_shared<IndexSizeByName>())

, part_type(part_type_)

, parent_part(parent_part_)

, parent_part_name(parent_part ? parent_part->name : "")

, mutable_name(name_)

{

auto component_guard = Coordination::setCurrentComponent("IMergeTreeDataPart::IMergeTreeDataPart");

if (parent_part)

{

chassert(parent_part_name.starts_with(parent_part->info.getPartitionId())); /// Make sure there's no prefix

state = MergeTreeDataPartState::Active;

}

incrementStateMetric(state);

incrementTypeMetric(part_type);

DimensionalMetrics::add(

DimensionalMetrics::MergeTreeParts,

{stateToString(), part_type.toString(), std::to_string(isProjectionPart())});

minmax_idx = std::make_shared<MinMaxIndex>();

initializeIndexGranularityInfo(storage_settings);

}

IMergeTreeDataPart::~IMergeTreeDataPart()

{

decrementStateMetric(state);

decrementTypeMetric(part_type);

if (columns_description)

{

columns_description.reset();

columns_description_with_collected_nested.reset();

storage.decrefColumnsDescriptionForColumns(columns);

}

DimensionalMetrics::sub(

DimensionalMetrics::MergeTreeParts,

{stateToString(), part_type.toString(), std::to_string(isProjectionPart())});

}

IMergeTreeDataPart::IndexPtr IMergeTreeDataPart::getIndex() const

{

std::scoped_lock lock(index_mutex);

if (index)

return index;

if (auto index_cache = storage.getPrimaryIndexCache())

return loadIndexToCache(*index_cache);

index = loadIndex();

return index;

}

IMergeTreeDataPart::IndexPtr IMergeTreeDataPart::loadIndexToCache(PrimaryIndexCache & index_cache) const

{

auto key = PrimaryIndexCache::hash(getDataPartStorage().getDiskName() + ":" + getRelativePathOfActivePart());

auto callback = [this] { return loadIndex(); };

return index_cache.getOrSet(key, callback);

}

void IMergeTreeDataPart::moveIndexToCache(PrimaryIndexCache & index_cache)

{

std::scoped_lock lock(index_mutex);

if (!index)

return;

auto key = PrimaryIndexCache::hash(getDataPartStorage().getDiskName() + ":" + getRelativePathOfActivePart());

index_cache.set(key, std::const_pointer_cast<Index>(index));

index.reset();

for (const auto & [_, projection] : projection_parts)

projection->moveIndexToCache(index_cache);

}

void IMergeTreeDataPart::removeIndexFromCache(PrimaryIndexCache * index_cache) const

{

if (!index_cache)

return;

auto key = PrimaryIndexCache::hash(getDataPartStorage().getDiskName() + ":" + getRelativePathOfActivePart());

index_cache->remove(key);

}

void IMergeTreeDataPart::removeFromVectorIndexCache(VectorSimilarityIndexCache * vector_similarity_index_cache) const

{

if (!vector_similarity_index_cache)

return;

vector_similarity_index_cache->removeEntriesFromCache(getDataPartStorage().getDiskName() + ":" + getRelativePathOfActivePart());

}

void IMergeTreeDataPart::setIndex(Columns index_columns)

{

std::scoped_lock lock(index_mutex);

if (index)

throw Exception(ErrorCodes::LOGICAL_ERROR, "The index of data part can be set only once");

optimizeIndexColumns(index_granularity->getMarksCount(), index_columns);

index = std::make_shared<Index>(std::move(index_columns));

}

void IMergeTreeDataPart::unloadIndex()

{

std::scoped_lock lock(index_mutex);

index.reset();

}

bool IMergeTreeDataPart::isIndexLoaded() const

{

std::scoped_lock lock(index_mutex);

return index != nullptr;

}

void IMergeTreeDataPart::setName(const String & new_name)

{

mutable_name = new_name;

for (auto & proj_part : projection_parts)

proj_part.second->parent_part_name = new_name;

}

String IMergeTreeDataPart::getNewName(const MergeTreePartInfo & new_part_info) const

{

if (storage.format_version < MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING)

{

/// NOTE: getting min and max dates from the part name (instead of part data) because we want

/// the merged part name be determined only by source part names.

/// It is simpler this way when the real min and max dates for the block range can change

/// (e.g. after an ALTER DELETE command).

DayNum min_date;

DayNum max_date;

MergeTreePartInfo::parseMinMaxDatesFromPartName(name, min_date, max_date);

return new_part_info.getPartNameV0(min_date, max_date);

}

return new_part_info.getPartNameV1();

}

std::optional<size_t> IMergeTreeDataPart::getColumnPosition(const String & column_name) const

{

auto it = column_name_to_position.find(column_name);

if (it == column_name_to_position.end())

return {};

return it->second;

}

void IMergeTreeDataPart::setState(MergeTreeDataPartState new_state) const

{

decrementStateMetric(state);

DimensionalMetrics::sub(

DimensionalMetrics::MergeTreeParts,

{stateToString(), part_type.toString(), std::to_string(isProjectionPart())});

state.store(new_state);

incrementStateMetric(state);

DimensionalMetrics::add(

DimensionalMetrics::MergeTreeParts,

{stateToString(), part_type.toString(), std::to_string(isProjectionPart())});

}

std::string_view IMergeTreeDataPart::stateString(MergeTreeDataPartState state)

{

return magic_enum::enum_name(state);

}

std::pair<DayNum, DayNum> IMergeTreeDataPart::getMinMaxDate() const

{

if (storage.minmax_idx_date_column_pos != -1 && minmax_idx->initialized && !info.isPatch())

{

const auto & hyperrectangle = minmax_idx->hyperrectangle[storage.minmax_idx_date_column_pos];

return {

DayNum(static_cast<DayNum::UnderlyingType>(hyperrectangle.left.safeGet<UInt64>())),

DayNum(static_cast<DayNum::UnderlyingType>(hyperrectangle.right.safeGet<UInt64>()))};

}

return {};

}

std::pair<time_t, time_t> IMergeTreeDataPart::getMinMaxTime() const

{

if (storage.minmax_idx_time_column_pos != -1 && minmax_idx->initialized && !info.isPatch())

{

const auto & hyperrectangle = minmax_idx->hyperrectangle[storage.minmax_idx_time_column_pos];

/// The case of DateTime

if (hyperrectangle.left.getType() == Field::Types::UInt64)

{

assert(hyperrectangle.right.getType() == Field::Types::UInt64);

return {hyperrectangle.left.safeGet<UInt64>(), hyperrectangle.right.safeGet<UInt64>()};

}

/// The case of DateTime64

if (hyperrectangle.left.getType() == Field::Types::Decimal64)

{

assert(hyperrectangle.right.getType() == Field::Types::Decimal64);

auto left = hyperrectangle.left.safeGet<DecimalField<Decimal64>>();

auto right = hyperrectangle.right.safeGet<DecimalField<Decimal64>>();

assert(left.getScale() == right.getScale());

return {left.getValue() / left.getScaleMultiplier(), right.getValue() / right.getScaleMultiplier()};

}

throw Exception(ErrorCodes::LOGICAL_ERROR, "Part minmax index by time is neither DateTime or DateTime64");

}

return {};

}

void IMergeTreeDataPart::setColumns(const NamesAndTypesList & new_columns, const SerializationInfoByName & new_infos, int32_t new_metadata_version)

{

columns = new_columns;

serialization_infos = new_infos;

metadata_version = new_metadata_version;

serializations.clear();

column_name_to_position.clear();

column_name_to_position.reserve(new_columns.size());

size_t pos = 0;

for (const auto & column : columns)

column_name_to_position.emplace(column.name, pos++);

for (const auto & column : columns)

{

auto it = serialization_infos.find(column.name);

auto serialization = it == serialization_infos.end()

? IDataType::getSerialization(column, serialization_infos.getSettings())

: IDataType::getSerialization(column, *it->second);

serializations.emplace(column.name, serialization);

IDataType::forEachSubcolumn([&](const auto &, const auto & subname, const auto & subdata)

{

auto full_name = Nested::concatenateName(column.name, subname);

/// Don't override the column serialization with subcolumn serialization if column with the same name exists.

if (!column_name_to_position.contains(full_name))

serializations.emplace(full_name, subdata.serialization);

}, ISerialization::SubstreamData(serialization));

}

auto columns_descriptions = storage.getColumnsDescriptionForColumns(columns);

columns_description = columns_descriptions.original;

columns_description_with_collected_nested = columns_descriptions.with_collected_nested;

}

String IMergeTreeDataPart::getProjectionName() const

{

if (!parent_part)

return "";

if (!temp_projection_block_number)

return name;

if (!is_temp)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Temporary block number {} is set but projection part {} is not temporary", *temp_projection_block_number, name);

String suffix = "_" + std::to_string(*temp_projection_block_number);

if (!name.ends_with(suffix))

throw Exception(ErrorCodes::LOGICAL_ERROR, "Temporary block number {} is set but projection part {} doesn't have it in name's suffix", *temp_projection_block_number, name);

auto pos = name.rfind(suffix);

chassert(pos != std::string::npos);

return name.substr(0, pos);

}

StorageMetadataPtr IMergeTreeDataPart::getMetadataSnapshot() const

{

if (info.isPatch())

return storage.getPatchPartMetadata(*columns_description, info.getPartitionId(), storage.getContext());

auto metadata_snapshot = storage.getInMemoryMetadataPtr();

if (!parent_part)

return metadata_snapshot;

return metadata_snapshot->projections.get(getProjectionName()).metadata;

}

NameAndTypePair IMergeTreeDataPart::getColumn(const String & column_name) const

{

return columns_description->getColumnOrSubcolumn(GetColumnsOptions::AllPhysical, column_name);

}

std::optional<NameAndTypePair> IMergeTreeDataPart::tryGetColumn(const String & column_name) const

{

return columns_description->tryGetColumnOrSubcolumn(GetColumnsOptions::AllPhysical, column_name);

}

SerializationPtr IMergeTreeDataPart::getSerialization(const String & column_name) const

{

auto serialization = tryGetSerialization(column_name);

if (!serialization)

throw Exception(ErrorCodes::NO_SUCH_COLUMN_IN_TABLE,

"There is no column or subcolumn {} in part {}", column_name, name);

return serialization;

}

SerializationPtr IMergeTreeDataPart::tryGetSerialization(const String & column_name) const

{

auto it = serializations.find(column_name);

return it == serializations.end() ? nullptr : it->second;

}

bool IMergeTreeDataPart::isMovingPart() const

{

fs::path part_directory_path = getDataPartStorage().getRelativePath();

if (part_directory_path.filename().empty())

part_directory_path = part_directory_path.parent_path();

return part_directory_path.parent_path().filename() == "moving";

}

void IMergeTreeDataPart::loadIndexMarksToCache(MarkCache * index_mark_cache) const

{

if (!index_mark_cache)

return;

auto metadata_snapshot = storage.getInMemoryMetadataPtr();

const auto & secondary_indices = metadata_snapshot->getSecondaryIndices();

if (secondary_indices.empty())

return;

auto info_for_read = std::make_shared<LoadedMergeTreeDataPartInfoForReader>(shared_from_this(), std::make_shared<AlterConversions>());

auto read_settings = storage.getContext()->getReadSettings();

std::vector<std::unique_ptr<MergeTreeMarksLoader>> loaders;

for (const auto & index_description : secondary_indices)

{

auto skip_index = MergeTreeIndexFactory::instance().get(index_description);

auto index_name = skip_index->getFileName();

auto index_format = skip_index->getDeserializedFormat(checksums, index_name);

if (!index_format)

continue;

for (const auto & substream : index_format.substreams)

{

auto stream_name = getStreamNameOrHash(index_name + substream.suffix, substream.extension, checksums);

if (!stream_name)

continue;

loaders.emplace_back(std::make_unique<MergeTreeMarksLoader>(

info_for_read,

index_mark_cache,

index_granularity_info.getMarksFilePath(*stream_name),

index_granularity->getMarksCountForSkipIndex(index_description.granularity),

index_granularity_info,

/*save_marks_in_cache=*/ true,

read_settings,

/*load_marks_threadpool=*/ nullptr,

/*num_columns_in_mark=*/ 1));

loaders.back()->startAsyncLoad();

}

}

for (auto & loader : loaders)

loader->loadMarks();

}

void IMergeTreeDataPart::removeIndexMarksFromCache(MarkCache * index_mark_cache) const

{

if (!index_mark_cache)

return;

/// Bypass QueryMetadataCache: this runs during part destruction, and caching a dying

/// storage's pointer would poison lookups if a new storage is allocated at the same address.

auto metadata_snapshot = storage.getInMemoryMetadataPtr(/*bypass_metadata_cache=*/ true);

const auto & secondary_indices = metadata_snapshot->getSecondaryIndices();

if (secondary_indices.empty())

return;

for (const auto & index_description : secondary_indices)

{

auto skip_index = MergeTreeIndexFactory::instance().get(index_description);

auto index_name = skip_index->getFileName();

auto index_format = skip_index->getDeserializedFormat(checksums, index_name);

if (!index_format)

continue;

for (const auto & substream : index_format.substreams)

{

auto stream_name = getStreamNameOrHash(index_name + substream.suffix, substream.extension, checksums);

if (!stream_name)

continue;

auto marks_file = index_granularity_info.getMarksFilePath(*stream_name);

auto key = MarkCache::hash(getDataPartStorage().getDiskName() + ":" + (fs::path(getRelativePathOfActivePart()) / marks_file).string());

index_mark_cache->remove(key);

}

}

}

void IMergeTreeDataPart::clearCaches()

{

if (cleared_data_in_caches.exchange(true) || is_duplicate)

return;

/// Remove index and marks from the cache, because otherwise the cache will grow to its maximum size

/// even if the overall index size is much less.

removeMarksFromCache(storage.getContext()->getMarkCache().get());

removeIndexMarksFromCache(storage.getContext()->getIndexMarkCache().get());

removeIndexFromCache(storage.getPrimaryIndexCache().get());

/// Remove from other caches of secondary indexes

removeFromVectorIndexCache(storage.getContext()->getVectorSimilarityIndexCache().get());

}

bool IMergeTreeDataPart::mayStoreDataInCaches() const

{

if (cleared_data_in_caches)

return false;

auto caches = storage.getCachesToPrewarm(getBytesUncompressedOnDisk());

return caches.hasAny();

}

void IMergeTreeDataPart::removeIfNeeded()

{

if (is_removed)

return;

fiu_do_on(FailPoints::remove_merge_tree_part_delay,

{

std::chrono::milliseconds sleep_time{1300 + thread_local_rng() % 200};

std::this_thread::sleep_for(sleep_time);

});

chassert(assertHasValidVersionMetadata());

std::string path;

try

{

path = getDataPartStorage().getRelativePath();

clearCaches();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__, fmt::format("while removing part {} with path {}", name, path));

}

if (!is_temp && state != MergeTreeDataPartState::DeleteOnDestroy)

return;

try

{

if (!getDataPartStorage().exists()) // path

return;

if (is_temp)

{

const auto & part_directory = getDataPartStorage().getPartDirectory();

String file_name = fileName(part_directory);

if (file_name.empty())

throw Exception(ErrorCodes::LOGICAL_ERROR, "relative_path {} of part {} is invalid or not set",

getDataPartStorage().getPartDirectory(), name);

bool is_moving_part = isMovingPart();

if (!startsWith(file_name, "tmp") && !endsWith(file_name, ".tmp_proj") && !is_moving_part)

{

LOG_ERROR(

storage.log,

"~DataPart() should remove part {} but its name doesn't start with \"tmp\" or end with \".tmp_proj\". Too "

"suspicious, keeping the part.",

path);

return;

}

if (is_moving_part)

{

LOG_TRACE(storage.log, "Removing unneeded moved part from {}", path);

}

}

remove();

if (state == MergeTreeDataPartState::DeleteOnDestroy)

{

LOG_TRACE(storage.log, "Removed part from old location {}", path);

}

is_removed = true;

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__, fmt::format("while removing part {} with path {}", name, path));

/// FIXME If part it temporary, then directory will not be removed for 1 day (temporary_directories_lifetime).

/// If it's tmp_merge_<part_name> or tmp_fetch_<part_name>,

/// then all future attempts to execute part producing operation will fail with "directory already exists".

}

}

UInt64 IMergeTreeDataPart::getIndexSizeInBytes() const

{

std::scoped_lock lock(index_mutex);

if (!index)

return 0;

UInt64 res = 0;

for (const auto & column : *index)

res += column->byteSize();

return res;

}

UInt64 IMergeTreeDataPart::getIndexSizeInAllocatedBytes() const

{

std::scoped_lock lock(index_mutex);

if (!index)

return 0;

UInt64 res = 0;

for (const auto & column : *index)

res += column->allocatedBytes();

return res;

}

UInt64 IMergeTreeDataPart::getIndexGranularityBytes() const

{

return index_granularity->getBytesSize();

}

UInt64 IMergeTreeDataPart::getIndexGranularityAllocatedBytes() const

{

return index_granularity->getBytesAllocated();

}

void IMergeTreeDataPart::assertState(const std::initializer_list<MergeTreeDataPartState> & affordable_states) const

{

if (!checkState(affordable_states))

{

String states_str;

for (auto affordable_state : affordable_states)

{

states_str += stateString(affordable_state);

states_str += ' ';

}

if (!states_str.empty())

states_str.pop_back();

throw Exception(ErrorCodes::NOT_FOUND_EXPECTED_DATA_PART, "Unexpected state of part {}. Expected: {}", getNameWithState(), states_str);

}

}

UInt64 IMergeTreeDataPart::getMarksCount() const

{

return index_granularity->getMarksCount();

}

UInt64 IMergeTreeDataPart::getExistingBytesOnDisk() const

{

if ((*storage.getSettings())[MergeTreeSetting::exclude_deleted_rows_for_part_size_in_merge] && supportLightweightDeleteMutate() && hasLightweightDelete()

&& existing_rows_count.has_value() && existing_rows_count.value() < rows_count && rows_count > 0)

return bytes_on_disk * existing_rows_count.value() / rows_count;

return bytes_on_disk;

}

size_t IMergeTreeDataPart::getFileSizeOrZero(const String & file_name) const

{

auto checksum = checksums.files.find(file_name);

if (checksum == checksums.files.end())

return 0;

return checksum->second.file_size;

}

String IMergeTreeDataPart::getColumnNameWithMinimumCompressedSize(const NamesAndTypesList & available_columns) const

{

std::optional<std::string> minimum_size_column;

UInt64 minimum_size = std::numeric_limits<UInt64>::max();

for (const auto & column : available_columns)

{

if (!hasColumnFiles(column))

continue;

const auto size = getColumnSize(column.name).data_compressed;

if (size < minimum_size)

{

minimum_size = size;

minimum_size_column = column.name;

}

}

if (!minimum_size_column)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Could not find a column of minimum size in MergeTree, part {}",

getDataPartStorage().getFullPath());