#include "MergeTreeDataPartWide.h" #include #include #include #include #include #include namespace DB { namespace ErrorCodes { extern const int CANNOT_READ_ALL_DATA; extern const int NO_FILE_IN_DATA_PART; extern const int BAD_SIZE_OF_FILE_IN_DATA_PART; extern const int LOGICAL_ERROR; } MergeTreeDataPartWide::MergeTreeDataPartWide( MergeTreeData & storage_, const String & name_, const VolumePtr & volume_, const std::optional & relative_path_) : IMergeTreeDataPart(storage_, name_, volume_, relative_path_, Type::WIDE) { } MergeTreeDataPartWide::MergeTreeDataPartWide( const MergeTreeData & storage_, const String & name_, const MergeTreePartInfo & info_, const VolumePtr & volume_, const std::optional & relative_path_) : IMergeTreeDataPart(storage_, name_, info_, volume_, relative_path_, Type::WIDE) { } IMergeTreeDataPart::MergeTreeReaderPtr MergeTreeDataPartWide::getReader( const NamesAndTypesList & columns_to_read, const StorageMetadataPtr & metadata_snapshot, const MarkRanges & mark_ranges, UncompressedCache * uncompressed_cache, MarkCache * mark_cache, const MergeTreeReaderSettings & reader_settings, const ValueSizeMap & avg_value_size_hints, const ReadBufferFromFileBase::ProfileCallback & profile_callback) const { auto new_settings = reader_settings; new_settings.convert_nested_to_subcolumns = true; auto ptr = std::static_pointer_cast(shared_from_this()); return std::make_unique( ptr, columns_to_read, metadata_snapshot, uncompressed_cache, mark_cache, mark_ranges, new_settings, avg_value_size_hints, profile_callback); } IMergeTreeDataPart::MergeTreeWriterPtr MergeTreeDataPartWide::getWriter( const NamesAndTypesList & columns_list, const StorageMetadataPtr & metadata_snapshot, const std::vector & indices_to_recalc, const CompressionCodecPtr & default_codec_, const MergeTreeWriterSettings & writer_settings, const MergeTreeIndexGranularity & computed_index_granularity) const { return std::make_unique( shared_from_this(), columns_list, metadata_snapshot, indices_to_recalc, index_granularity_info.marks_file_extension, default_codec_, writer_settings, computed_index_granularity); } /// Takes into account the fact that several columns can e.g. share their .size substreams. /// When calculating totals these should be counted only once. ColumnSize MergeTreeDataPartWide::getColumnSizeImpl( const NameAndTypePair & column, std::unordered_set * processed_substreams) const { ColumnSize size; if (checksums.empty()) return size; column.type->enumerateStreams([&](const IDataType::SubstreamPath & substream_path, const IDataType & /* substream_type */) { String file_name = IDataType::getFileNameForStream(column, substream_path); if (processed_substreams && !processed_substreams->insert(file_name).second) return; auto bin_checksum = checksums.files.find(file_name + ".bin"); if (bin_checksum != checksums.files.end()) { size.data_compressed += bin_checksum->second.file_size; size.data_uncompressed += bin_checksum->second.uncompressed_size; } auto mrk_checksum = checksums.files.find(file_name + index_granularity_info.marks_file_extension); if (mrk_checksum != checksums.files.end()) size.marks += mrk_checksum->second.file_size; }, {}); return size; } void MergeTreeDataPartWide::loadIndexGranularity() { String full_path = getFullRelativePath(); index_granularity_info.changeGranularityIfRequired(volume->getDisk(), full_path); if (columns.empty()) throw Exception("No columns in part " + name, ErrorCodes::NO_FILE_IN_DATA_PART); /// We can use any column, it doesn't matter std::string marks_file_path = index_granularity_info.getMarksFilePath(full_path + getFileNameForColumn(columns.front())); if (!volume->getDisk()->exists(marks_file_path)) throw Exception("Marks file '" + fullPath(volume->getDisk(), marks_file_path) + "' doesn't exist", ErrorCodes::NO_FILE_IN_DATA_PART); size_t marks_file_size = volume->getDisk()->getFileSize(marks_file_path); if (!index_granularity_info.is_adaptive) { size_t marks_count = marks_file_size / index_granularity_info.getMarkSizeInBytes(); index_granularity.resizeWithFixedGranularity(marks_count, index_granularity_info.fixed_index_granularity); /// all the same } else { auto buffer = volume->getDisk()->readFile(marks_file_path, marks_file_size); while (!buffer->eof()) { buffer->seek(sizeof(size_t) * 2, SEEK_CUR); /// skip offset_in_compressed file and offset_in_decompressed_block size_t granularity; readIntBinary(granularity, *buffer); index_granularity.appendMark(granularity); } if (index_granularity.getMarksCount() * index_granularity_info.getMarkSizeInBytes() != marks_file_size) throw Exception("Cannot read all marks from file " + fullPath(volume->getDisk(), marks_file_path), ErrorCodes::CANNOT_READ_ALL_DATA); } index_granularity.setInitialized(); } MergeTreeDataPartWide::~MergeTreeDataPartWide() { removeIfNeeded(); } void MergeTreeDataPartWide::checkConsistency(bool require_part_metadata) const { checkConsistencyBase(); String path = getFullRelativePath(); if (!checksums.empty()) { if (require_part_metadata) { for (const NameAndTypePair & name_type : columns) { IDataType::SubstreamPath stream_path; name_type.type->enumerateStreams([&](const IDataType::SubstreamPath & substream_path, const IDataType & /* substream_type */) { String file_name = IDataType::getFileNameForStream(name_type, substream_path); String mrk_file_name = file_name + index_granularity_info.marks_file_extension; String bin_file_name = file_name + ".bin"; if (!checksums.files.count(mrk_file_name)) throw Exception("No " + mrk_file_name + " file checksum for column " + name_type.name + " in part " + fullPath(volume->getDisk(), path), ErrorCodes::NO_FILE_IN_DATA_PART); if (!checksums.files.count(bin_file_name)) throw Exception("No " + bin_file_name + " file checksum for column " + name_type.name + " in part " + fullPath(volume->getDisk(), path), ErrorCodes::NO_FILE_IN_DATA_PART); }, stream_path); } } } else { /// Check that all marks are nonempty and have the same size. std::optional marks_size; for (const NameAndTypePair & name_type : columns) { name_type.type->enumerateStreams([&](const IDataType::SubstreamPath & substream_path, const IDataType & /* substream_type */) { auto file_path = path + IDataType::getFileNameForStream(name_type, substream_path) + index_granularity_info.marks_file_extension; /// Missing file is Ok for case when new column was added. if (volume->getDisk()->exists(file_path)) { UInt64 file_size = volume->getDisk()->getFileSize(file_path); if (!file_size) throw Exception("Part " + path + " is broken: " + fullPath(volume->getDisk(), file_path) + " is empty.", ErrorCodes::BAD_SIZE_OF_FILE_IN_DATA_PART); if (!marks_size) marks_size = file_size; else if (file_size != *marks_size) throw Exception("Part " + path + " is broken: marks have different sizes.", ErrorCodes::BAD_SIZE_OF_FILE_IN_DATA_PART); } }); } } } bool MergeTreeDataPartWide::hasColumnFiles(const NameAndTypePair & column) const { bool res = true; column.type->enumerateStreams([&](const IDataType::SubstreamPath & substream_path, const IDataType & /* substream_type */) { String file_name = IDataType::getFileNameForStream(column, substream_path); auto bin_checksum = checksums.files.find(file_name + ".bin"); auto mrk_checksum = checksums.files.find(file_name + index_granularity_info.marks_file_extension); if (bin_checksum == checksums.files.end() || mrk_checksum == checksums.files.end()) res = false; }, {}); return res; } String MergeTreeDataPartWide::getFileNameForColumn(const NameAndTypePair & column) const { String filename; column.type->enumerateStreams([&](const IDataType::SubstreamPath & substream_path, const IDataType & /* substream_type */) { if (filename.empty()) filename = IDataType::getFileNameForStream(column, substream_path); }); return filename; } void MergeTreeDataPartWide::calculateEachColumnSizes(ColumnSizeByName & each_columns_size, ColumnSize & total_size) const { std::unordered_set processed_substreams; for (const NameAndTypePair & column : columns) { ColumnSize size = getColumnSizeImpl(column, &processed_substreams); each_columns_size[column.name] = size; total_size.add(size); #ifndef NDEBUG /// Most trivial types if (rows_count != 0 && column.type->isValueRepresentedByNumber() && !column.type->haveSubtypes()) { size_t rows_in_column = size.data_uncompressed / column.type->getSizeOfValueInMemory(); if (rows_in_column != rows_count) { throw Exception( ErrorCodes::LOGICAL_ERROR, "Column {} has rows count {} according to size in memory " "and size of single value, but data part {} has {} rows", backQuote(column.name), rows_in_column, name, rows_count); } } #endif } } }