Non-blocking alter for vanilla merge tree

dbms/src/Interpreters/InterpreterAlterQuery.cpp

@@ -89,7 +89,6 @@ BlockIO InterpreterAlterQuery::execute()
 
     if (!partition_commands.empty())
     {
-        partition_commands.validate(*table);
         table->alterPartition(query_ptr, partition_commands, context);
     }
 

dbms/src/Storages/AlterCommands.cpp

@@ -86,16 +86,18 @@ std::optional<AlterCommand> AlterCommand::parse(const ASTAlterCommand * command_
 
         return command;
     }
-    else if (command_ast->type == ASTAlterCommand::DROP_COLUMN && !command_ast->partition)
+    else if (command_ast->type == ASTAlterCommand::DROP_COLUMN)
     {
-        if (command_ast->clear_column)
-            throw Exception(R"("ALTER TABLE table CLEAR COLUMN column" queries are not supported yet. Use "CLEAR COLUMN column IN PARTITION".)", ErrorCodes::NOT_IMPLEMENTED);
-
         AlterCommand command;
         command.ast = command_ast->clone();
         command.type = AlterCommand::DROP_COLUMN;
         command.column_name = getIdentifierName(command_ast->column);
         command.if_exists = command_ast->if_exists;
+        if (command_ast->clear_column)
+            command.clear = true;
+
+        if (command_ast->partition)
+            command.partition = command_ast->partition;
         return command;
     }
     else if (command_ast->type == ASTAlterCommand::MODIFY_COLUMN)
@@ -186,11 +188,8 @@ std::optional<AlterCommand> AlterCommand::parse(const ASTAlterCommand * command_
 
         return command;
     }
