enhance: cache JSON flat validity bitmaps (#51291)

## Summary

- Reuse the existing `ExprResCacheManager/ExprCacheHelper` from the
expression layer.
- Cache JSON-flat exact-path validity bitmaps by segment + field/full
JSON pointer + `Any|Numeric|String|Bool` canonical key.
- Exclude literal/operator from the key so validity can be reused across
predicates.
- Do not add a `JsonFlatIndex`-owned LRU or change the index format.

This is based on the latest `master`, including merged PR #51235.

## Verification

Verification is pending and will continue after the PR is created.

issue: #51234

Signed-off-by: Buqian Zheng <zhengbuqian@gmail.com>

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

@@ -1790,12 +1790,14 @@ class SegmentExpr : public Expr {
             PinWrapper<const index::IndexBase*> json_pw;
             std::shared_ptr<index::JsonFlatIndexQueryExecutor<IndexInnerType>>
                 executor;
+            const auto json_pointer = field_type_ == DataType::JSON
+                                          ? milvus::Json::pointer(nested_path_)
+                                          : std::string();
             auto prepare_index = [&]() {
                 if (index_ptr != nullptr) {
                     return;
                 }
                 if (field_type_ == DataType::JSON) {
-                    auto pointer = milvus::Json::pointer(nested_path_);
                     json_pw = pinned_index_[0];
                     auto json_flat_index =
                         dynamic_cast<const index::JsonFlatIndex*>(
@@ -1806,7 +1808,7 @@ class SegmentExpr : public Expr {
                         executor =
                             json_flat_index
                                 ->template create_executor<IndexInnerType>(
-                                    pointer.substr(index_path.size()));
+                                    json_pointer.substr(index_path.size()));
                         index_ptr = executor.get();
                     } else {
                         auto json_index =
@@ -1831,9 +1833,45 @@ class SegmentExpr : public Expr {
                     TargetBitmap res = func(index_ptr, values...);
 
                     TargetBitmap valid_res;
-                    if (validity_mode == IndexValidityMode::JsonExactPath &&
-                        executor != nullptr) {
-                        valid_res = executor->ExactPathExists();
+                    std::optional<index::JsonValueType> json_value_type;
+                    if (executor != nullptr) {
+                        if (validity_mode == IndexValidityMode::JsonExactPath) {
+                            json_value_type = index::JsonValueType::Any;
+                        } else if constexpr (std::is_same_v<IndexInnerType,
+                                                            bool>) {
+                            json_value_type = index::JsonValueType::Bool;
+                        } else if constexpr (std::is_integral_v<
+                                                 IndexInnerType> ||
+                                             std::is_floating_point_v<
+                                                 IndexInnerType>) {
+                            json_value_type = index::JsonValueType::Numeric;
+                        } else if constexpr (std::is_same_v<IndexInnerType,
+                                                            std::string>) {
+                            json_value_type = index::JsonValueType::String;
+                        }
+                    }
+
+                    if (json_value_type.has_value()) {
+                        const auto family =
+                            static_cast<unsigned int>(json_value_type.value());
+                        const auto signature = fmt::format(
+                            "json-flat-validity:v1:field={}:path-length={}:"
+                            "path={}:family={}",
+                            field_id_.get(),
+                            json_pointer.size(),
+                            json_pointer,
+                            family);
+                        if (ExprResCacheManager::IsEnabled()) {
+                            auto validity = ExprCacheHelper::GetOrComputeBitmap(
+                                segment_, signature, active_count_, [&]() {
+                                    return executor->ExactPathExists(
+                                        json_value_type.value());
+                                });
+                            valid_res = std::move(*validity);
+                        } else {
+                            valid_res = executor->ExactPathExists(
+                                json_value_type.value());
+                        }
                     } else if (cached_is_nested_index_ &&
                                func_returns_row_level) {
                         valid_res = TargetBitmap(active_count_, true);

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

@@ -139,6 +139,30 @@ class ExprCacheHelper {
 
         return {result, valid};
     }
+
+    // Cache one immutable bitmap artifact using the same backend, admission,
+    // staleness, and segment-invalidation rules as full expression results.
+    // The all-ones companion satisfies the existing two-bitmap cache value
+    // contract and is compressed efficiently by the memory backend.
+    template <typename ComputeFn>
+    static std::shared_ptr<TargetBitmap>
+    GetOrComputeBitmap(const segcore::SegmentInternalInterface* segment,
+                       const std::string& artifact_signature,
+                       int64_t active_count,
+                       ComputeFn&& compute,
+                       bool enable_cache_write = true) {
+        auto cached = GetOrCompute(
+            segment,
+            artifact_signature,
+            active_count,
+            [&]() -> ComputeResult {
+                auto result = compute();
+                TargetBitmap valid(result.size(), true);
+                return {std::move(result), std::move(valid)};
+            },
+            enable_cache_write);
+        return cached.result;
+    }
 };
 
 // ============================================================

internal/core/src/index/JsonFlatIndexTest.cpp

@@ -35,6 +35,7 @@
 #include "common/Tracer.h"
 #include "common/Types.h"
 #include "common/protobuf_utils.h"
+#include "exec/expression/ExprCache.h"
 #include "expr/ITypeExpr.h"
 #include "filemanager/InputStream.h"
 #include "gtest/gtest.h"
@@ -837,6 +838,10 @@ class JsonFlatIndexExprTest : public ::testing::Test {
  protected:
     void
     SetUp() override {
+        auto& cache = exec::ExprResCacheManager::Instance();
+        cache.Clear();
+        exec::ExprResCacheManager::SetEnabled(false);
+
         json_data_ = {
             R"({"a": 1.0})",
             R"({"a": "abc"})",
@@ -919,6 +924,9 @@ class JsonFlatIndexExprTest : public ::testing::Test {
 
     void
     TearDown() override {
+        auto& cache = exec::ExprResCacheManager::Instance();
+        cache.Clear();
+        exec::ExprResCacheManager::SetEnabled(false);
     }
 
     FieldId json_fid_;
@@ -931,6 +939,10 @@ class JsonFlatIndexContainsExprTest : public ::testing::Test {
  protected:
     void
     SetUp() override {
+        auto& cache = exec::ExprResCacheManager::Instance();
+        cache.Clear();
+        exec::ExprResCacheManager::SetEnabled(false);
+
         json_data_ = {
             R"({"a": [1, 2]})",
             R"({"a": [2]})",
@@ -1008,6 +1020,13 @@ class JsonFlatIndexContainsExprTest : public ::testing::Test {
         segment_->LoadFieldData(load_info);
     }
 
+    void
+    TearDown() override {
+        auto& cache = exec::ExprResCacheManager::Instance();
+        cache.Clear();
+        exec::ExprResCacheManager::SetEnabled(false);
+    }
+
     ColumnVectorPtr
     Evaluate(proto::plan::JSONContainsExpr_JSONOp op,
              std::vector<int64_t> values,
@@ -1176,6 +1195,28 @@ TEST_F(JsonFlatIndexContainsExprTest, PreservesLargeInt64LiteralPrecision) {
                  true});
 }
 
+TEST_F(JsonFlatIndexContainsExprTest,
+       ReusesExactPathValidityAcrossLiteralsAndOperators) {
+    auto& cache = exec::ExprResCacheManager::Instance();
+    exec::CacheConfig config;
+    config.mode = exec::CacheMode::Memory;
+    config.mem_max_bytes = 1 << 20;
+    config.compression_enabled = true;
+    config.admission_threshold = 1;
+    config.mem_min_eval_duration_us = 0;
+    ASSERT_TRUE(cache.SetConfig(config));
+    exec::ExprResCacheManager::SetEnabled(true);
+
+    Evaluate(proto::plan::JSONContainsExpr_JSONOp_Contains, {1});
+    EXPECT_EQ(cache.GetEntryCount(), 2);
+
+    Evaluate(proto::plan::JSONContainsExpr_JSONOp_Contains, {2});
+    EXPECT_EQ(cache.GetEntryCount(), 3);
+
+    Evaluate(proto::plan::JSONContainsExpr_JSONOp_ContainsAny, {1, 3});
+    EXPECT_EQ(cache.GetEntryCount(), 4);
+}
+
 TEST_F(JsonFlatIndexExprTest, TestUnaryExpr) {
     proto::plan::GenericValue value;
     value.set_int64_val(1);
@@ -1228,6 +1269,77 @@ TEST_F(JsonFlatIndexExprTest, TestComparisonUnknowns) {
     EXPECT_TRUE(final[13]);
 }
 
+TEST_F(JsonFlatIndexExprTest, ReusesValidityAcrossLiteralsAndOperators) {
+    auto& cache = exec::ExprResCacheManager::Instance();
+    exec::CacheConfig config;
+    config.mode = exec::CacheMode::Memory;
+    config.mem_max_bytes = 1 << 20;
+    config.compression_enabled = true;
+    config.admission_threshold = 1;
+    config.mem_min_eval_duration_us = 0;
+    ASSERT_TRUE(cache.SetConfig(config));
+    exec::ExprResCacheManager::SetEnabled(true);
+
+    auto evaluate = [&](std::vector<std::string> nested_path,
+                        proto::plan::OpType op,
+                        proto::plan::GenericValue value) {
+        auto unary_expr = std::make_shared<expr::UnaryRangeFilterExpr>(
+            expr::ColumnInfo(json_fid_, DataType::JSON, std::move(nested_path)),
+            op,
+            value,
+            std::vector<proto::plan::GenericValue>());
+        auto plan = std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
+                                                           unary_expr);
+        return query::ExecuteQueryExpr(
+            plan, segment_.get(), json_data_.size(), MAX_TIMESTAMP);
+    };
+    auto int_value = [](int64_t literal) {
+        proto::plan::GenericValue value;
+        value.set_int64_val(literal);
+        return value;
+    };
+
+    EXPECT_EQ(evaluate({"a"}, proto::plan::OpType::Equal, int_value(1)).count(),
+              2);
+    EXPECT_EQ(cache.GetEntryCount(), 2);
+
+    exec::ExprResCacheManager::Key artifact_key{
+        segment_->get_segment_id(),
+        fmt::format("json-flat-validity:v1:field={}:path-length=2:"
+                    "path=/a:family={}",
+                    json_fid_.get(),
+                    static_cast<unsigned int>(index::JsonValueType::Numeric))};
+    exec::ExprResCacheManager::Value artifact;
+    artifact.active_count = json_data_.size();
+    ASSERT_TRUE(cache.Get(artifact_key, artifact));
+    EXPECT_EQ(artifact.result->count(), 6);
+    EXPECT_EQ(artifact.valid_result->count(), json_data_.size());
+    artifact.active_count = json_data_.size() + 1;
+    EXPECT_FALSE(cache.Get(artifact_key, artifact));
+
+    EXPECT_EQ(evaluate({"a"}, proto::plan::OpType::Equal, int_value(3)).count(),
+              1);
+    EXPECT_EQ(cache.GetEntryCount(), 3);
+
+    EXPECT_EQ(
+        evaluate({"a"}, proto::plan::OpType::GreaterThan, int_value(1)).count(),
+        4);
+    EXPECT_EQ(cache.GetEntryCount(), 4);
+
+    proto::plan::GenericValue string_value;
+    string_value.set_string_val("abc");
+    EXPECT_EQ(evaluate({"a"}, proto::plan::OpType::Equal, string_value).count(),
+              1);
+    EXPECT_EQ(cache.GetEntryCount(), 6);
+
+    EXPECT_EQ(evaluate({"b"}, proto::plan::OpType::Equal, int_value(2)).count(),
+              1);
+    EXPECT_EQ(cache.GetEntryCount(), 8);
+
+    EXPECT_EQ(cache.EraseSegment(segment_->get_segment_id()), 8);
+    EXPECT_EQ(cache.GetEntryCount(), 0);
+}
+
 TEST_F(JsonFlatIndexExprTest, PreservesLargeInt64LiteralPrecision) {
     proto::plan::GenericValue value;
     value.set_int64_val(9007199254740993LL);