enhance: push down vortex varchar filters

Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>

configs/milvus.yaml

@@ -581,6 +581,7 @@ queryNode:
     deletePoolSizeFactor: 1 # size factor (CPUNum * factor) of the DeleteBatch dispatch pool
     storageV2:
       cellTargetSizeBytes: 4194304 # Target average byte size per storage v2 cache cell. Parquet row groups are greedily packed so that rgs_per_cell * avg_row_group_size ≈ this target. Each cell always contains at least one row group and cells never cross file boundaries. Tune larger for bigger batch IO (fewer cells) or smaller to get finer cache granularity. Default 4 MiB.
+    enableVortexScanPushdown: true # Enable Vortex local-format varchar filter pushdown through row-id scan.
     deleteDumpBatchSize: 10000 # Batch size for delete snapshot dump in segcore.
   loadMemoryUsageFactor: 1 # The multiply factor of calculating the memory usage while loading segments
   enableDisk: false # enable querynode load disk index, and search on disk index

internal/core/src/common/Common.cpp

@@ -43,6 +43,8 @@ std::atomic<bool> CONFIG_PARAM_TYPE_CHECK_ENABLED(
     DEFAULT_CONFIG_PARAM_TYPE_CHECK_ENABLED);
 std::atomic<bool> ENABLE_PARQUET_STATS_SKIP_INDEX(
     DEFAULT_ENABLE_PARQUET_STATS_SKIP_INDEX);
+std::atomic<bool> ENABLE_VORTEX_SCAN_PUSHDOWN(
+    DEFAULT_ENABLE_VORTEX_SCAN_PUSHDOWN);
 
 void
 SetIndexSliceSize(const int64_t size) {
@@ -110,6 +112,13 @@ SetDefaultEnableParquetStatsSkipIndex(bool val) {
              ENABLE_PARQUET_STATS_SKIP_INDEX.load());
 }
 