-    else if (command_ast->type == ASTAlterCommand::DROP_CONSTRAINT && !command_ast->partition)
+    else if (command_ast->type == ASTAlterCommand::DROP_CONSTRAINT)
     {
-        if (command_ast->clear_column)
-            throw Exception(R"("ALTER TABLE table CLEAR COLUMN column" queries are not supported yet. Use "CLEAR COLUMN column IN PARTITION".)", ErrorCodes::NOT_IMPLEMENTED);
-
         AlterCommand command;
         command.ast = command_ast->clone();
         command.if_exists = command_ast->if_exists;
@@ -199,16 +198,18 @@ std::optional<AlterCommand> AlterCommand::parse(const ASTAlterCommand * command_
 
         return command;
     }
-    else if (command_ast->type == ASTAlterCommand::DROP_INDEX && !command_ast->partition)
+    else if (command_ast->type == ASTAlterCommand::DROP_INDEX)
     {
-        if (command_ast->clear_column)
-            throw Exception(R"("ALTER TABLE table CLEAR INDEX index" queries are not supported yet. Use "CLEAR INDEX index IN PARTITION".)", ErrorCodes::NOT_IMPLEMENTED);
-
         AlterCommand command;
         command.ast = command_ast->clone();
         command.type = AlterCommand::DROP_INDEX;
         command.index_name = command_ast->index->as<ASTIdentifier &>().name;
         command.if_exists = command_ast->if_exists;
+        if (command_ast->clear_index)
+            command.clear = true;
+
+        if (command_ast->partition)
+            command.partition = command_ast->partition;
 
         return command;
     }
@@ -263,7 +264,9 @@ void AlterCommand::apply(StorageInMemoryMetadata & metadata) const
     }
     else if (type == DROP_COLUMN)
     {
-        metadata.columns.remove(column_name);
+        /// Otherwise just clear data on disk
+        if (!clear && !partition)
+            metadata.columns.remove(column_name);
     }
     else if (type == MODIFY_COLUMN)
     {
@@ -354,23 +357,25 @@ void AlterCommand::apply(StorageInMemoryMetadata & metadata) const
     }
     else if (type == DROP_INDEX)
     {
-        auto erase_it = std::find_if(
-                metadata.indices.indices.begin(),
-                metadata.indices.indices.end(),
-                [this](const ASTPtr & index_ast)
-                {
-                    return index_ast->as<ASTIndexDeclaration &>().name == index_name;
-                });
-
-        if (erase_it == metadata.indices.indices.end())
+        if (!partition && !clear)
         {
-            if (if_exists)
-                return;
-            throw Exception("Wrong index name. Cannot find index " + backQuote(index_name) + " to drop.",
-                            ErrorCodes::BAD_ARGUMENTS);
-        }
+            auto erase_it = std::find_if(
+                    metadata.indices.indices.begin(),
+                    metadata.indices.indices.end(),
+                    [this](const ASTPtr & index_ast)
+                    {
+                        return index_ast->as<ASTIndexDeclaration &>().name == index_name;
+                    });
+
+            if (erase_it == metadata.indices.indices.end())
+            {
+                if (if_exists)
+                    return;
+                throw Exception("Wrong index name. Cannot find index " + backQuote(index_name) + " to drop.", ErrorCodes::BAD_ARGUMENTS);
+            }
 
-        metadata.indices.indices.erase(erase_it);
+            metadata.indices.indices.erase(erase_it);
+        }
     }
     else if (type == ADD_CONSTRAINT)
     {
@@ -515,7 +520,7 @@ bool AlterCommand::isRequireMutationStage(const StorageInMemoryMetadata & metada
     if (ignore)
         return false;
 
-    if (type == DROP_COLUMN)
+    if (type == DROP_COLUMN || type == DROP_INDEX)
         return true;
 
     if (type != MODIFY_COLUMN || data_type == nullptr)
@@ -564,12 +569,21 @@ std::optional<MutationCommand> AlterCommand::tryConvertToMutationCommand(const S
     {
         result.type = MutationCommand::Type::DROP_COLUMN;
         result.column_name = column_name;
+        if (clear)
+            result.clear = true;
+        if (partition)
+            result.partition = partition;
         result.predicate = nullptr;
     }
     else if (type == DROP_INDEX)
     {
         result.type = MutationCommand::Type::DROP_INDEX;
-        result.column_name = column_name;
+        result.column_name = index_name;
+        if (clear)
+            result.clear = true;
+        if (partition)
+            result.partition = partition;
+
         result.predicate = nullptr;
     }
 

dbms/src/Storages/AlterCommands.h

@@ -41,8 +41,8 @@ struct AlterCommand
 
     String column_name;
 
-    /// For DROP COLUMN ... FROM PARTITION
-    String partition_name;
+    /// For DROP/CLEAR COLUMN/INDEX ... IN PARTITION
+    ASTPtr partition;
 
     /// For ADD and MODIFY, a new column type.
     DataTypePtr data_type = nullptr;
@@ -84,6 +84,9 @@ struct AlterCommand
     /// indicates that this command should not be applied, for example in case of if_exists=true and column doesn't exist.
     bool ignore = false;
 
+    /// Clear columns or index (don't drop from metadata)
+    bool clear = false;
+
     /// For ADD and MODIFY
     CompressionCodecPtr codec = nullptr;
 

dbms/src/Storages/IStorage.h

@@ -44,7 +44,8 @@ using SettingsChanges = std::vector<SettingChange>;
 
 class AlterCommands;
 class MutationCommands;
-class PartitionCommands;
+struct PartitionCommand;
+using PartitionCommands = std::vector<PartitionCommand>;
 
 class IProcessor;
 using ProcessorPtr = std::shared_ptr<IProcessor>;

dbms/src/Storages/MergeTree/DataPartsExchange.cpp

@@ -89,8 +89,6 @@ void Service::processQuery(const Poco::Net::HTMLForm & params, ReadBuffer & /*bo
 
         MergeTreeData::DataPartPtr part = findPart(part_name);
 
-        std::shared_lock<std::shared_mutex> part_lock(part->columns_lock);
-
         CurrentMetrics::Increment metric_increment{CurrentMetrics::ReplicatedSend};
 
         /// We'll take a list of files from the list of checksums.

dbms/src/Storages/MergeTree/IMergeTreeDataPart.cpp

@@ -735,8 +735,6 @@ void IMergeTreeDataPart::remove() const
 #    pragma GCC diagnostic push
 #    pragma GCC diagnostic ignored "-Wunused-variable"
 #endif
-        std::shared_lock<std::shared_mutex> lock(columns_lock);
-
         for (const auto & [file, _] : checksums.files)
             disk->remove(to + "/" + file);
 #if !__clang__

dbms/src/Storages/MergeTree/IMergeTreeDataPart.h

@@ -15,7 +15,6 @@
 #include <Storages/MergeTree/MergeTreeDataPartChecksum.h>
 #include <Storages/MergeTree/MergeTreeDataPartTTLInfo.h>
 #include <Storages/MergeTree/MergeTreeIOSettings.h>
-#include <Storages/MergeTree/AlterAnalysisResult.h>
 #include <Storages/MergeTree/KeyCondition.h>
 #include <Columns/IColumn.h>
 
@@ -94,19 +93,13 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
 
     /// NOTE: Returns zeros if column files are not found in checksums.
     /// NOTE: You must ensure that no ALTERs are in progress when calculating ColumnSizes.
-    ///   (either by locking columns_lock, or by locking table structure).
+    ///   (by locking table structure).
     virtual ColumnSize getColumnSize(const String & /* name */, const IDataType & /* type */) const { return {}; }
 
     virtual ColumnSize getTotalColumnsSize() const { return {}; }
 
     virtual String getFileNameForColumn(const NameAndTypePair & column) const = 0;
 
-    /// Returns rename map of column files for the alter converting expression onto new table files.
-    /// Files to be deleted are mapped to an empty string in rename map.
-    virtual NameToNameMap createRenameMapForAlter(
-        AlterAnalysisResult & /* analysis_result */,
-        const NamesAndTypesList & /* old_columns */) const { return {}; }
-
     virtual ~IMergeTreeDataPart();
 
     using ColumnToSize = std::map<std::string, UInt64>;
@@ -283,11 +276,6 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
     /// Columns with values, that all have been zeroed by expired ttl
     NameSet expired_columns;
 
-    /** It is blocked for writing when changing columns, checksums or any part files.
-        * Locked to read when reading columns, checksums or any part files.
-        */
-    mutable std::shared_mutex columns_lock;
-
     /// For data in RAM ('index')
     UInt64 getIndexSizeInBytes() const;
     UInt64 getIndexSizeInAllocatedBytes() const;

dbms/src/Storages/MergeTree/IMergedBlockOutputStream.cpp

@@ -33,4 +33,74 @@ Block IMergedBlockOutputStream::getBlockAndPermute(const Block & block, const Na
     return result;
 }
 
+NameSet IMergedBlockOutputStream::removeEmptyColumnsFromPart(
+    const MergeTreeDataPartPtr & data_part,
+    NamesAndTypesList & columns,
+    MergeTreeData::DataPart::Checksums & checksums)
+{
+    const NameSet & empty_columns = data_part->expired_columns;
+
+    /// For compact part we have to override whole file with data, it's not
+    /// worth it
+    if (empty_columns.empty() || isCompactPart(data_part))
+        return {};
+
+    /// Collect counts for shared streams of different columns. As an example, Nested columns have shared stream with array sizes.
+    std::map<String, size_t> stream_counts;
+    for (const NameAndTypePair & column : columns)
+    {
+        column.type->enumerateStreams(
+            [&](const IDataType::SubstreamPath & substream_path)
+            {
+                ++stream_counts[IDataType::getFileNameForStream(column.name, substream_path)];
+            },
+            {});
+    }
+
+    NameSet remove_files;
+    const String mrk_extension = data_part->getMarksFileExtension();
+    for (const auto & column_name : empty_columns)
+    {
+        IDataType::StreamCallback callback = [&](const IDataType::SubstreamPath & substream_path)
+        {
+            String stream_name = IDataType::getFileNameForStream(column_name, substream_path);
+            /// Delete files if they are no longer shared with another column.
+            if (--stream_counts[stream_name] == 0)
+            {
+                remove_files.emplace(stream_name + ".bin");
+                remove_files.emplace(stream_name + mrk_extension);
+            }
+        };
+        IDataType::SubstreamPath stream_path;
+        auto column_with_type = columns.tryGetByName(column_name);
+        if (column_with_type)
+            column_with_type->type->enumerateStreams(callback, stream_path);
+    }
+
+    /// Remove files on disk and checksums
+    for (const String & removed_file : remove_files)
+    {
+        if (checksums.files.count(removed_file))
+        {
+            data_part->disk->remove(data_part->getFullRelativePath() + removed_file);
+            checksums.files.erase(removed_file);
+        }
+    }
+
+    /// Remove columns from columns array
+    for (const String & empty_column_name : empty_columns)
+    {
+        auto find_func = [&empty_column_name](const auto & pair) -> bool
+        {
+            return pair.name == empty_column_name;
+        };
+        auto remove_it
+            = std::find_if(columns.begin(), columns.end(), find_func);
+
+        if (remove_it != columns.end())
+            columns.erase(remove_it);
+    }
+    return remove_files;
+}
+
 }

dbms/src/Storages/MergeTree/IMergedBlockOutputStream.h

@@ -27,6 +27,13 @@ class IMergedBlockOutputStream : public IBlockOutputStream
 
     IDataType::OutputStreamGetter createStreamGetter(const String & name, WrittenOffsetColumns & offset_columns, bool skip_offsets);
 
+    /// Remove all columns marked expired in data_part. Also, clears checksums
+    /// and columns array. Return set of removed files names.
+    NameSet removeEmptyColumnsFromPart(
+        const MergeTreeDataPartPtr & data_part,
+        NamesAndTypesList & columns,
+        MergeTreeData::DataPart::Checksums & checksums);
+
 protected:
     const MergeTreeData & storage;
 

dbms/src/Storages/MergeTree/MergeList.cpp

@@ -27,8 +27,6 @@ MergeListElement::MergeListElement(const std::string & database_, const std::str
         source_part_names.emplace_back(source_part->name);
         source_part_paths.emplace_back(source_part->getFullPath());
 
-        std::shared_lock<std::shared_mutex> part_lock(source_part->columns_lock);
-
         total_size_bytes_compressed += source_part->bytes_on_disk;
         total_size_marks += source_part->getMarksCount();
         total_rows_count += source_part->index_granularity.getTotalRows();