Merge pull request ClickHouse#88073 from ClickHouse/backport/25.8/87928

Backport ClickHouse#87928 to 25.8: Fix merge with projections when the last block is empty

Author: Algunenano

src/Storages/MergeTree/MergeProjectionPartsTask.h

@@ -50,7 +50,10 @@ class MergeProjectionPartsTask : public IExecutableTask
         }
 
     void onCompleted() override { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not implemented"); }
-    void cancel() noexcept override { chassert(false, "Not implemented"); }
+    void cancel() noexcept override
+    {
+        /// FIXME: See `executeHere` from MergeTask.h called in executeStep
+    }
     StorageID getStorageID() const override { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not implemented"); }
     Priority getPriority() const override { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not implemented"); }
     String getQueryId() const override { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not implemented"); }

src/Storages/MergeTree/MergeTask.cpp

@@ -852,6 +852,9 @@ void MergeTask::ExecuteAndFinalizeHorizontalPart::calculateProjections(const Blo
     {
         const auto & projection = *global_ctx->projections_to_rebuild[i];
         Block block_to_squash = projection.calculate(block, global_ctx->context);
+        /// Avoid replacing the projection squash header if nothing was generated (it used to return an empty block)
+        if (block_to_squash.rows() == 0)
+            return;
         auto & projection_squash_plan = ctx->projection_squashes[i];
         projection_squash_plan.setHeader(block_to_squash.cloneEmpty());
         Chunk squashed_chunk = Squashing::squash(projection_squash_plan.add({block_to_squash.getColumns(), block_to_squash.rows()}));

src/Storages/ProjectionsDescription.cpp

@@ -503,22 +503,20 @@ Block ProjectionDescription::calculate(const Block & block, ContextPtr context,
     pipeline.complete(sink);
     CompletedPipelineExecutor executor(pipeline);
     executor.execute();
-    Block projection_block;
-    if (sink->isAccumulatedSomething())
+
+    /// Always return the proper header, even if nothing was accumulated, in case the caller needs to use it
+    Block projection_block = sink->isAccumulatedSomething() ? sink->getPort().getHeader().cloneWithColumns(sink->detachAccumulatedColumns())
+                                                            : sink->getPort().getHeader().cloneEmpty();
+    /// Rename parent _part_offset to _parent_part_offset column
+    if (with_parent_part_offset)
     {
-      projection_block = sink->getPort().getHeader().cloneWithColumns(sink->detachAccumulatedColumns());
-
-      /// Rename parent _part_offset to _parent_part_offset column
-      if (with_parent_part_offset)
-      {
-          chassert(projection_block.has("_part_offset"));
-          chassert(!projection_block.has("_parent_part_offset"));
-
-          auto new_column = projection_block.getByName("_part_offset");
-          new_column.name = "_parent_part_offset";
-          projection_block.erase("_part_offset");
-          projection_block.insert(std::move(new_column));
-      }
+        chassert(projection_block.has("_part_offset"));
+        chassert(!projection_block.has("_parent_part_offset"));
+
+        auto new_column = projection_block.getByName("_part_offset");
+        new_column.name = "_parent_part_offset";
+        projection_block.erase("_part_offset");
+        projection_block.insert(std::move(new_column));
     }
 
     return projection_block;

tests/queries/0_stateless/03636_empty_projection_block.reference

@@ -0,0 +1,26 @@
+-- This test would hit a LOGICAL_ERROR during merge
+CREATE TABLE post_state
+(
+    `ts` DateTime,
+    `id` Int64,
+    `state` Nullable(UInt8) TTL ts + INTERVAL 1 MONTH,
+    PROJECTION p_digest_posts_state
+    (
+        SELECT
+            id,
+            argMax(state, ts) AS state
+        GROUP BY id
+    )
+)
+ENGINE = MergeTree()
+ORDER BY id
+TTL ts + toIntervalSecond(0) WHERE state IS NULL
+SETTINGS index_granularity = 8192, deduplicate_merge_projection_mode='rebuild';
+
+SYSTEM STOP MERGES post_state;
+INSERT INTO post_state VALUES ('2024-01-01 00:00:00', 1, NULL);
+INSERT INTO post_state VALUES ('2024-01-01 00:00:00', 1, NULL);
+INSERT INTO post_state VALUES ('2024-01-01 00:00:00', 1, 1);
+INSERT INTO post_state VALUES ('2024-01-01 00:00:00', 1, NULL);
+SYSTEM START MERGES post_state;
+OPTIMIZE TABLE post_state;