Merge pull request ClickHouse#79281 from CurtizJ/lwu-on-fly

Support applying lightweight deletes on the fly

Author: CurtizJ

src/Storages/MergeTree/AlterConversions.cpp

@@ -1,11 +1,13 @@
 #include <Storages/MergeTree/AlterConversions.h>
 #include <Storages/MergeTree/LoadedMergeTreeDataPartInfoForReader.h>
 #include <Storages/MergeTree/MergeTreeRangeReader.h>
+#include <Storages/MergeTree/MergeTreeVirtualColumns.h>
 #include <Storages/MutationCommands.h>
 #include <Interpreters/MutationsInterpreter.h>
 #include <Interpreters/MutationsNonDeterministicHelpers.h>
 #include <Parsers/ASTAlterQuery.h>
 #include <Parsers/ASTAssignment.h>
+#include <Parsers/ASTLiteral.h>
 #include <Common/ProfileEvents.h>
 #include <ranges>
 
@@ -64,6 +66,41 @@ static MutationCommand createCommandWithUpdatedColumns(
     return res;
 }
 
+static bool isLightweightDeleteCommand(const String & column_name, const ASTPtr & ast)
+{
+    if (column_name != RowExistsColumn::name)
+        return false;
+
+    const auto * literal = ast->as<ASTLiteral>();
+    if (!literal)
+        return false;
+
+    if (literal->value.getType() != Field::Types::UInt64)
+        return false;
+
+    return literal->value.safeGet<UInt64>() == 0;
+}
+
+static MutationCommand createLightweightDeleteCommand(const MutationCommand & command)
+{
+    chassert(command.type == MutationCommand::Type::UPDATE);
+    chassert(command.predicate != nullptr);
+
+    auto alter_command = std::make_shared<ASTAlterCommand>();
+    alter_command->type = ASTAlterCommand::DELETE;
+
+    if (command.partition)
+        alter_command->partition = alter_command->children.emplace_back(command.partition->clone()).get();
+
+    alter_command->predicate = alter_command->children.emplace_back(command.predicate->clone()).get();
+    auto mutation_command = MutationCommand::parse(alter_command.get());
+
+    if (!mutation_command)
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Failed to parse command {}", alter_command->formatForErrorMessage());
+
+    return *mutation_command;
+}
+
 AlterConversions::AlterConversions(
     const MutationCommands & mutation_commands_,
     const ContextPtr & context)
@@ -308,16 +345,29 @@ MutationCommands AlterConversions::filterMutationCommands(Names & read_columns,
         }
         else if (command.type == MutationCommand::Type::UPDATE)
         {
+            bool has_lightweight_delete = false;
             std::unordered_map<String, ASTPtr> new_updated_columns;
+
             for (const auto & [column, ast] : command.column_to_update_expression)
             {
-                if (read_columns_set.contains(column))
+                if (isLightweightDeleteCommand(column, ast))
+                {
+                    has_lightweight_delete = true;
+                }
+                else if (read_columns_set.contains(column))
                 {
                     ast->collectIdentifierNames(source_columns);
                     new_updated_columns.emplace(column, ast->clone());
                 }
             }
 
+            if (has_lightweight_delete)
+            {
+                auto new_command = createLightweightDeleteCommand(command);
+                new_command.predicate->collectIdentifierNames(source_columns);
+                filtered_commands.push_back(std::move(new_command));
+            }
+
             if (!new_updated_columns.empty())
             {
                 auto new_command = createCommandWithUpdatedColumns(command, std::move(new_updated_columns));

tests/queries/0_stateless/03442_lightweight_deletes_on_fly.reference

@@ -0,0 +1,2 @@
+3
+9850

tests/queries/0_stateless/03442_lightweight_deletes_on_fly.sql

@@ -0,0 +1,31 @@
+DROP TABLE IF EXISTS test_deletes;
+
+CREATE TABLE test_deletes (a UInt64) ENGINE = MergeTree ORDER BY a;
+
+INSERT INTO test_deletes VALUES (1) (2) (3);
+
+SYSTEM STOP MERGES test_deletes;
+SET mutations_sync = 0;
+SET lightweight_deletes_sync = 0;
+
+ALTER TABLE test_deletes DELETE WHERE a = 1 SETTINGS mutations_sync = 0;
+DELETE FROM test_deletes WHERE a = 2 SETTINGS lightweight_deletes_sync = 0;
+
+SELECT a FROM test_deletes SETTINGS apply_mutations_on_fly = 1;
+
+DROP TABLE test_deletes;
+
+CREATE TABLE test_deletes (a UInt64, b UInt64) ENGINE = MergeTree ORDER BY a;
+
+INSERT INTO test_deletes SELECT number, 0 FROM numbers(10000);
+
+DELETE FROM test_deletes WHERE a >= 100 AND a < 200 SETTINGS lightweight_deletes_sync = 1;
+
+SYSTEM STOP MERGES test_deletes;
+
+ALTER TABLE test_deletes UPDATE b = 1 WHERE a >= 150 AND a < 250 SETTINGS mutations_sync = 0;
+DELETE FROM test_deletes WHERE b = 1 SETTINGS lightweight_deletes_sync = 0;
+
+SELECT count() FROM test_deletes SETTINGS apply_mutations_on_fly = 1;
+
+DROP TABLE test_deletes;