storage: fix block rows not match when filter column is the first non-empty column in the block (#9483)

ref #9472

storage: fix block rows not match when the filter column is the first non-empty column in the block

Signed-off-by: Lloyd-Pottiger <yan1579196623@gmail.com>

Co-authored-by: JaySon <tshent@qq.com>

Author: Lloyd-Pottiger

dbms/src/Core/Block.cpp

@@ -541,11 +541,9 @@ Block hstackBlocks(Blocks && blocks, const Block & header)
         return {};
 
     Block res = header.cloneEmpty();
-    size_t num_rows = blocks.front().rows();
     auto rs_result = DM::RSResult::All;
     for (const auto & block : blocks)
     {
-        RUNTIME_CHECK_MSG(block.rows() == num_rows, "Cannot hstack blocks with different number of rows");
         for (const auto & elem : block)
         {
             if (likely(res.has(elem.name)))
@@ -556,6 +554,7 @@ Block hstackBlocks(Blocks && blocks, const Block & header)
         rs_result = rs_result && block.getRSResult();
     }
     res.setRSResult(rs_result);
+    res.checkNumberOfRows();
 
     return res;
 }

dbms/src/Core/Block.h

@@ -180,11 +180,14 @@ using BucketBlocksListMap = std::map<Int32, BlocksList>;
 /// Join blocks by columns
 /// The schema of the output block is the same as the header block.
 /// The columns not in the header block will be ignored.
-/// For example:
-/// header: (a UInt32, b UInt32, c UInt32, d UInt32)
-/// block1: (a UInt32, b UInt32, c UInt32, e UInt32), rows: 3
-/// block2: (d UInt32), rows: 3
-/// result: (a UInt32, b UInt32, c UInt32, d UInt32), rows: 3
+/// NOTE: The input blocks can have columns with different sizes,
+///       but the columns in the header block must have the same size,
+///       Otherwise, an exception will be thrown.
+/// Example:
+///   header: (a UInt32, b UInt32, c UInt32, d UInt32)
+///   block1: (a UInt32, b UInt32, c UInt32, e UInt32), rows: 3
+///   block2: (d UInt32), rows: 3
+///   result: (a UInt32, b UInt32, c UInt32, d UInt32), rows: 3
 Block hstackBlocks(Blocks && blocks, const Block & header);
 
 /// Join blocks by rows

dbms/src/DataStreams/FilterTransformAction.cpp

@@ -39,8 +39,8 @@ FilterTransformAction::FilterTransformAction(
     /// Determine position of filter column.
     expression->execute(header);
 
-    filter_column = header.getPositionByName(filter_column_name);
-    auto & column_elem = header.safeGetByPosition(filter_column);
+    filter_column_position = header.getPositionByName(filter_column_name);
+    auto & column_elem = header.safeGetByPosition(filter_column_position);
 
     /// Isn't the filter already constant?
     if (column_elem.column)
@@ -76,9 +76,10 @@ bool FilterTransformAction::transform(Block & block, FilterPtr & res_filter, boo
 
     if (block.getRSResult().allMatch())
     {
-        block.insert(
-            filter_column,
-            header.safeGetByPosition(filter_column)); // Make some checks on block structure happy.
+        auto filter_column = header.safeGetByPosition(filter_column_position).cloneEmpty();
+        filter_column.column = filter_column.type->createColumnConst(block.rows(), static_cast<UInt64>(1));
+        // Make some checks on block structure happy.
+        block.insert(filter_column_position, std::move(filter_column));
         if (return_filter)
             res_filter = nullptr;
         return true;
@@ -95,7 +96,7 @@ bool FilterTransformAction::transform(Block & block, FilterPtr & res_filter, boo
 
     size_t columns = block.columns();
     size_t rows = block.rows();
-    ColumnPtr column_of_filter = block.safeGetByPosition(filter_column).column;
+    ColumnPtr column_of_filter = block.safeGetByPosition(filter_column_position).column;
 
     /** It happens that at the stage of analysis of expressions (in sample_block) the columns-constants have not been calculated yet,
         *  and now - are calculated. That is, not all cases are covered by the code above.
@@ -139,8 +140,8 @@ bool FilterTransformAction::transform(Block & block, FilterPtr & res_filter, boo
     if (filtered_rows == rows)
     {
         /// Replace the column with the filter by a constant.
-        block.safeGetByPosition(filter_column).column
-            = block.safeGetByPosition(filter_column).type->createColumnConst(filtered_rows, static_cast<UInt64>(1));
+        auto filter_column = block.safeGetByPosition(filter_column_position);
+        filter_column.column = filter_column.type->createColumnConst(filtered_rows, static_cast<UInt64>(1));
         /// No need to touch the rest of the columns.
         return true;
     }
@@ -150,7 +151,7 @@ bool FilterTransformAction::transform(Block & block, FilterPtr & res_filter, boo
     {
         ColumnWithTypeAndName & current_column = block.safeGetByPosition(i);
 
-        if (i == filter_column)
+        if (i == filter_column_position)
         {
             /// The column with filter itself is replaced with a column with a constant `1`, since after filtering, nothing else will remain.
             /// NOTE User could pass column with something different than 0 and 1 for filter.

dbms/src/DataStreams/FilterTransformAction.h

@@ -41,7 +41,7 @@ struct FilterTransformAction
 private:
     Block header;
     ExpressionActionsPtr expression;
-    size_t filter_column;
+    size_t filter_column_position;
 
     ConstantFilterDescription constant_filter_description;
     IColumn::Filter * filter = nullptr;

dbms/src/Storages/DeltaMerge/LateMaterializationBlockInputStream.cpp

@@ -21,6 +21,32 @@
 namespace DB::DM
 {
 
+namespace
+{
+
+void filterFilterColumnBlock(
+    const Block & header,
+    Block & block,
+    const IColumn::Filter & filter,
+    size_t passed_count,
+    const String & filter_column_name)
+{
+    ColumnPtr filter_column;
+    for (auto & col : block)
+    {
+        if (col.name == filter_column_name)
+        {
+            filter_column = col.column;
+            continue;
+        }
+        col.column = col.column->filter(filter, passed_count);
+    }
+    if (header.has(filter_column_name))
+        filter_column = filter_column->filter(filter, passed_count);
+}
+
+} // namespace
+
 LateMaterializationBlockInputStream::LateMaterializationBlockInputStream(
     const ColumnDefines & columns_to_read,
     const String & filter_column_name_,
@@ -69,12 +95,7 @@ Block LateMaterializationBlockInputStream::read()
                 {
                     col.column = col.column->filter(col_filter, passed_count);
                 }
-                for (auto & col : filter_column_block)
-                {
-                    if (col.name == filter_column_name)
-                        continue;
-                    col.column = col.column->filter(col_filter, passed_count);
-                }
+                filterFilterColumnBlock(header, filter_column_block, col_filter, passed_count, filter_column_name);
             }
             return hstackBlocks({std::move(filter_column_block), std::move(rest_column_block)}, header);
         }
@@ -112,12 +133,7 @@ Block LateMaterializationBlockInputStream::read()
                 // so only if the number of rows left after filtering out is large enough,
                 // we can skip some packs of the next block, call readWithFilter to get the next block.
                 rest_column_block = rest_column_stream->readWithFilter(*filter);
-                for (auto & col : filter_column_block)
-                {
-                    if (col.name == filter_column_name)
-                        continue;
-                    col.column = col.column->filter(*filter, passed_count);
-                }
+                filterFilterColumnBlock(header, filter_column_block, *filter, passed_count, filter_column_name);
             }
             else if (filter_out_count > 0)
             {
@@ -128,12 +144,7 @@ Block LateMaterializationBlockInputStream::read()
                 {
                     col.column = col.column->filter(*filter, passed_count);
                 }
-                for (auto & col : filter_column_block)
-                {
-                    if (col.name == filter_column_name)
-                        continue;
-                    col.column = col.column->filter(*filter, passed_count);
-                }
+                filterFilterColumnBlock(header, filter_column_block, *filter, passed_count, filter_column_name);
             }
             else
             {