+void
+SetDefaultVortexScanPushdownEnable(bool val) {
+    ENABLE_VORTEX_SCAN_PUSHDOWN.store(val);
+    LOG_INFO("set default enable vortex scan pushdown: {}",
+             ENABLE_VORTEX_SCAN_PUSHDOWN.load());
+}
+
 void
 SetEnableLatestDeleteSnapshotOptimization(bool val) {
     ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(val);

internal/core/src/common/Common.h

@@ -34,6 +34,7 @@ extern std::atomic<bool> JSON_KEY_STATS_ENABLED;
 extern std::atomic<bool> GROWING_JSON_KEY_STATS_ENABLED;
 extern std::atomic<bool> CONFIG_PARAM_TYPE_CHECK_ENABLED;
 extern std::atomic<bool> ENABLE_PARQUET_STATS_SKIP_INDEX;
+extern std::atomic<bool> ENABLE_VORTEX_SCAN_PUSHDOWN;
 
 void
 SetIndexSliceSize(const int64_t size);
@@ -62,6 +63,9 @@ SetDefaultConfigParamTypeCheck(bool val);
 void
 SetDefaultEnableParquetStatsSkipIndex(bool val);
 
+void
+SetDefaultVortexScanPushdownEnable(bool val);
+
 void
 SetEnableLatestDeleteSnapshotOptimization(bool val);
 

internal/core/src/common/Consts.h

@@ -118,6 +118,7 @@ const bool DEFAULT_JSON_KEY_STATS_ENABLED = true;
 const bool DEFAULT_GROWING_JSON_KEY_STATS_ENABLED = false;
 const bool DEFAULT_CONFIG_PARAM_TYPE_CHECK_ENABLED = true;
 const bool DEFAULT_ENABLE_PARQUET_STATS_SKIP_INDEX = false;
+const bool DEFAULT_ENABLE_VORTEX_SCAN_PUSHDOWN = true;
 
 // skipindex stats related
 const double DEFAULT_BLOOM_FILTER_FALSE_POSITIVE_RATE = 0.01;

internal/core/src/common/init_c.cpp

@@ -111,6 +111,11 @@ SetDefaultEnableParquetStatsSkipIndex(bool val) {
     milvus::SetDefaultEnableParquetStatsSkipIndex(val);
 }
 
+void
+SetDefaultVortexScanPushdownEnable(bool val) {
+    milvus::SetDefaultVortexScanPushdownEnable(val);
+}
+
 void
 SetEnableLatestDeleteSnapshotOptimization(bool val) {
     milvus::SetEnableLatestDeleteSnapshotOptimization(val);

internal/core/src/common/init_c.h

@@ -67,6 +67,9 @@ SetDefaultConfigParamTypeCheck(bool val);
 void
 SetDefaultEnableParquetStatsSkipIndex(bool val);
 
+void
+SetDefaultVortexScanPushdownEnable(bool val);
+
 void
 SetEnableLatestDeleteSnapshotOptimization(bool val);
 

internal/core/src/exec/expression/BinaryRangeExpr.cpp

@@ -483,6 +483,54 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
             processed_size = ProcessDataChunksForElementLevel<T>(
                 execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
         } else {
+            auto make_scan_value = [](const HighPrecisionType& value) {
+                proto::plan::GenericValue scan_value;
+                if constexpr (std::is_same_v<HighPrecisionType, bool>) {
+                    scan_value.set_bool_val(value);
+                } else if constexpr (std::is_integral_v<HighPrecisionType>) {
+                    scan_value.set_int64_val(static_cast<int64_t>(value));
+                } else if constexpr (std::is_floating_point_v<
+                                         HighPrecisionType>) {
+                    scan_value.set_float_val(value);
+                } else if constexpr (std::is_same_v<HighPrecisionType,
+                                                    std::string>) {
+                    scan_value.set_string_val(value);
+                }
+                return scan_value;
+            };
+
+            auto column = segment_->GetChunkedColumn(field_id_);
+            if (!row_id_scan_initialized_) {
+                row_id_scan_initialized_ = true;
+                if (column != nullptr) {
+                    auto options =
+                        ChunkedColumnInterface::ScanOptions::ForBinaryRange(
+                            current_data_global_pos_,
+                            active_count_ - current_data_global_pos_,
+                            make_scan_value(val1),
+                            lower_inclusive,
+                            make_scan_value(val2),
+                            upper_inclusive);
+                    if (column->SupportsScanPushdown(options)) {
+                        row_id_scan_cursor_ = column->Scan(op_ctx_, options);
+                        AssertInfo(row_id_scan_cursor_ != nullptr,
+                                   "row id scan cursor is null for field {}",
+                                   field_id_.get());
+                    }
+                }
+            }
+            if (row_id_scan_cursor_ != nullptr) {
+                auto bitmaps = RowIdScanToBitmaps(row_id_scan_cursor_.get(),
+                                                  buffered_scan_entries_,
+                                                  row_id_scan_batch_,
+                                                  current_data_global_pos_,
+                                                  real_batch_size,
+                                                  bitmap_input);
+                MoveCursor();
+                return std::make_shared<ColumnVector>(
+                    std::move(bitmaps.result), std::move(bitmaps.validity));
+            }
+
             processed_size = ProcessDataChunks<T>(
                 execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
         }

internal/core/src/exec/expression/BinaryRangeExpr.h

@@ -20,6 +20,7 @@
 #include <simdjson.h>
 #include <stdint.h>
 #include <cstddef>
+#include <deque>
 #include <memory>
 #include <optional>
 #include <string>
@@ -44,6 +45,7 @@
 #include "expr/ITypeExpr.h"
 #include "index/ScalarIndex.h"
 #include "index/json_stats/bson_inverted.h"
+#include "mmap/ChunkedColumnInterface.h"
 #include "pb/plan.pb.h"
 #include "segcore/SegmentInterface.h"
 #include "simdjson/error.h"
@@ -372,6 +374,11 @@ class PhyBinaryRangeFilterExpr : public SegmentExpr {
     SingleElement upper_arg_;
     bool arg_inited_{false};
     PinWrapper<index::BsonInvertedIndex*> bson_index_{nullptr};
+    bool row_id_scan_initialized_{false};
+    std::unique_ptr<ChunkedColumnInterface::ScanCursor> row_id_scan_cursor_{
+        nullptr};
+    ChunkedColumnInterface::ScanBatch row_id_scan_batch_;
+    std::deque<RowIdScanEntry> buffered_scan_entries_;
 };
 }  //namespace exec
 }  // namespace milvus

internal/core/src/exec/expression/Expr.h

@@ -19,6 +19,7 @@
 #include <algorithm>
 #include <bit>
 #include <chrono>
+#include <deque>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -121,6 +122,152 @@ ApplyValidMask(const bool* valid_data,
     }
 }
 
