fix data race in PhysicalFilter (infiniflow#548)

Author: yangzq50

src/executor/expression/expression_selector.cpp

@@ -32,7 +32,6 @@ import data_type;
 
 import infinity_exception;
 
-
 namespace infinity {
 
 SizeT ExpressionSelector::Select(const SharedPtr<BaseExpression> &expr,
@@ -48,8 +47,8 @@ SizeT ExpressionSelector::Select(const SharedPtr<BaseExpression> &expr,
 
     Select(expr, state, count, input_select, output_true_select, output_false_select);
 
-    output_data_block->UnInit();
     // Shrink the input data block into output data block
+    // this Init function will throw if output_data_block is already initialized before
     output_data_block->Init(input_data_block, output_true_select);
     return output_true_select->Size();
 }

src/executor/operator/physical_filter.cpp

@@ -14,6 +14,8 @@
 
 module;
 
+module physical_filter;
+
 import stl;
 import txn;
 import query_context;
@@ -23,14 +25,13 @@ import data_table;
 import physical_operator_type;
 import operator_state;
 import expression_state;
+import expression_selector;
 import data_block;
 import logger;
 import third_party;
 
 import infinity_exception;
 
-module physical_filter;
-
 namespace infinity {
 
 void PhysicalFilter::Init() {
@@ -66,19 +67,17 @@ bool PhysicalFilter::Execute(QueryContext *, OperatorState *operator_state) {
 
     for(SizeT block_idx = 0; block_idx < input_block_count; ++ block_idx) {
 
+        // create uninitialized data block for output
         UniquePtr<DataBlock> data_block = DataBlock::MakeUniquePtr();
-        data_block->Init(*GetOutputTypes());
         DataBlock* output_data_block = data_block.get();
         operator_state->data_block_array_.emplace_back(std::move(data_block));
 
         SharedPtr<ExpressionState> condition_state = ExpressionState::CreateState(condition_);
         DataBlock* input_data_block = prev_op_state->data_block_array_[block_idx].get();
 
-        SizeT selected_count = selector_.Select(condition_,
-                                                condition_state,
-                                                input_data_block,
-                                                output_data_block,
-                                                input_data_block->row_count());
+        // selector contains a pointer to input data, which should not be shared by multiple tasks
+        ExpressionSelector selector;
+        SizeT selected_count = selector.Select(condition_, condition_state, input_data_block, output_data_block, input_data_block->row_count());
 
         LOG_TRACE(fmt::format("{} rows after filter", selected_count));
     }

src/executor/operator/physical_filter.cppm

@@ -24,8 +24,6 @@ import physical_operator;
 import physical_operator_type;
 import base_expression;
 import data_table;
-import expression_evaluator;
-import expression_selector;
 import load_meta;
 import infinity_exception;
 import internal_types;
@@ -55,9 +53,6 @@ public:
 private:
     SharedPtr<BaseExpression> condition_;
 
-    ExpressionEvaluator executor_;
-    ExpressionSelector selector_;
-
     SharedPtr<DataTable> input_table_{};
 };
 

src/executor/operator/physical_index_scan.cpp

@@ -22,22 +22,12 @@ module physical_index_scan;
 import query_context;
 import operator_state;
 import default_values;
-import base_expression;
-import expression_type;
-import value_expression;
-import column_expression;
-import cast_expression;
-import function_expression;
-import expression_evaluator;
-import expression_state;
-import column_vector;
+import buffer_handle;
 import infinity_exception;
 import logger;
 import third_party;
 import txn;
 import data_block;
-import default_values;
-import buffer_handle;
 import secondary_index_scan_execute_expression;
 import logical_type;
 import segment_column_index_entry;
@@ -465,7 +455,6 @@ class FilterResult {
         // delete_filter: return false if the row is deleted
         std::visit(Overload{[&](const Bitmask &bitmask) {
                                 u32 output_block_row_id = 0;
-                                u32 output_block_idx = 0;
                                 DataBlock *output_block_ptr = output_data_blocks.back().get();
                                 // TODO: 64 bit in a loop?
                                 const u32 segment_row_count = SegmentRowCount();
@@ -495,8 +484,7 @@ class FilterResult {
                             },
                             [&](const Vector<u32> &selected_rows) {
                                 u32 output_block_row_id = 0;
-                                u32 output_block_idx = 0;
-                                DataBlock *output_block_ptr = output_data_blocks[output_block_idx++].get();
+                                DataBlock *output_block_ptr = output_data_blocks.back().get();
                                 for (u32 segment_offset : selected_rows) {
                                     if (!delete_filter(segment_offset)) {
                                         // deleted