furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: [cp2.6]preserve unknown semantics for json path predicates (#50723)

Browse Source 
issue: #50699
pr: #50702

## What this PR does

This is the 2.6 backport of #50702. It makes JSON path predicate
missing/null/type-mismatch behavior preserve UNKNOWN consistently:

- Treats parent JSON NULL, missing nested paths, incompatible path
types, and invalid array paths as UNKNOWN instead of operator-specific
boolean constants.
- Preserves JSON stats/index validity bitmaps when returning cached
bitmap results.
- Updates parser rewrite behavior and tests that previously assumed
scalar JSON missing-path `!=` was a definite true result.
- Aligns raw, stats/index, brute-force, offset/iterative, and `NOT`
behavior around SQL-style three-valued logic.

## Verification

Validation reported for this 2.6 cherry-pick:

- `git diff --check` passed.
- `go test -buildvcs=false -count=1
./internal/parser/planparserv2/rewriter` passed.

C++ validation was blocked in the local environment because
`cmake_build` attempted to regenerate with Conan-1-style
`conanbuildinfo.cmake`, while local Conan output was Conan 2 and did not
include that file; syntax-only compile also hit missing `bsoncxx`
headers in the local dependency cache.

---------

Signed-off-by: Buqian Zheng <zhengbuqian@gmail.com>
This commit is contained in:
Buqian Zheng
2026-07-01 14:48:29 +08:00
committed by GitHub
parent 1d8fb77560
commit 936b0f74ac
12 changed files with 678 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/src/common/Json.h
+8
View File
@@ -257,6 +257,14 @@ class Json {
: doc().at_pointer(pointer).get_number_type();
}
value_result<simdjson::ondemand::number>
at_numeric(std::string_view pointer) const {
if (pointer.empty()) {
return doc().get_number();
}
return doc().at_pointer(pointer).get_number();
}
template <typename T>
value_result<T>
at(std::string_view pointer) const {
internal/core/src/exec/expression/BinaryRangeExpr.cpp
+61 -85
View File
@@ -622,18 +622,20 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
// process shredding data
auto try_execute = [&](milvus::index::JSONType json_type,
TargetBitmapView& res_view,
TargetBitmapView& valid_res_view,
auto GetType) {
auto target_field = index->GetShreddingField(pointer, json_type);
if (!target_field.empty()) {
using ColType = decltype(GetType);
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
auto shredding_executor =
[val1, val2, lower_inclusive, upper_inclusive](
const ColType* src,
@@ -661,8 +663,11 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonStats() {
target_field,
shredding_executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
LOG_DEBUG("using shredding data's field: {} count {}",
target_field,
res_view.count());
@@ -676,94 +681,65 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonStats() {
if constexpr (std::is_same_v<GetType, int64_t>) {
// int64 compare
try_execute(milvus::index::JSONType::INT64,
res_view,
valid_res_view,
int64_t{});
try_execute(milvus::index::JSONType::INT64, int64_t{});
// and double compare
TargetBitmap res_double(active_count_, false);
TargetBitmapView res_double_view(res_double);
TargetBitmap res_double_valid(active_count_, true);
TargetBitmapView valid_res_double_view(res_double_valid);
try_execute(milvus::index::JSONType::DOUBLE,
res_double_view,
valid_res_double_view,
double{});
res_view.inplace_or_with_count(res_double_view, active_count_);
valid_res_view.inplace_or_with_count(valid_res_double_view,
active_count_);
try_execute(milvus::index::JSONType::DOUBLE, double{});
} else if constexpr (std::is_same_v<GetType, double>) {
try_execute(milvus::index::JSONType::DOUBLE,
res_view,
valid_res_view,
double{});
try_execute(milvus::index::JSONType::DOUBLE, double{});
// and int64 compare
TargetBitmap res_int64(active_count_, false);
TargetBitmapView res_int64_view(res_int64);
TargetBitmap res_int64_valid(active_count_, true);
TargetBitmapView valid_res_int64_view(res_int64_valid);
try_execute(milvus::index::JSONType::INT64,
res_int64_view,
valid_res_int64_view,
int64_t{});
res_view.inplace_or_with_count(res_int64_view, active_count_);
valid_res_view.inplace_or_with_count(valid_res_int64_view,
active_count_);
try_execute(milvus::index::JSONType::INT64, int64_t{});
} else if constexpr (std::is_same_v<GetType, std::string_view> ||
std::is_same_v<GetType, std::string>) {
try_execute(milvus::index::JSONType::STRING,
res_view,
valid_res_view,
std::string_view{});
}
}
// process shared data
auto shared_executor =
[val1, val2, lower_inclusive, upper_inclusive, &res_view](
milvus::BsonView bson, uint32_t row_id, uint32_t value_offset) {
if constexpr (std::is_same_v<GetType, int64_t> ||
std::is_same_v<GetType, double>) {
auto val = bson.ParseAsValueAtOffset<double>(value_offset);
if (!val.has_value()) {
res_view[row_id] = false;
return;
}
if (lower_inclusive && upper_inclusive) {
res_view[row_id] =
val.value() >= val1 && val.value() <= val2;
} else if (lower_inclusive && !upper_inclusive) {
res_view[row_id] =
val.value() >= val1 && val.value() < val2;
} else if (!lower_inclusive && upper_inclusive) {
res_view[row_id] =
val.value() > val1 && val.value() <= val2;
} else {
res_view[row_id] =
val.value() > val1 && val.value() < val2;
}
} else {
auto val = bson.ParseAsValueAtOffset<GetType>(value_offset);
if (!val.has_value()) {
res_view[row_id] = false;
return;
}
if (lower_inclusive && upper_inclusive) {
res_view[row_id] =
val.value() >= val1 && val.value() <= val2;
} else if (lower_inclusive && !upper_inclusive) {
res_view[row_id] =
val.value() >= val1 && val.value() < val2;
} else if (!lower_inclusive && upper_inclusive) {
res_view[row_id] =
val.value() > val1 && val.value() <= val2;
} else {
res_view[row_id] =
val.value() > val1 && val.value() < val2;
}
}
auto shared_executor = [val1,
val2,
lower_inclusive,
upper_inclusive,
&res_view,
&valid_res_view](milvus::BsonView bson,
uint32_t row_id,
uint32_t value_offset) {
auto set_known = [&](bool value) {
res_view[row_id] = value;
valid_res_view[row_id] = true;
};
if constexpr (std::is_same_v<GetType, int64_t> ||
std::is_same_v<GetType, double>) {
auto val = bson.ParseAsValueAtOffset<double>(value_offset);
if (!val.has_value()) {
return;
}
if (lower_inclusive && upper_inclusive) {
set_known(val.value() >= val1 && val.value() <= val2);
} else if (lower_inclusive && !upper_inclusive) {
set_known(val.value() >= val1 && val.value() < val2);
} else if (!lower_inclusive && upper_inclusive) {
set_known(val.value() > val1 && val.value() <= val2);
} else {
set_known(val.value() > val1 && val.value() < val2);
}
} else {
auto val = bson.ParseAsValueAtOffset<GetType>(value_offset);
if (!val.has_value()) {
return;
}
if (lower_inclusive && upper_inclusive) {
set_known(val.value() >= val1 && val.value() <= val2);
} else if (lower_inclusive && !upper_inclusive) {
set_known(val.value() >= val1 && val.value() < val2);
} else if (!lower_inclusive && upper_inclusive) {
set_known(val.value() > val1 && val.value() <= val2);
} else {
set_known(val.value() > val1 && val.value() < val2);
}
}
};
{
milvus::ScopedTimer timer(
"binary_range_json_stats_shared_data",
@@ -774,12 +750,12 @@ PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
} // namespace exec
template <typename ValueType>
internal/core/src/exec/expression/BinaryRangeExpr.h
+39 -24
View File
@@ -85,30 +85,45 @@ struct BinaryRangeElementFunc {
}
};
#define BinaryRangeJSONCompare(cmp) \
do { \
if (valid_data != nullptr && !valid_data[offset]) { \
res[i] = valid_res[i] = false; \
break; \
} \
if (has_bitmap_input && !bitmap_input[i + start_cursor]) { \
break; \
} \
auto x = src[offset].template at<GetType>(pointer); \
if (x.error()) { \
if constexpr (std::is_same_v<GetType, int64_t>) { \
auto x = src[offset].template at<double>(pointer); \
if (!x.error()) { \
auto value = x.value(); \
res[i] = (cmp); \
break; \
} \
} \
res[i] = false; \
break; \
} \
auto value = x.value(); \
res[i] = (cmp); \
// For int64_t GetType, uses at_numeric() (get_number()) to extract any JSON
// number in a single parse. Branches on actual type to preserve int64
// precision; uint64 and double values fall back to double comparison,
// consistent with the Tantivy index and JSON-stats paths.
// 'cmp' must reference 'value' (int64_t or double depending on the JSON value).
#define BinaryRangeJSONCompare(cmp) \
do { \
if (valid_data != nullptr && !valid_data[offset]) { \
res[i] = valid_res[i] = false; \
break; \
} \
if (has_bitmap_input && !bitmap_input[i + start_cursor]) { \
break; \
} \
if constexpr (std::is_same_v<GetType, int64_t>) { \
auto x = src[offset].at_numeric(pointer); \
if (x.error()) { \
res[i] = valid_res[i] = false; \
break; \
} \
auto n = x.value(); \
if (n.is_int64()) { \
auto value = n.get_int64(); \
res[i] = (cmp); \
} else { \
auto value = n.is_uint64() \
? static_cast<double>(n.get_uint64()) \
: n.get_double(); \
res[i] = (cmp); \
} \
} else { \
auto x = src[offset].template at<GetType>(pointer); \
if (x.error()) { \
res[i] = valid_res[i] = false; \
break; \
} \
auto value = x.value(); \
res[i] = (cmp); \
} \
} while (false)
template <typename ValueType,
internal/core/src/exec/expression/Expr.h
+13
View File
@@ -1528,6 +1528,19 @@ class SegmentExpr : public Expr {
return false;
};
VectorPtr
MoveOrSliceBitmap(TargetBitmap& cached_res,
TargetBitmap& cached_valid_res,
int64_t pos,
int64_t size) {
TargetBitmap result;
TargetBitmap valid_result;
result.append(cached_res, pos, size);
valid_result.append(cached_valid_res, pos, size);
return std::make_shared<ColumnVector>(std::move(result),
std::move(valid_result));
}
protected:
const segcore::SegmentInternalInterface* segment_;
const FieldId field_id_;
internal/core/src/exec/expression/JsonContainsByStatsTest.cpp
+10 -8
View File
@@ -263,7 +263,7 @@ TEST(JsonContainsByStatsTest, BasicContainsAnyOnArray) {
}
}
TEST(JsonStatsUnaryRangeTest, NotEqualKeepsJsonPathErrorsButMasksFieldNull) {
TEST(JsonStatsUnaryRangeTest, NotEqualKeepsJsonPathUnknownsAndMasksFieldNull) {
auto schema = std::make_shared<Schema>();
auto json_fid = schema->AddDebugField("json", DataType::JSON, true);
@@ -272,10 +272,10 @@ TEST(JsonStatsUnaryRangeTest, NotEqualKeepsJsonPathErrorsButMasksFieldNull) {
std::vector<std::string> json_raw_data = {
R"({"a": "1"})", // equal, filtered out
R"({"a": "123"})", // string mismatch, kept
R"({"a": 1})", // type mismatch for string compare, kept
R"({"b": 1})", // path missing, kept
R"({"a": null})", // JSON path error, kept
R"({})", // path missing, kept
R"({"a": 1})", // type mismatch for string compare, UNKNOWN
R"({"b": 1})", // path missing, UNKNOWN
R"({"a": null})", // JSON path null, UNKNOWN
R"({})", // path missing, UNKNOWN
R"({"a": "321"})", // string mismatch, kept
R"({"a": "123"})", // field-level null, filtered out by valid data
};
@@ -325,9 +325,11 @@ TEST(JsonStatsUnaryRangeTest, NotEqualKeepsJsonPathErrorsButMasksFieldNull) {
ASSERT_EQ(result.size(), json_raw_data.size());
EXPECT_FALSE(result[0]);
for (int i = 1; i <= 6; ++i) {
EXPECT_TRUE(result[i]) << "row " << i;
EXPECT_TRUE(result[1]);
for (int i = 2; i <= 5; ++i) {
EXPECT_FALSE(result[i]) << "row " << i;
}
EXPECT_TRUE(result[6]);
EXPECT_FALSE(result[7]);
EXPECT_EQ(result.count(), 6);
EXPECT_EQ(result.count(), 2);
}
internal/core/src/exec/expression/JsonContainsExpr.cpp
+387 -152
View File
@@ -15,14 +15,93 @@
// limitations under the License.
#include "JsonContainsExpr.h"
#include <algorithm>
#include <cmath>
#include <cstdint>
#include <unordered_map>
#include <utility>
#include <vector>
#include "common/ScopedTimer.h"
#include "common/Types.h"
namespace milvus {
namespace exec {
template <typename T>
class ContainsAllMatcher {
public:
explicit ContainsAllMatcher(const std::set<T>& targets) {
target_count_ = targets.size();
use_small_ = target_count_ <= 64;
uint32_t idx = 0;
for (const auto& target : targets) {
value_to_bit_[target] = idx++;
}
if (use_small_) {
full_mask_ = target_count_ == 64
? ~uint64_t(0)
: (uint64_t(1) << target_count_) - 1;
} else {
num_words_ = (target_count_ + 63) / 64;
}
}
bool
set_if_found(const T& value, uint64_t& found) const {
auto it = value_to_bit_.find(value);
if (it == value_to_bit_.end()) {
return false;
}
found |= uint64_t(1) << it->second;
return found == full_mask_;
}
bool
set_if_found(const T& value,
std::vector<uint64_t>& found,
size_t& remaining) const {
auto it = value_to_bit_.find(value);
if (it == value_to_bit_.end()) {
return false;
}
auto idx = it->second;
auto bit = uint64_t(1) << (idx % 64);
auto& word = found[idx / 64];
if ((word & bit) != 0) {
return false;
}
word |= bit;
return --remaining == 0;
}
bool
use_small() const {
return use_small_;
}
size_t
target_count() const {
return target_count_;
}
uint64_t
full_mask() const {
return full_mask_;
}
size_t
num_words() const {
return num_words_;
}
private:
std::unordered_map<T, uint32_t> value_to_bit_;
size_t target_count_{0};
bool use_small_{true};
uint64_t full_mask_{0};
size_t num_words_{0};
};
void
PhyJsonContainsFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
tracer::AutoSpan span(
@@ -358,7 +437,7 @@ PhyJsonContainsFilterExpr::ExecJsonContains(EvalCtx& context) {
auto doc = data[i].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
for (auto&& it : array) {
auto val = it.template get<GetType>();
@@ -370,17 +449,17 @@ PhyJsonContainsFilterExpr::ExecJsonContains(EvalCtx& context) {
std::floor(double_val.value())) {
if (elements->In(static_cast<int64_t>(
double_val.value())) > 0) {
return true;
return std::make_pair(true, true);
}
}
}
continue;
}
if (elements->In(val.value()) > 0) {
return true;
return std::make_pair(true, true);
}
}
return false;
return std::make_pair(true, false);
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -395,7 +474,12 @@ PhyJsonContainsFilterExpr::ExecJsonContains(EvalCtx& context) {
if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -473,7 +557,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
// process shredding data for ARRAY type (non-shared)
@@ -484,6 +568,10 @@ PhyJsonContainsFilterExpr::ExecJsonContainsByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsAnyExecutor<GetType> executor(
arg_set_, arg_set_double_);
@@ -492,20 +580,24 @@ PhyJsonContainsFilterExpr::ExecJsonContainsByStats() {
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
// process shared data
auto shared_executor = [this, &res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto shared_executor = [this, &res_view, &valid_res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto val = bson.ParseAsArrayAtOffset(value_offset);
if (!val.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
for (const auto& element : val.value()) {
if constexpr (std::is_same_v<GetType, int64_t> ||
@@ -539,12 +631,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
VectorPtr
@@ -607,11 +699,11 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArray(EvalCtx& context) {
processed_cursor += size;
return;
}
auto executor = [&](size_t i) -> bool {
auto executor = [&](size_t i) {
auto doc = data[i].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
for (auto&& it : array) {
auto val = it.get_array();
@@ -627,11 +719,11 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArray(EvalCtx& context) {
}
for (auto const& element : elements) {
if (CompareTwoJsonArray(json_array, element)) {
return true;
return std::make_pair(true, true);
}
}
}
return false;
return std::make_pair(true, false);
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -646,7 +738,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArray(EvalCtx& context) {
if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -705,7 +802,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArrayByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
@@ -717,25 +814,34 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArrayByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsArrayExecutor executor(elements);
index->ExecutorForShreddingData<std::string_view>(
op_ctx_,
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
auto shared_executor = [&elements, &res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto shared_executor = [&elements, &res_view, &valid_res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto array = bson.ParseAsArrayAtOffset(value_offset);
if (!array.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
for (const auto& sub_value : array.value()) {
auto sub_array = milvus::BsonView::GetValueFromBsonView<
@@ -746,11 +852,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArrayByStats() {
for (const auto& element : elements) {
if (CompareTwoJsonArray(sub_array.value(), element)) {
return true;
res_view[row_offset] = true;
return;
}
}
}
return false;
res_view[row_offset] = false;
};
{
milvus::ScopedTimer timer(
@@ -761,12 +868,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsArrayByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
template <typename ExprValueType>
@@ -917,9 +1024,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAll(EvalCtx& context) {
auto elements =
std::static_pointer_cast<std::set<GetType>>(arg_cached_set_);
int processed_cursor = 0;
ContainsAllMatcher<GetType> matcher(*elements);
std::vector<uint64_t> found_large(
matcher.use_small() ? 0 : matcher.num_words());
auto execute_sub_batch =
[&processed_cursor, &
bitmap_input ]<FilterType filter_type = FilterType::sequential>(
[&processed_cursor, &bitmap_input, &matcher, &
found_large ]<FilterType filter_type = FilterType::sequential>(
const milvus::Json* data,
const bool* valid_data,
const int32_t* offsets,
@@ -934,37 +1044,67 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAll(EvalCtx& context) {
processed_cursor += size;
return;
}
auto executor = [&](const size_t i) -> bool {
auto executor = [&](const size_t i) {
auto doc = data[i].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
std::set<GetType> tmp_elements(elements);
// Note: array can only be iterated once
for (auto&& it : array) {
auto val = it.template get<GetType>();
if (val.error()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_val = it.template get<double>();
if (!double_val.error() &&
double_val.value() ==
std::floor(double_val.value())) {
tmp_elements.erase(
static_cast<int64_t>(double_val.value()));
if (tmp_elements.size() == 0) {
return true;
if (matcher.use_small()) {
uint64_t found = 0;
for (auto&& it : array) {
auto val = it.template get<GetType>();
if (val.error()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_val = it.template get<double>();
if (!double_val.error() &&
double_val.value() ==
std::floor(double_val.value())) {
if (matcher.set_if_found(
static_cast<int64_t>(
double_val.value()),
found)) {
return std::make_pair(true, true);
}
}
}
continue;
}
if (matcher.set_if_found(val.value(), found)) {
return std::make_pair(true, true);
}
continue;
}
tmp_elements.erase(val.value());
if (tmp_elements.size() == 0) {
return true;
return std::make_pair(true, found == matcher.full_mask());
} else {
std::fill(found_large.begin(), found_large.end(), 0);
size_t remaining = matcher.target_count();
for (auto&& it : array) {
auto val = it.template get<GetType>();
if (val.error()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_val = it.template get<double>();
if (!double_val.error() &&
double_val.value() ==
std::floor(double_val.value())) {
if (matcher.set_if_found(
static_cast<int64_t>(
double_val.value()),
found_large,
remaining)) {
return std::make_pair(true, true);
}
}
}
continue;
}
if (matcher.set_if_found(
val.value(), found_large, remaining)) {
return std::make_pair(true, true);
}
}
return std::make_pair(true, remaining == 0);
}
return tmp_elements.size() == 0;
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -979,7 +1119,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAll(EvalCtx& context) {
if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -1049,7 +1194,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
// process shredding data for ARRAY type (non-shared)
@@ -1060,6 +1205,10 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsAllExecutor<GetType> executor(
*elements);
@@ -1068,52 +1217,97 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllByStats() {
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
// process shared data
auto shared_executor = [this, &elements, &res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
ContainsAllMatcher<GetType> shared_matcher(*elements);
std::vector<uint64_t> shared_found_large(
shared_matcher.use_small() ? 0 : shared_matcher.num_words());
auto shared_executor = [&shared_matcher,
&res_view,
&valid_res_view,
&shared_found_large](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto val = bson.ParseAsArrayAtOffset(value_offset);
if (!val.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
std::set<GetType> tmp_elements(*elements);
for (const auto& element : val.value()) {
auto value = milvus::BsonView::GetValueFromBsonView<GetType>(
element.get_value());
if (!value.has_value()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_value =
milvus::BsonView::GetValueFromBsonView<double>(
element.get_value());
if (double_value.has_value()) {
if (double_value.value() ==
std::floor(double_value.value())) {
tmp_elements.erase(
static_cast<int64_t>(double_value.value()));
}
if (tmp_elements.size() == 0) {
res_view[row_offset] = true;
return;
if (shared_matcher.use_small()) {
uint64_t found = 0;
for (const auto& element : val.value()) {
auto value =
milvus::BsonView::GetValueFromBsonView<GetType>(
element.get_value());
if (!value.has_value()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_value =
milvus::BsonView::GetValueFromBsonView<double>(
element.get_value());
if (double_value.has_value() &&
double_value.value() ==
std::floor(double_value.value())) {
if (shared_matcher.set_if_found(
static_cast<int64_t>(
double_value.value()),
found)) {
res_view[row_offset] = true;
return;
}
}
}
continue;
}
if (shared_matcher.set_if_found(value.value(), found)) {
res_view[row_offset] = true;
return;
}
continue;
}
tmp_elements.erase(value.value());
if (tmp_elements.size() == 0) {
res_view[row_offset] = true;
return;
res_view[row_offset] = found == shared_matcher.full_mask();
} else {
std::fill(
shared_found_large.begin(), shared_found_large.end(), 0);
size_t remaining = shared_matcher.target_count();
for (const auto& element : val.value()) {
auto value =
milvus::BsonView::GetValueFromBsonView<GetType>(
element.get_value());
if (!value.has_value()) {
if constexpr (std::is_same_v<GetType, int64_t>) {
auto double_value =
milvus::BsonView::GetValueFromBsonView<double>(
element.get_value());
if (double_value.has_value() &&
double_value.value() ==
std::floor(double_value.value())) {
if (shared_matcher.set_if_found(
static_cast<int64_t>(
double_value.value()),
shared_found_large,
remaining)) {
res_view[row_offset] = true;
return;
}
}
}
continue;
}
if (shared_matcher.set_if_found(
value.value(), shared_found_large, remaining)) {
res_view[row_offset] = true;
return;
}
}
res_view[row_offset] = remaining == 0;
}
res_view[row_offset] = tmp_elements.empty();
};
{
milvus::ScopedTimer timer(
@@ -1125,12 +1319,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
VectorPtr
@@ -1190,12 +1384,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffType(EvalCtx& context) {
processed_cursor += size;
return;
}
auto executor = [&](size_t i) -> bool {
auto executor = [&](size_t i) {
const auto& json = data[i];
auto doc = json.dom_doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
std::unordered_set<int> tmp_elements_index(elements_index);
for (auto&& it : array) {
@@ -1264,14 +1458,14 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffType(EvalCtx& context) {
element.val_case());
}
if (tmp_elements_index.size() == 0) {
return true;
return std::make_pair(true, true);
}
}
if (tmp_elements_index.size() == 0) {
return true;
return std::make_pair(true, true);
}
}
return tmp_elements_index.size() == 0;
return std::make_pair(true, tmp_elements_index.size() == 0);
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -1287,7 +1481,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffType(EvalCtx& context) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -1350,7 +1549,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffTypeByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
@@ -1362,6 +1561,10 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffTypeByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsAllWithDiffTypeExecutor executor(
elements, elements_index);
index->ExecutorForShreddingData<std::string_view>(
@@ -1369,21 +1572,26 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffTypeByStats() {
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
auto shared_executor = [&elements, &elements_index, &res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto shared_executor = [&elements,
&elements_index,
&res_view,
&valid_res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
std::set<int> tmp_elements_index(elements_index);
auto array = bson.ParseAsArrayAtOffset(value_offset);
if (!array.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
for (const auto& sub_value : array.value()) {
int i = -1;
@@ -1477,12 +1685,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllWithDiffTypeByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
VectorPtr
@@ -1544,7 +1752,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArray(EvalCtx& context) {
auto doc = data[i].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
std::unordered_set<int> exist_elements_index;
for (auto&& it : array) {
@@ -1565,10 +1773,11 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArray(EvalCtx& context) {
}
}
if (exist_elements_index.size() == elements.size()) {
return true;
return std::make_pair(true, true);
}
}
return exist_elements_index.size() == elements.size();
return std::make_pair(
true, exist_elements_index.size() == elements.size());
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -1584,7 +1793,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArray(EvalCtx& context) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -1643,7 +1857,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArrayByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
@@ -1655,25 +1869,33 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArrayByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsAllArrayExecutor executor(elements);
index->ExecutorForShreddingData<std::string_view>(
op_ctx_,
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
auto shared_executor = [&elements, &res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto shared_executor = [&elements, &res_view, &valid_res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto array = bson.ParseAsArrayAtOffset(value_offset);
if (!array.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
std::set<int> exist_elements_index;
for (const auto& sub_value : array.value()) {
@@ -1707,12 +1929,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsAllArrayByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
VectorPtr
@@ -1771,7 +1993,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
auto doc = json.dom_doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
return false;
return std::make_pair(false, false);
}
// Note: array can only be iterated once
for (auto&& it : array) {
@@ -1783,7 +2005,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
continue;
}
if (val.value() == element.bool_val()) {
return true;
return std::make_pair(true, true);
}
break;
}
@@ -1793,12 +2015,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
auto double_val = it.template get<double>();
if (!double_val.error() &&
double_val.value() == element.int64_val()) {
return true;
return std::make_pair(true, true);
}
continue;
}
if (val.value() == element.int64_val()) {
return true;
return std::make_pair(true, true);
}
break;
}
@@ -1808,7 +2030,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
continue;
}
if (val.value() == element.float_val()) {
return true;
return std::make_pair(true, true);
}
break;
}
@@ -1818,7 +2040,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
continue;
}
if (val.value() == element.string_val()) {
return true;
return std::make_pair(true, true);
}
break;
}
@@ -1828,7 +2050,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
continue;
}
if (CompareTwoJsonArray(val, element.array_val())) {
return true;
return std::make_pair(true, true);
}
break;
}
@@ -1839,7 +2061,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
}
}
}
return false;
return std::make_pair(true, false);
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -1855,7 +2077,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffType(EvalCtx& context) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -1910,7 +2137,7 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffTypeByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
@@ -1922,6 +2149,10 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffTypeByStats() {
auto target_field = index->GetShreddingField(
pointer, milvus::index::JSONType::ARRAY);
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonContainsAnyWithDiffTypeExecutor executor(
elements);
index->ExecutorForShreddingData<std::string_view>(
@@ -1929,19 +2160,23 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffTypeByStats() {
target_field,
executor,
nullptr,
res_view,
valid_res_view);
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
}
}
auto shared_executor = [&elements, &res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto shared_executor = [&elements, &res_view, &valid_res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto array = bson.ParseAsArrayAtOffset(value_offset);
if (!array.has_value()) {
res_view[row_offset] = false;
return;
}
valid_res_view[row_offset] = true;
for (const auto& sub_value : array.value()) {
for (auto const& element : elements) {
@@ -2028,12 +2263,12 @@ PhyJsonContainsFilterExpr::ExecJsonContainsWithDiffTypeByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
VectorPtr
internal/core/src/exec/expression/JsonContainsExpr.h
+6 -6
View File
@@ -52,7 +52,7 @@ class ShreddingArrayBsonContainsArrayExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array_view = bson.ParseAsArrayAtOffset(0);
if (!array_view.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
bool matched = false;
@@ -100,7 +100,7 @@ class ShreddingArrayBsonContainsAllArrayExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array_view = bson.ParseAsArrayAtOffset(0);
if (!array_view.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
std::set<int> exist_elements_index;
@@ -154,7 +154,7 @@ class ShreddingArrayBsonContainsAnyExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array_view = bson.ParseAsArrayAtOffset(0);
if (!array_view.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
bool matched = false;
@@ -210,7 +210,7 @@ class ShreddingArrayBsonContainsAllExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array_view = bson.ParseAsArrayAtOffset(0);
if (!array_view.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
std::set<GetType> tmp_elements(elements_);
@@ -257,7 +257,7 @@ class ShreddingArrayBsonContainsAllWithDiffTypeExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array = bson.ParseAsArrayAtOffset(0);
if (!array.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
std::set<int> tmp_elements_index(elements_index_);
@@ -369,7 +369,7 @@ class ShreddingArrayBsonContainsAnyWithDiffTypeExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array = bson.ParseAsArrayAtOffset(0);
if (!array.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
bool matched = false;
internal/core/src/exec/expression/TermExpr.cpp
+72 -77
View File
@@ -498,18 +498,19 @@ PhyTermFilterExpr::ExecTermJsonVariableInField(EvalCtx& context) {
auto executor = [&](size_t i) {
auto doc = data[i].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error())
return false;
if (array.error()) {
return std::make_pair(false, false);
}
for (auto it = array.begin(); it != array.end(); ++it) {
auto val = (*it).template get<GetType>();
if (val.error()) {
return false;
continue;
}
if (val.value() == target_val) {
return true;
return std::make_pair(true, true);
}
}
return false;
return std::make_pair(true, false);
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -524,7 +525,12 @@ PhyTermFilterExpr::ExecTermJsonVariableInField(EvalCtx& context) {
if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
@@ -592,18 +598,20 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
// process shredding data
auto try_execute = [&](milvus::index::JSONType json_type,
TargetBitmapView& res_view,
TargetBitmapView& valid_res_view,
auto GetType) {
auto target_field = index->GetShreddingField(pointer, json_type);
if (!target_field.empty()) {
using ColType = decltype(GetType);
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
auto shredding_executor = [this](const ColType* src,
const bool* valid,
size_t size,
@@ -621,13 +629,15 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
}
}
};
index->template ExecutorForShreddingData<ColType>(
op_ctx_,
target_field,
shredding_executor,
nullptr,
res_view,
valid_res_view);
index->ExecutorForShreddingData<ColType>(op_ctx_,
target_field,
shredding_executor,
nullptr,
target_res_view,
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
LOG_DEBUG("using shredding data's field: {} count {}",
target_field,
res_view.count());
@@ -640,60 +650,28 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
[this](double us) { json_stats_shredding_latency_us_ += us; });
if constexpr (std::is_same_v<GetType, bool>) {
try_execute(milvus::index::JSONType::BOOL,
res_view,
valid_res_view,
bool{});
try_execute(milvus::index::JSONType::BOOL, bool{});
} else if constexpr (std::is_same_v<GetType, int64_t>) {
try_execute(milvus::index::JSONType::INT64,
res_view,
valid_res_view,
int64_t{});
try_execute(milvus::index::JSONType::INT64, int64_t{});
// and double compare
TargetBitmap res_double(active_count_, false);
TargetBitmapView res_double_view(res_double);
TargetBitmap res_double_valid(active_count_, true);
TargetBitmapView valid_res_double_view(res_double_valid);
try_execute(milvus::index::JSONType::DOUBLE,
res_double_view,
valid_res_double_view,
double{});
res_view.inplace_or_with_count(res_double_view, active_count_);
valid_res_view.inplace_or_with_count(valid_res_double_view,
active_count_);
try_execute(milvus::index::JSONType::DOUBLE, double{});
} else if constexpr (std::is_same_v<GetType, double>) {
try_execute(milvus::index::JSONType::DOUBLE,
res_view,
valid_res_view,
double{});
try_execute(milvus::index::JSONType::DOUBLE, double{});
// and int64 compare
TargetBitmap res_int64(active_count_, false);
TargetBitmapView res_int64_view(res_int64);
TargetBitmap res_int64_valid(active_count_, true);
TargetBitmapView valid_res_int64_view(res_int64_valid);
try_execute(milvus::index::JSONType::INT64,
res_int64_view,
valid_res_int64_view,
int64_t{});
res_view.inplace_or_with_count(res_int64_view, active_count_);
valid_res_view.inplace_or_with_count(valid_res_int64_view,
active_count_);
try_execute(milvus::index::JSONType::INT64, int64_t{});
} else if constexpr (std::is_same_v<GetType, std::string_view> ||
std::is_same_v<GetType, std::string>) {
try_execute(milvus::index::JSONType::STRING,
res_view,
valid_res_view,
std::string_view{});
}
}
// process shared data
auto shared_executor = [this, &res_view](milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
auto get_value = bson.ParseAsValueAtOffset<GetType>(value_offset);
auto shared_executor = [this, &res_view, &valid_res_view](
milvus::BsonView bson,
uint32_t row_offset,
uint32_t value_offset) {
if constexpr (std::is_same_v<GetType, int64_t> ||
std::is_same_v<GetType, double>) {
auto get_value =
@@ -701,6 +679,7 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
if (get_value.has_value()) {
res_view[row_offset] =
this->arg_set_double_->In(get_value.value());
valid_res_view[row_offset] = true;
}
return;
} else {
@@ -709,6 +688,7 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
if (get_value.has_value()) {
res_view[row_offset] =
this->arg_set_->In(get_value.value());
valid_res_view[row_offset] = true;
}
return;
}
@@ -723,12 +703,12 @@ PhyTermFilterExpr::ExecJsonInVariableByStats() {
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
template <typename ValueType>
@@ -795,22 +775,32 @@ PhyTermFilterExpr::ExecTermJsonFieldInVariable(EvalCtx& context) {
return;
}
auto executor = [&](size_t i) {
auto x = data[i].template at<GetType>(pointer);
if (x.error()) {
if constexpr (std::is_same_v<GetType, std::int64_t>) {
auto x = data[i].template at<double>(pointer);
if (x.error()) {
return false;
}
auto value = x.value();
// if the term set is {1}, and the value is 1.1, we should not return true.
return std::floor(value) == value &&
terms->In(ValueType(x.value()));
if constexpr (std::is_same_v<GetType, std::int64_t>) {
auto x_num = data[i].at_numeric(pointer);
if (x_num.error()) {
return std::make_pair(false, false);
}
return false;
auto n = x_num.value();
if (n.is_int64()) {
return std::make_pair(true,
terms->In(ValueType(n.get_int64())));
}
// uint64 or double → compare as double, consistent with
// index/stats paths.
auto dval = n.is_uint64() ? static_cast<double>(n.get_uint64())
: n.get_double();
// if the term set is {1}, and the value is 1.1, we should
// not return true.
return std::make_pair(
true,
std::floor(dval) == dval && terms->In(ValueType(dval)));
} else {
auto x = data[i].template at<GetType>(pointer);
if (x.error()) {
return std::make_pair(false, false);
}
return std::make_pair(true, terms->In(ValueType(x.value())));
}
return terms->In(ValueType(x.value()));
};
bool has_bitmap_input = !bitmap_input.empty();
for (size_t i = 0; i < size; ++i) {
@@ -830,7 +820,12 @@ PhyTermFilterExpr::ExecTermJsonFieldInVariable(EvalCtx& context) {
if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
continue;
}
res[i] = executor(offset);
auto [valid, matched] = executor(offset);
if (!valid) {
res[i] = valid_res[i] = false;
continue;
}
res[i] = matched;
}
processed_cursor += size;
};
internal/core/src/exec/expression/UnaryExpr.cpp
+76 -58
View File
@@ -705,36 +705,40 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
auto op_type = expr_->op_type_;
auto pointer = milvus::Json::pointer(expr_->column_.nested_path_);
#define UnaryRangeJSONCompare(cmp) \
do { \
auto x = data[offset].template at<GetType>(pointer); \
if (x.error()) { \
if constexpr (std::is_same_v<GetType, int64_t>) { \
auto x = data[offset].template at<double>(pointer); \
res[i] = !x.error() && (cmp); \
break; \
} \
res[i] = false; \
break; \
} \
res[i] = (cmp); \
// For int64_t GetType, uses at_numeric() (get_number()) to extract any JSON
// number in a single parse. Branches on actual type to preserve int64
// precision; uint64 and double values fall back to double comparison,
// consistent with the Tantivy index and JSON-stats paths.
// - 'cmp' must reference 'value' (auto-typed as int64_t or double).
// Missing path and type mismatch are UNKNOWN/NULL under JSON 3VL semantics.
#define UnaryRangeJSONCompare(cmp) \
do { \
if constexpr (std::is_same_v<GetType, int64_t>) { \
auto x_num = data[offset].at_numeric(pointer); \
if (x_num.error()) { \
res[i] = valid_res[i] = false; \
break; \
} \
auto n = x_num.value(); \
if (n.is_int64()) { \
auto value = n.get_int64(); \
res[i] = (cmp); \
} else { \
auto value = n.is_uint64() \
? static_cast<double>(n.get_uint64()) \
: n.get_double(); \
res[i] = (cmp); \
} \
} else { \
auto x = data[offset].template at<GetType>(pointer); \
if (x.error()) { \
res[i] = valid_res[i] = false; \
break; \
} \
auto value = x.value(); \
res[i] = (cmp); \
} \
} while (false)
#define UnaryRangeJSONCompareNotEqual(cmp) \
do { \
auto x = data[offset].template at<GetType>(pointer); \
if (x.error()) { \
if constexpr (std::is_same_v<GetType, int64_t>) { \
auto x = data[offset].template at<double>(pointer); \
res[i] = x.error() || (cmp); \
break; \
} \
res[i] = true; \
break; \
} \
res[i] = (cmp); \
} while (false)
int processed_cursor = 0;
auto execute_sub_batch =
[ op_type, pointer, &processed_cursor, &
@@ -765,7 +769,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(x.value() > val);
UnaryRangeJSONCompare(value > val);
}
}
break;
@@ -787,7 +791,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(x.value() >= val);
UnaryRangeJSONCompare(value >= val);
}
}
break;
@@ -809,7 +813,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(x.value() < val);
UnaryRangeJSONCompare(value < val);
}
}
break;
@@ -831,7 +835,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(x.value() <= val);
UnaryRangeJSONCompare(value <= val);
}
}
break;
@@ -851,15 +855,15 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
continue;
}
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
auto doc = data[i].doc();
auto doc = data[offset].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
res[i] = CompareTwoJsonArray(array, val);
} else {
UnaryRangeJSONCompare(x.value() == val);
UnaryRangeJSONCompare(value == val);
}
}
break;
@@ -879,15 +883,15 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
continue;
}
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
auto doc = data[i].doc();
auto doc = data[offset].doc();
auto array = doc.at_pointer(pointer).get_array();
if (array.error()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
res[i] = !CompareTwoJsonArray(array, val);
} else {
UnaryRangeJSONCompareNotEqual(x.value() != val);
UnaryRangeJSONCompare(value != val);
}
}
break;
@@ -911,8 +915,8 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(milvus::query::Match(
ExprValueType(x.value()), val, op_type));
UnaryRangeJSONCompare(
milvus::query::Match(value, val, op_type));
}
}
break;
@@ -937,8 +941,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJson(EvalCtx& context) {
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
res[i] = false;
} else {
UnaryRangeJSONCompare(
matcher(ExprValueType(x.value())));
UnaryRangeJSONCompare(matcher(ExprValueType(value)));
}
}
break;
@@ -1016,9 +1019,11 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
Assert(index != nullptr);
cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
cached_index_chunk_valid_res_ =
std::make_shared<TargetBitmap>(active_count_, true);
std::make_shared<TargetBitmap>(active_count_);
TargetBitmapView res_view(*cached_index_chunk_res_);
TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);
TargetBitmap field_valid(active_count_, true);
TargetBitmapView field_valid_view(field_valid);
int64_t valid_processed_size = 0;
for (size_t i = 0;
i < num_data_chunk_ && valid_processed_size < active_count_;
@@ -1039,7 +1044,7 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
i,
0,
size,
valid_res_view + valid_processed_size);
field_valid_view + valid_processed_size);
valid_processed_size += size;
}
@@ -1053,6 +1058,8 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
using ValType = decltype(ValType);
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingExecutor<ColType, ValType> executor(
op_type, pointer, val);
index->ExecutorForShreddingData<ColType>(op_ctx_,
@@ -1060,8 +1067,10 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
executor,
nullptr,
target_res_view,
target_res_view);
target_valid_view);
res_view.inplace_or_with_count(target_res_view, active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
LOG_DEBUG(
"using shredding data's field: {} with value {}, count {} "
"for segment {}",
@@ -1104,6 +1113,8 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
if (!target_field.empty()) {
TargetBitmap target_res(active_count_, false);
TargetBitmapView target_res_view(target_res);
TargetBitmap target_valid(active_count_, true);
TargetBitmapView target_valid_view(target_valid);
ShreddingArrayBsonExecutor executor(op_type, pointer, val);
index->template ExecutorForShreddingData<std::string_view>(
op_ctx_,
@@ -1111,9 +1122,11 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
executor,
nullptr,
target_res_view,
target_res_view);
target_valid_view);
res_view.inplace_or_with_count(target_res_view,
active_count_);
valid_res_view.inplace_or_with_count(target_valid_view,
active_count_);
LOG_DEBUG("using shredding array field: {}, count {}",
target_field,
res_view.count());
@@ -1122,15 +1135,19 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
}
// process shared data
auto shared_executor = [op_type, val, array_index, &res_view](
milvus::BsonView bson,
uint32_t row_id,
uint32_t value_offset) {
auto shared_executor = [op_type,
val,
array_index,
&res_view,
&valid_res_view](milvus::BsonView bson,
uint32_t row_id,
uint32_t value_offset) {
auto set_unknown = [&](uint32_t row_id) {
res_view[row_id] = false;
res_view[row_id] = valid_res_view[row_id] = false;
};
auto set_known = [&](uint32_t row_id, bool value) {
res_view[row_id] = value;
valid_res_view[row_id] = true;
};
if constexpr (std::is_same_v<GetType, proto::plan::Array>) {
Assert(op_type == proto::plan::OpType::Equal ||
@@ -1265,20 +1282,21 @@ PhyUnaryRangeFilterExpr::ExecRangeVisitorImplJsonByStats() {
index->ExecuteForSharedData(op_ctx_, pointer, shared_executor);
}
// for NotEqual: flip the result
// for NotEqual: flip the Equal result, then mask out UNKNOWN rows.
if (expr_->op_type_ == proto::plan::OpType::NotEqual) {
cached_index_chunk_res_->flip();
}
valid_res_view.inplace_and(field_valid_view, active_count_);
res_view.inplace_and(valid_res_view, active_count_);
cached_index_chunk_id_ = 0;
}
TargetBitmap result;
result.append(
*cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
*cached_index_chunk_valid_res_,
current_data_global_pos_,
real_batch_size);
MoveCursor();
return std::make_shared<ColumnVector>(std::move(result),
TargetBitmap(real_batch_size, true));
return res;
}
template <typename T>
internal/core/src/exec/expression/UnaryExpr.h
+1 -1
View File
@@ -706,7 +706,7 @@ class ShreddingArrayBsonExecutor {
reinterpret_cast<const uint8_t*>(src[i].data()), src[i].size());
auto array_view = bson.ParseAsArrayAtOffset(0);
if (!array_view.has_value()) {
res[i] = false;
res[i] = valid_res[i] = false;
continue;
}
bool equal = CompareTwoJsonArray(array_view.value(), val_);
internal/parser/planparserv2/rewriter/range_json_test.go
+4 -4
View File
@@ -69,7 +69,7 @@ func TestRewrite_JSON_NestedPath_Nullable_ContradictionsKeepPredicate(t *testing
}
}
func TestRewrite_JSON_NestedPath_ScalarNotEqualRewriteAllowed(t *testing.T) {
func TestRewrite_JSON_NestedPath_ScalarNotEqualRewriteBlocked(t *testing.T) {
helper := buildSchemaHelperWithJSON(t)
for _, exprStr := range []string{
@@ -79,9 +79,9 @@ func TestRewrite_JSON_NestedPath_ScalarNotEqualRewriteAllowed(t *testing.T) {
expr, err := parser.ParseExpr(helper, exprStr, nil)
require.NoError(t, err, exprStr)
require.NotNil(t, expr, exprStr)
ure := expr.GetUnaryRangeExpr()
require.NotNil(t, ure, "scalar JSON missing-path != is compatible with NOT(==): %s", exprStr)
require.Equal(t, planpb.OpType_NotEqual, ure.GetOp(), exprStr)
unary := expr.GetUnaryExpr()
require.NotNil(t, unary, "scalar JSON missing-path NOT must remain explicit: %s", exprStr)
require.Equal(t, planpb.UnaryExpr_Not, unary.GetOp(), exprStr)
}
}
internal/parser/planparserv2/rewriter/util.go
+1 -15
View File
@@ -221,21 +221,7 @@ func canFoldInNotEqualTautologyToTrue(col *planpb.ColumnInfo) bool {
}
func hasMissingPathNotEqualSemantics(col *planpb.ColumnInfo, values ...*planpb.GenericValue) bool {
if !hasMissingPathSemantics(col) {
return false
}
if col.GetDataType() != schemapb.DataType_JSON {
return true
}
for _, value := range values {
if value == nil || value.GetVal() == nil {
continue
}
if _, ok := value.GetVal().(*planpb.GenericValue_ArrayVal); ok {
return true
}
}
return false
return hasMissingPathSemantics(col)
}
func newAlwaysFalseExpr() *planpb.Expr {