+struct RowIdScanBitmaps {
+    TargetBitmap result;
+    TargetBitmap validity;
+};
+
+struct RowIdScanEntry {
+    int64_t row_id;
+    bool valid;
+};
+
+inline RowIdScanBitmaps
+RowIdScanToBitmaps(ChunkedColumnInterface::ScanCursor* cursor,
+                   std::deque<RowIdScanEntry>& buffered_entries,
+                   ChunkedColumnInterface::ScanBatch& scan_batch,
+                   int64_t batch_start,
+                   int64_t batch_size,
+                   const TargetBitmap& bitmap_input,
+                   bool mask_validity_by_bitmap_input = true) {
+    AssertInfo(cursor != nullptr, "row id scan cursor is null");
+    const int64_t batch_end = batch_start + batch_size;
+    RowIdScanBitmaps bitmaps{TargetBitmap(batch_size, false),
+                             TargetBitmap(batch_size, true)};
+
+    auto apply_entry = [&](const RowIdScanEntry& entry) {
+        const auto row_id = entry.row_id;
+        if (row_id < batch_start || row_id >= batch_end) {
+            return;
+        }
+        const auto local_index = static_cast<size_t>(row_id - batch_start);
+        if (!entry.valid) {
+            if (!mask_validity_by_bitmap_input || bitmap_input.empty() ||
+                bitmap_input[local_index]) {
+                bitmaps.validity[local_index] = false;
+            }
+            return;
+        }
+        if (bitmap_input.empty() || bitmap_input[local_index]) {
+            bitmaps.result[local_index] = true;
+        }
+    };
+
+    auto buffer_scan_batch_entries = [&]() {
+        AssertInfo(scan_batch.values.empty(),
+                   "row id payload scan batch should not contain values");
+        AssertInfo(scan_batch.validity.size ==
+                       static_cast<int64_t>(scan_batch.row_ids.size()),
+                   "row id payload scan validity size {} does not match "
+                   "row ids size {}",
+                   scan_batch.validity.size,
+                   scan_batch.row_ids.size());
+        AssertInfo(
+            scan_batch.size == static_cast<int64_t>(scan_batch.row_ids.size()),
+            "row id payload scan size {} does not match row ids size {}",
+            scan_batch.size,
+            scan_batch.row_ids.size());
+        for (size_t i = 0; i < scan_batch.row_ids.size(); ++i) {
+            const auto row_id = scan_batch.row_ids[i];
+            if (!buffered_entries.empty()) {
+                AssertInfo(buffered_entries.back().row_id <= row_id,
+                           "row id payload is not ordered: {} before {}",
+                           buffered_entries.back().row_id,
+                           row_id);
+            }
+            buffered_entries.push_back(RowIdScanEntry{
+                row_id, scan_batch.validity.IsValid(static_cast<int64_t>(i))});
+        }
+    };
+
+    auto apply_all_valid_scan_batch_entries = [&]() {
+        AssertInfo(scan_batch.values.empty(),
+                   "row id payload scan batch should not contain values");
+        AssertInfo(scan_batch.validity.size ==
+                       static_cast<int64_t>(scan_batch.row_ids.size()),
+                   "row id payload scan validity size {} does not match "
+                   "row ids size {}",
+                   scan_batch.validity.size,
+                   scan_batch.row_ids.size());
+        AssertInfo(
+            scan_batch.size == static_cast<int64_t>(scan_batch.row_ids.size()),
+            "row id payload scan size {} does not match row ids size {}",
+            scan_batch.size,
+            scan_batch.row_ids.size());
+
+        std::optional<int64_t> previous_row_id;
+        size_t i = 0;
+        for (; i < scan_batch.row_ids.size(); ++i) {
+            const auto row_id = scan_batch.row_ids[i];
+            if (previous_row_id.has_value()) {
+                AssertInfo(previous_row_id.value() <= row_id,
+                           "row id payload is not ordered: {} before {}",
+                           previous_row_id.value(),
+                           row_id);
+            }
+            previous_row_id = row_id;
+            if (row_id < batch_start) {
+                continue;
+            }
+            if (row_id >= batch_end) {
+                break;
+            }
+            const auto local_index = static_cast<size_t>(row_id - batch_start);
+            if (bitmap_input.empty() || bitmap_input[local_index]) {
+                bitmaps.result[local_index] = true;
+            }
+        }
+
+        for (; i < scan_batch.row_ids.size(); ++i) {
+            const auto row_id = scan_batch.row_ids[i];
+            if (previous_row_id.has_value()) {
+                AssertInfo(previous_row_id.value() <= row_id,
+                           "row id payload is not ordered: {} before {}",
+                           previous_row_id.value(),
+                           row_id);
+            }
+            previous_row_id = row_id;
+            buffered_entries.push_back(RowIdScanEntry{row_id, true});
+        }
+        return !buffered_entries.empty();
+    };
+
+    while (true) {
+        while (!buffered_entries.empty()) {
+            const auto entry = buffered_entries.front();
+            if (entry.row_id >= batch_end) {
+                return bitmaps;
+            }
+            buffered_entries.pop_front();
+            apply_entry(entry);
+        }
+        if (!cursor->Next(&scan_batch)) {
+            break;
+        }
+        AssertInfo(scan_batch.size > 0, "invalid row id scan batch");
+        if (scan_batch.validity.encoding ==
+                ChunkedColumnInterface::ValidityEncoding::AllValid ||
+            scan_batch.validity.all_valid) {
+            if (apply_all_valid_scan_batch_entries()) {
+                return bitmaps;
+            }
+            continue;
+        }
+        buffer_scan_batch_entries();
+    }
+    return bitmaps;
+}
+
 class Expr : public std::enable_shared_from_this<Expr> {
  public:
     Expr(DataType type,
@@ -494,7 +641,10 @@ class SegmentExpr : public Expr {
 
     void
     PrefetchRawDataChunksForScanIfNeeded(const ChunkedColumnInterface* column) {
-        if (column == nullptr || prefetched_ || !segment_->is_chunked()) {
+        if (column == nullptr ||
+            column->GetLocalFormat() !=
+                ChunkedColumnInterface::LocalFormat::Raw ||
+            prefetched_ || !segment_->is_chunked()) {
             return;
         }
         std::vector<int64_t> chunk_ids;
@@ -1085,7 +1235,7 @@ class SegmentExpr : public Expr {
     }
 
     // accept sorted offsets array and process with one continuous Scan.
-    // TODO: push the offset selection into Scan when the interface supports it.
+    // TODO: push the offset bitmap down into Scan when Vortex supports bitmap scan.
     template <typename T, typename FUNC, typename... ValTypes>
     int64_t
     ProcessSortedDataByOffsetsByScan(

internal/core/src/exec/expression/UnaryExpr.cpp

@@ -1654,6 +1654,38 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
         return nullptr;
     }
 
+    if (!has_offset_input_ && !expr_->column_.element_level_) {
+        auto column = segment_->GetChunkedColumn(field_id_);
+        if (!row_id_scan_initialized_) {
+            row_id_scan_initialized_ = true;
+            if (column != nullptr) {
+                auto options = ChunkedColumnInterface::ScanOptions::ForUnary(
+                    current_data_global_pos_,
+                    active_count_ - current_data_global_pos_,
+                    expr_->op_type_,
+                    expr_->val_);
+                if (column->SupportsScanPushdown(options)) {
+                    row_id_scan_cursor_ = column->Scan(op_ctx_, options);
+                    AssertInfo(row_id_scan_cursor_ != nullptr,
+                               "row id scan cursor is null for field {}",
+                               field_id_.get());
+                }
+            }
+        }
+        if (row_id_scan_cursor_ != nullptr) {
+            auto bitmaps = RowIdScanToBitmaps(row_id_scan_cursor_.get(),
+                                              buffered_scan_entries_,
+                                              row_id_scan_batch_,
+                                              current_data_global_pos_,
+                                              real_batch_size,
+                                              bitmap_input,
+                                              true);
+            MoveCursor();
+            return std::make_shared<ColumnVector>(std::move(bitmaps.result),
+                                                  std::move(bitmaps.validity));
+        }
+    }
+
     if (!arg_inited_) {
         value_arg_.SetValue<IndexInnerType>(expr_->val_);
         arg_inited_ = true;

internal/core/src/exec/expression/UnaryExpr.h

@@ -19,6 +19,7 @@
 #include <fmt/core.h>
 #include <folly/Unit.h>
 
+#include <deque>
 #include <optional>
 #include <utility>
 
@@ -39,6 +40,7 @@
 #include "index/json_stats/bson_inverted.h"
 #include "cachinglayer/CacheSlot.h"
 #include "index/NgramInvertedIndex.h"
+#include "mmap/ChunkedColumnInterface.h"
 
 namespace milvus {
 namespace exec {
@@ -1216,6 +1218,12 @@ class PhyUnaryRangeFilterExpr : public SegmentExpr {
         auto pattern = GetValueFromProto<std::string>(expr_->val_);
         cached_like_matcher_ = std::make_unique<LikePatternMatcher>(pattern);
     }
+
+    bool row_id_scan_initialized_{false};
+    std::unique_ptr<ChunkedColumnInterface::ScanCursor> row_id_scan_cursor_{
+        nullptr};
+    ChunkedColumnInterface::ScanBatch row_id_scan_batch_;
+    std::deque<RowIdScanEntry> buffered_scan_entries_;
 };
 }  // namespace exec
 }  // namespace milvus

internal/core/src/mmap/VortexColumn.cpp

@@ -1505,6 +1505,10 @@ VortexColumn::BuildVortexPredicate(const ScanOptions& options) const {
 
 bool
 VortexColumn::SupportsScanPushdown(const ScanOptions& options) const {
+    if (!ENABLE_VORTEX_SCAN_PUSHDOWN.load()) {
+        return false;
+    }
+
     switch (data_type_) {
         case DataType::STRING:
         case DataType::VARCHAR:

internal/core/src/mmap/VortexColumnTest.cpp

@@ -202,6 +202,21 @@ IsVortexStringPushdownType(DataType type) {
     return type == DataType::STRING || type == DataType::VARCHAR;
 }
 
+class ScopedVortexScanPushdownEnable {
+ public:
+    explicit ScopedVortexScanPushdownEnable(bool enable)
+        : old_(ENABLE_VORTEX_SCAN_PUSHDOWN.load()) {
+        SetDefaultVortexScanPushdownEnable(enable);
+    }
+
+    ~ScopedVortexScanPushdownEnable() {
+        SetDefaultVortexScanPushdownEnable(old_);
+    }
+
+ private:
+    bool old_;
+};
+
 void
 CheckNullableFilteredScanReturnsValidity(VortexColumn& column, DataType type) {
     const auto value = StringValue(ExpectedString(type, 8));
@@ -978,6 +993,14 @@ TEST(VortexColumnTest, MultiFieldColumnsShareColumnGroup) {
     auto string_filter_options = ChunkedColumnInterface::ScanOptions::ForUnary(
         0, 16, proto::plan::OpType::Equal, StringValue("v4"));
     EXPECT_TRUE(string_column.SupportsScanPushdown(string_filter_options));
+    {
+        ScopedVortexScanPushdownEnable disable_pushdown(false);
+        EXPECT_FALSE(string_column.SupportsScanPushdown(string_filter_options));
+        EXPECT_THROW(
+            CollectFilteredRowIdPayload(string_column, string_filter_options),
+            std::exception);
+    }
+    EXPECT_TRUE(string_column.SupportsScanPushdown(string_filter_options));
 
     auto filter_options = ChunkedColumnInterface::ScanOptions::ForBinaryRange(
         0, 16, IntValue(30), true, IntValue(60), true);

internal/util/initcore/init_core.go

@@ -654,6 +654,15 @@ func SetupCoreConfigChangelCallback() {
 			return nil
 		})
 
+		paramtable.Get().QueryNodeCfg.EnableVortexScanPushdown.RegisterCallback(func(ctx context.Context, key, oldValue, newValue string) error {
+			enable, err := strconv.ParseBool(newValue)
+			if err != nil {
+				return err
+			}
+			UpdateDefaultVortexScanPushdownEnable(enable)
+			return nil
+		})
+
 		paramtable.Get().QueryNodeCfg.EnableLatestDeleteSnapshotOptimization.RegisterCallback(func(ctx context.Context, key, oldValue, newValue string) error {
 			enable, err := strconv.ParseBool(newValue)
 			if err != nil {

internal/util/initcore/query_node.go

@@ -134,6 +134,9 @@ func doInitQueryNodeOnce(ctx context.Context) error {
 	cDeleteDumpBatchSize := C.int64_t(paramtable.Get().QueryNodeCfg.DeleteDumpBatchSize.GetAsInt64())
 	C.SetDefaultDeleteDumpBatchSize(cDeleteDumpBatchSize)
 
+	cVortexScanPushdownEnabled := C.bool(paramtable.Get().QueryNodeCfg.EnableVortexScanPushdown.GetAsBool())
+	C.SetDefaultVortexScanPushdownEnable(cVortexScanPushdownEnabled)
+
 	cEnableLatestDeleteSnapshotOptimization := C.bool(paramtable.Get().QueryNodeCfg.EnableLatestDeleteSnapshotOptimization.GetAsBool())
 	C.SetEnableLatestDeleteSnapshotOptimization(cEnableLatestDeleteSnapshotOptimization)
 

internal/util/initcore/util.go

@@ -79,6 +79,10 @@ func UpdateDefaultDeleteDumpBatchSize(size int) {
 	C.SetDefaultDeleteDumpBatchSize(C.int64_t(size))
 }
 
+func UpdateDefaultVortexScanPushdownEnable(enable bool) {
+	C.SetDefaultVortexScanPushdownEnable(C.bool(enable))
+}
+
 func UpdateDefaultOptimizeExprEnable(enable bool) {
 	C.SetDefaultOptimizeExprEnable(C.bool(enable))
 }

pkg/util/paramtable/component_param.go

@@ -3683,6 +3683,8 @@ type queryNodeConfig struct {
 	// are packed into cells so rgs_per_cell * avg_rg_size ≈ this value.
 	StorageV2CellTargetSizeBytes ParamItem `refreshable:"true"`
 
+	EnableVortexScanPushdown ParamItem `refreshable:"true"`
+
 	EnableWorkerSQCostMetrics ParamItem `refreshable:"true"`
 
 	ExprEvalBatchSize ParamItem `refreshable:"false"`
@@ -4909,6 +4911,15 @@ user-task-polling:
 	}
 	p.StorageV2CellTargetSizeBytes.Init(base.mgr)
 
+	p.EnableVortexScanPushdown = ParamItem{
+		Key:          "queryNode.segcore.enableVortexScanPushdown",
+		Version:      "3.0.0",
+		DefaultValue: "true",
+		Doc:          "Enable Vortex local-format varchar filter pushdown through row-id scan.",
+		Export:       true,
+	}
+	p.EnableVortexScanPushdown.Init(base.mgr)
+
 	p.EnableWorkerSQCostMetrics = ParamItem{
 		Key:          "queryNode.enableWorkerSQCostMetrics",
 		Version:      "2.3.0",