Support batch delete[part 2]

be/src/olap/reader.cpp

@@ -92,7 +92,7 @@ class CollectIterator {
         }
 
         const RowCursor* current_row(bool* delete_flag) const {
-            *delete_flag = _is_delete;
+            *delete_flag = _is_delete || _current_row->is_delete();
             return _current_row;
         }
 
@@ -109,6 +109,9 @@ class CollectIterator {
             auto res = _refresh_current_row();
             *row = _current_row;
             *delete_flag = _is_delete;
+            if (_current_row!= nullptr) {
+                *delete_flag = _is_delete || _current_row->is_delete();
+            };
             return res;
         }
 
@@ -411,7 +414,6 @@ OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, MemPool* mem_pool
             *eof = true;
             return OLAP_SUCCESS;
         }
-
         cur_delete_flag = _next_delete_flag;
         // the verion is in reverse order, the first row is the highest version,
         // in UNIQUE_KEY highest version is the final result, there is no need to
@@ -427,14 +429,16 @@ OLAPStatus Reader::_unique_key_next_row(RowCursor* row_cursor, MemPool* mem_pool
                 }
                 break;
             }
-
             // break while can NOT doing aggregation
             if (!equal_row(_key_cids, *row_cursor, *_next_key)) {
                 break;
             }
             ++merged_count;
         }
-        if (!cur_delete_flag) {
+
+        // if reader needs to filter delete row and current delete_flag is ture, 
+        // then continue
+        if (!(cur_delete_flag && _filter_delete)) {
             break;
         }
         _stats.rows_del_filtered++;
@@ -1039,8 +1043,13 @@ OLAPStatus Reader::_init_delete_condition(const ReaderParams& read_params) {
                                               _tablet->delete_predicates(),
                                               read_params.version.second);
         _tablet->release_header_lock();
+
+        if (read_params.reader_type == READER_BASE_COMPACTION) {
+            _filter_delete = true;
+        }
         return ret;
-    } else {
+    } 
+    else {
         return OLAP_SUCCESS;
     }
 }

be/src/olap/reader.h

@@ -224,6 +224,7 @@ class Reader {
     bool _need_agg_finalize = true;
     ReaderType _reader_type = READER_QUERY;
     bool _next_delete_flag = false;
+    bool _filter_delete = false;
     const RowCursor* _next_key = nullptr;
     CollectIterator* _collect_iter = nullptr;
     std::vector<uint32_t> _key_cids;

be/src/olap/row_cursor.h

@@ -109,6 +109,14 @@ class RowCursor {
         return column_schema(index)->index_size();
     }
 
+    inline bool is_delete() const {
+        auto sign_idx = _schema->delete_sign_idx();
+        if (sign_idx < 0) {
+            return false;
+        }
+        return *reinterpret_cast<const char*>(cell(sign_idx).cell_ptr()) > 0;
+    }
+
     // set max/min for key field in _field_array
     OLAPStatus build_max_key();
     OLAPStatus build_min_key();