Support preidcate delete part5(#555)

[MrPresent-Han]

related: #555

[sre-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: MrPresent-Han
To complete the pull request process, please assign sunby after the PR has been reviewed.
You can assign the PR to them by writing /assign @sunby in a comment when ready.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[codecov]

Codecov Report

❌ Patch coverage is 87.35225% with 107 lines in your changes missing coverage. Please review.
✅ Project coverage is 75.40%. Comparing base (a27f61b) to head (a4761df).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
cpp/src/delete_evaluator.cpp	87.50%	42 Missing ⚠️
cpp/src/ffi/reader_c.cpp	68.29%	39 Missing ⚠️
cpp/src/common/sql_predicate_arrow.cpp	92.19%	22 Missing ⚠️
cpp/src/ffi/manifest_c.cpp	88.88%	2 Missing ⚠️
cpp/src/reader.cpp	97.22%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #556      +/-   ##
==========================================
+ Coverage   74.91%   75.40%   +0.49%     
==========================================
  Files         159      164       +5     
  Lines       15272    16421    +1149     
  Branches     2336     2528     +192     
==========================================
+ Hits        11441    12383     +942     
- Misses       3831     4038     +207     

Flag	Coverage Δ
cpp	78.08% <87.35%> (+0.33%)	⬆️
python	44.76% <ø> (+0.02%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[liliu-z]

The FFI alive-reader path (loon_alive_reader_new) creates the Reader and eagerly builds the masked stream without ever supplying a key retriever: the manifest constructor sets key_retriever_callback_ to null (reader.cpp:796), the function has no retriever parameter, and unlike the normal reader (which exposes loon_reader_set_keyretriever) the returned handle is an AliveReaderFFIState rather than a Reader, so a retriever can never be set. Reading an encrypted dataset through the FFI alive reader therefore fails. The C++ API supports encryption via set_keyretriever before get_masked_record_batch_reader; thread the retriever through the FFI alive-reader path the same way.

[MrPresent-Han]

The DataNode-side segment delta log path currently does not go through the CMEK encryption path, so delta logs are not encrypted today. I do not think we should implement key retriever support for this path ahead of actual encrypted delta-log support.

[liliu-z]

LoadDeltaLogs dispatches each delta log solely by inspecting the file schema (ShouldReadAsPredicateDeltaLog); the persisted delta_log.type is never consulted, so anything not predicate-shaped — including a POSITIONAL or EQUALITY delta log — silently falls into the PRIMARY_KEY branch, where LoadPrimaryKeyDeltaLog's positional column fallback (:812-814) then consumes its columns as pk/ts. parse_delta_log_type (manifest_c.cpp:242) accepts these types and Avro round-trips them (manifest.cpp:136), so an external producer can create one and have wrong delete semantics applied. No in-tree producer emits POSITIONAL/EQUALITY today, but the evaluator should reject unsupported types instead of coercing them to PRIMARY_KEY.

[MrPresent-Han]

Updated the design accordingly: the alive reader now dispatches by the manifest delta_log.type instead of using the delta-log file schema as the routing source. PREDICATE delta logs still assert their file-level predicate metadata/schema during loading, and POSITIONAL/EQUALITY delta logs now fail fast as unsupported instead of falling back to PRIMARY_KEY.

[liliu-z]

delta_log_type was inserted as the 3rd parameter of loon_transaction_add_delta_log (before num_entries) rather than appended, changing the ABI while the symbol name is unchanged. A caller compiled against the old 3-arg signature and linked without recompiling would pass num_entries where the new function reads delta_log_type, and read num_entries from an unset register. All in-tree consumers (including the Python binding, updated in this PR to pass path, 0, num_entries) rebuild in lockstep, so realized risk is low; appending the parameter (or versioning the symbol) would remove the footgun.

[MrPresent-Han]

Updated. I moved the new delta_log_type argument to the end of loon_transaction_add_delta_log, so the existing parameter order stays (handle, path, num_entries) and the new type parameter is appended instead of inserted in the middle.

[liliu-z]

loon_alive_reader_new default-constructs AliveReadOptions and exposes none of its fields, so FFI consumers cannot set visible_until_ts (nor batch_size/parallelism). With visible_until_ts unset the evaluator applies every delete regardless of timestamp (delete_evaluator.cpp:844 only skips deletes when the option has a value), so snapshot/time-bounded alive reads — supported by the C++ API and covered by C++ tests — cannot be expressed through the FFI. Add the options to the FFI entry point and forward them to get_masked_record_batch_reader.

[MrPresent-Han]

changed

[liliu-z]

For every row, EvaluatePredicateDeletes (loop at :988) calls EvalPredicateNode, which resolves each referenced field via FindFieldIndexByFieldId at :590 — a linear scan over all schema fields that calls GetFieldId (which std::stoll-parses the field's PARQUET:field_id metadata or name) on each field. The field-id-to-index mapping is constant for the whole batch, so this repeats O(rows × predicate-nodes × fields) string conversions on the hot path and degrades sharply on large batches and wide schemas. Resolve the mapping once before the row loop and index into it.

[MrPresent-Han]

changed

[liliu-z]

This PR adds a third pointer field delta_log_types to the C LoonDeltaLogs struct (and manifest_export now populates it), but the hand-written cffi cdef mirror in python/milvus_storage/_ffi.py (the LoonDeltaLogs typedef near line 174) was not updated to match. In cffi ABI mode the cdef defines the struct layout, so Manifest._from_c reads num_delta_logs — and the stats member that follows delta_logs inside LoonManifest — at shifted offsets, producing a garbage count and an out-of-bounds read for any manifest that carries a delta log (the primary path for this delete feature). Fix by adding uint32_t* delta_log_types; before num_delta_logs in the Python cdef.

[MrPresent-Han]

Good catch — confirmed and fixed. Added uint32_t* delta_log_types; before num_delta_logs in the LoonDeltaLogs cdef in python/milvus_storage/_ffi.py, so the ABI-mode struct layout now matches the C header and num_delta_logs (and the following stats member) are read at the correct offsets. (6cb7582)

[liliu-z]

The tokenizer rejects the - character, so negative numeric literals never parse and predicates such as col < -5 or col in (-1, -2) fail. Instead of filtering, such a predicate aborts the entire segment read and makes the segment unreadable — though this only bites if Milvus's serializer actually emits negative literals, which isn't verified from this repo. Handle a leading - as part of a numeric literal during tokenization.

[MrPresent-Han]

Fixed. The tokenizer now treats a - immediately followed by a digit as a negative numeric literal (e.g. col < -5, col in (-1, -2)); a - not followed by a digit (unary minus like -a) still falls through to ReadOperator and is rejected, so unary minus stays unsupported. Added SupportsNegativeNumericLiterals test coverage. (6cb7582)

[liliu-z]

Every predicate delete expression is re-bound on every batch via expression.Bind(*batch->schema()), even though all batches from the reader share the same projected schema, so the bind result is identical each time. Bind once (lazily on the first batch, or against the projected schema at construction) and cache the bound expressions. Note the load-time Bind(*schema_) at line 421 binds against the full (unprojected) schema and is deliberately discarded for validation, so it cannot be reused directly for the projected batch schema — the redundant work to eliminate is the per-batch rebind, not simply 'reusing the load-time result'.

[MrPresent-Han]

Fixed. Predicate expressions are now bound once against the first batch's schema and cached in bound_predicate_expressions_, then reused for subsequent batches. As you noted, the load-time Bind(*schema_) is validation-only against the full schema and is left untouched. (6cb7582)

[liliu-z]

loon_alive_reader_new (reader_c.cpp:644) builds the read stream eagerly while ReaderImpl::key_retriever_callback_ is still null, and LoonAliveReaderHandle exposes no setter to install one, so that null callback is forwarded both to CreateDeleteEvaluator here and to the base reader at reader.cpp:900. No code path on the alive reader can therefore supply a decryption key, so any encrypted dataset or encrypted delta log fails to decrypt and reads break. Install the callback before constructing the stream, or add a setter on the handle and defer stream creation until it is provided.

[jiaqizho]

why need pass batch_size_ and parallelism to override old logical?

[MrPresent-Han]

Good point — these two were redundant, so I removed batch_size and parallelism from both AliveReadOptions and the FFI LoonAliveReadOptions. The alive read now derives batch size from the existing per-reader property reader.record_batch_max_rows and parallelism from the global thread pool (ThreadPoolHolder), i.e. the > 0 ? override : old_source logic here falls through to the old sources. No behavior lost; one less duplicated knob to keep in sync across the C ABI. (Note: this drops two fields from LoonAliveReadOptions, so the Milvus-side caller needs to update the struct.)

[jiaqizho]

Current reader is streaming reader.

I'm wondering if get_chunks and take here should also include corresponding handling? There might not be a need for take right now( But lance already support), but perhaps there is for get_chunks?

[MrPresent-Han]

Right — the alive / delete-mask path is intentionally streaming-only in this PR: the keep-mask is computed and applied per batch inside loon_alive_reader_next, so the streaming alive reader returns each batch together with its alive bitset. Masked get_chunks/take (applying the delete mask to random-access chunk/row reads) is a reasonable follow-up but out of scope here — take isn't needed for the current streaming consumption, and if a chunk-level masked read turns out to be needed we can add it in a separate PR. Keeping this PR focused on the streaming alive read.

[jiaqizho]

Also there are a problem: If current column is TEXT field, then the LOB would be apply.

[tedxu]

A few naming / API-shape and hot-path comments. Individual threads inline.

[tedxu]

Two problems with the Alive* terminology across the API/FFI (AliveReadOptions, LoonAliveReaderHandle, LoonAliveReadOptions, LoonAliveBitset, loon_alive_reader_*, AliveReaderFFIState, AliveBitsetPrivateData):

1. "Alive" implies deleted rows are dead, but that's not what this reader models. Visibility is bounded by visible_until_ts, so a row deleted at ts=100 is still "alive" for a snapshot at ts=50. The name lies about the semantics.
2. "Stream" is redundant. Every RecordBatchReader in the codebase is already streaming; we don't spell that out anywhere else (get_record_batch_reader doesn't need to say "stream").

The internal C++ type is already MaskedRecordBatchReader and that reads well. Suggest aligning the FFI/options names with it: LoonMaskedReaderHandle, LoonMaskedReadOptions, MaskedReadOptions, loon_masked_reader_new/next/destroy, LoonRowMask (instead of LoonAliveBitset). If Masked isn't preferred, DeleteAware/Filtered are also more accurate than Alive.

[MrPresent-Han]

Done — renamed the entire API/FFI surface to Masked, matching the internal MaskedRecordBatchReader: MaskedReadOptions, LoonMaskedReadOptions, LoonMaskedReaderHandle, loon_masked_reader_new/next/destroy, and LoonRowMask (replacing LoonAliveBitset). The internal FFI state / private-data types (MaskedReaderFFIState, RowMaskPrivateData) were renamed to match. No Alive* or redundant *Stream names remain on the public API/FFI.

[tedxu]

EQUALITY = 2 is marked deprecated, but we still keep it in the enum, parse it in the FFI (manifest_c.cpp:242), round-trip it through Avro (manifest.cpp:136), and then explicitly reject it in the reader (delete_evaluator.cpp:260-262). That's four places carrying an unshipped concept just to say "don't use this."

Since equality delete has never been published, we can just drop the variant. If we want to distinguish "predicate that's just an equality" from "arbitrary SQL predicate" as a downstream optimization hint, rename it to something like PREDICATE_EQUALITY and treat it as a subtype — but don't leave a rejected-only enum value on the public API.

[MrPresent-Han]

Dropped the variant entirely, per "never published." Removed EQUALITY from DeltaLogType and LoonDeltaLogType, from parse_delta_log_type (manifest_c.cpp), and from both reader switches in delete_evaluator.cpp. PREDICATE keeps its wire value 3 and value 2 is left as a documented retired slot, so there's no on-disk renumbering.

Added regression coverage: MaskedReaderRejectsUnsupportedDeltaLogTypes now also feeds the retired value 2 and asserts it is rejected as an unknown type rather than silently coerced into the PRIMARY_KEY branch.

[tedxu]

The comment at :157-158 already states "milvus-storage has no inherent primary-key concept" — that's why AliveReadOptions.pk_field_id exists and is validated at :296-300. But we then hardcode kMilvusTimestampFieldId = 1 and require every schema to have an int64 field at field-id 1 named as the row timestamp (see also ResolveTimestampFieldNameByFieldId and the use at :197, :511). That's the same class of Milvus-specific coupling we said we didn't want.

Two ways out:

• Symmetrical fix: add AliveReadOptions.row_timestamp_field_id and require the caller to declare it, matching how PK is handled.
• Or accept that row-timestamp is a storage-model primitive (unlike PK): document it explicitly on the reader header, drop the kMilvusTimestamp prefix, and stop apologizing for the PK coupling in the surrounding comments. Pick one story.

As written the code says "we're generic" but behaves "we're Milvus-shaped" — the worst spot to be in.

[MrPresent-Han]

Took the symmetrical option. Added MaskedReadOptions.row_timestamp_field_id (mirrored as LoonMaskedReadOptions.row_timestamp_field_id); the caller must declare it, exactly like pk_field_id. Removed the hardcoded kMilvusTimestampFieldId — a new RowTimestampFieldId() helper returns the caller-declared id and errors when deletes are present but it is unset. The header comment now documents both pk_field_id and row_timestamp_field_id as caller-declared, so the "storage has no inherent PK/timestamp concept" story is consistent with the behavior end to end.

[tedxu]

LoadPrimaryKeyDeltaLog at :336-344 reads by these names first, then silently falls back to positional column 0/1 if the names don't match:

std::shared_ptr<arrow::Array> pk_array = batch->GetColumnByName(kDeltaPkColumnName);
std::shared_ptr<arrow::Array> ts_column = batch->GetColumnByName(kDeltaTsColumnName);
if (!pk_array || !ts_column) {
  pk_array = batch->column(0);
  ts_column = batch->column(1);
}

Two problems:

1. The strings "pk"/"ts" are the delta writer's implicit contract but they aren't declared anywhere. A producer that uses different names still "works" because of the positional fallback, so the name lookup is decorative.
2. The fallback masks bugs: if a future writer swaps column order (column 0 = ts, column 1 = pk), the delete semantics silently invert — no error surfaces.

LoadPredicateDeltaLog at :409-414 already does the honest thing: pure positional (columns 0/1), enforced by type checks in ValidatePredicateDeltaBatchSchema, constants documented at :48-52. Do the same for PRIMARY_KEY: pick position, enforce with type checks, drop the name lookup, delete these two constants.

[MrPresent-Han]

Done — LoadPrimaryKeyDeltaLog is now pure positional (column 0 = primary key, column 1 = delete_timestamp) with a type check on the timestamp column, mirroring LoadPredicateDeltaLog. Dropped the GetColumnByName lookup and deleted the kDeltaPkColumnName / kDeltaTsColumnName constants; the position contract is documented on the kPrimaryKeyDelta*Column constants. Added PrimaryKeyDeltaLogReadsByPhysicalColumnOrder (a delta whose column names don't match) to lock the positional behavior so a future writer swapping order can no longer silently invert delete semantics.

[tedxu]

pk_array->GetString(i) materializes a std::string for every row just to use as an unordered_map key. For an 8K-row batch with string PKs that's 8K allocations per batch we immediately throw away, on the delete hot path.

A heterogeneous-lookup map (C++20 transparent hash/equal so we can find(std::string_view)) or a byte-hash map with a string_view overload would let us skip the copy. Not urgent for correctness, but this is the kind of thing that will show up in a flamegraph the first time someone benches predicate + PK delete on string-PK collections.

[MrPresent-Han]

Done — string_pk_delete_ts_ now uses a transparent-hash map (TransparentStringHash with is_transparent, plus std::equal_to<>), and the per-row lookup uses pk_array->GetView(i) (a std::string_view). No std::string is materialized per row on the delete path anymore.

[tedxu]

Two related observations on how PREDICATE deltas accumulate here:

1. No deduplication at load time. Each row in a PREDICATE delta log becomes its own Expression pushed into predicate_delete_expressions_, and EvaluatePredicateDeletes (:588) loops through every one on every batch. A segment with N delete events × M batches ⇒ N·M ExecuteScalarExpression calls, even when many delete rows share (predicate_sql, delete_ts) or share predicate_sql with different delete_ts (mergeable as predicate AND ts <= max(delete_ts)). Not blocking for this PR — worth a comment marking the O(delete_events × batches) scaling as intentional near-term.

2. Non-atomic state on partial failure. If parsing/validation of a later row fails, predicate_delete_expressions_ already holds the predicates from earlier iterations (and earlier delta files in the same load). Currently harmless because Create throws the evaluator away on error, but it's a footgun the moment we expose reload or incremental delta loading.

[MrPresent-Han]

Addressed both points:

1. Deduplication at load time. Predicate deltas are now deduplicated by predicate_sql, keeping max(delete_ts) — the mergeable predicate AND row_ts <= max(delete_ts) form you described. predicate_max_ts_ accumulates during load, and BuildPredicateExpressions emits one bound expression per unique predicate, so EvaluatePredicateDeletes loops over unique predicates instead of raw delete rows (bounding it by distinct predicates × batches, not delete_events × batches). Covered by SamePredicateMultipleTimestampsUseMaxTimestamp.

2. Atomicity on partial failure. A predicate that fails to parse/bind is treated as a correctness error and aborts Create — the evaluator is discarded and no partial delete set is ever applied. We deliberately do not skip an unparseable delete row (skipping would under-delete and surface deleted data), so the state is all-or-nothing at construction.

[MrPresent-Han]

1. store pk/ts field directly in the manifest to avoid passing into milvus-storage everytime
2. confirm whether the returned alive bitset can be used in the query-node DeleteRecord, especially in the TieredStrorage case where only a separated part is loaded, may be we need an api to only return alivebitset like as using neededColumns={}
3. if one Record is deleted completely, we should return a total-null record batch