From f662cec16d248b56f09998c564fb8107e5866400 Mon Sep 17 00:00:00 2001 From: aoiasd <45024769+aoiasd@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:08:39 +0800 Subject: [PATCH] feat: add local format metadata and split policy (#51204) relate: #50304 ## Summary Add the local format design doc, use the storage writer format constant, and introduce local_format metadata propagation with column-group split policy support. --------- Signed-off-by: aoiasd Co-authored-by: Claude Opus 4.6 --- .../design_docs/20260305-local_format.md | 541 ++++++++++++++++++ internal/core/src/common/Consts.h | 4 + internal/core/src/common/FieldMeta.cpp | 38 +- internal/core/src/common/FieldMeta.h | 48 +- internal/core/src/common/FieldMetaTest.cpp | 46 ++ internal/flushcommon/syncmgr/task.go | 2 +- .../rootcoord/create_collection_task_test.go | 26 + internal/rootcoord/util.go | 34 ++ internal/rootcoord/util_test.go | 64 +++ .../storagecommon/column_group_splitter.go | 3 +- internal/storagecommon/split_policy.go | 177 +++++- internal/storagecommon/split_policy_test.go | 354 ++++++++++++ internal/storagev2/packed/ffi_common.go | 2 +- pkg/common/common.go | 7 + 14 files changed, 1290 insertions(+), 56 deletions(-) create mode 100644 docs/design-docs/design_docs/20260305-local_format.md diff --git a/docs/design-docs/design_docs/20260305-local_format.md b/docs/design-docs/design_docs/20260305-local_format.md new file mode 100644 index 0000000000..5e7401ce73 --- /dev/null +++ b/docs/design-docs/design_docs/20260305-local_format.md @@ -0,0 +1,541 @@ +# MEP: Local Format for Storage V3 Scalar Fields + +- **Created:** 2026-03-05 +- **Author(s):** @zhicheng +- **Status:** Under Review +- **Component:** QueryNode | DataNode | Storage +- **Related Issues:** milvus-io/milvus#50304 +- **Released:** TBD + +## Summary + +Add a field-level `local_format` type parameter for sealed segment scalar data +loaded through Storage V3. The default value is `raw`, which keeps the existing +Milvus on-node raw chunk layout. The first alternate value is `vortex`, which +loads Vortex column group files through a cell-based local format path. + +The proposal keeps the public field schema model small: + +- `local_format=raw`: existing behavior. +- `local_format=vortex`: use Vortex local format when the Storage V3 manifest + also points to a Vortex physical column group for that field. + +Vortex local format is a read-path feature for sealed scalar fields. It does not +change growing segment execution, vector index execution, or the public query +language. It changes how QueryNode loads and scans sealed scalar column data. + +## Motivation + +Raw scalar chunks are simple and fast when fully resident, but they require +Milvus to materialize the field data in its raw on-node layout. This is costly +for large VARCHAR, JSON, ARRAY, and other scalar fields when a query only needs a +subset of rows or only needs a predicate result. + +Vortex provides compressed, columnar files with row-group metadata and optional +zone maps. Local format support lets Milvus keep Vortex data in its native file +layout and materialize only the cells needed by scan or take operations. + +The design goals are: + +- Reduce sealed scalar field load memory for Storage V3 segments. +- Keep the existing raw path as the default and avoid adding copies to it. +- Let expression evaluation consume scalar data through a scan cursor instead + of repeatedly materializing chunks. +- Pin Vortex data at a well-defined cell granularity in the Milvus cache layer. +- Keep the common `FormatReader` interface stable; Vortex-specific operations + are exposed as Vortex extensions. +- Support normal filter, offset-input filter, and retrieve/requery output paths + with clear and separate execution plans. + +The non-goals for the initial implementation are: + +- Vortex local format for vector fields. +- Vortex local format for primary-key fields. +- Changing the query expression language. +- Full predicate pushdown for every scalar expression. +- Bitmap/selection pushdown for offset-input execution. + +## Public Interfaces + +### Field Type Parameter + +`local_format` is a field type parameter. + +Valid values: + +| Value | Meaning | +|-------|---------| +| `raw` | Default. Load sealed scalar data into the existing raw local format. | +| `vortex` | Prefer Vortex local format for this field when the physical Storage V3 column group is Vortex. | + +Example schema intent: + +```python +schema.add_field( + field_name="description", + datatype=DataType.VARCHAR, + max_length=65535, + type_params={"local_format": "vortex"}, +) +``` + +SDKs may expose this as a direct field option, but the Milvus server stores and +validates it as a field type parameter. + +Validation rules: + +- Missing `local_format` means `raw`. +- `local_format=vortex` is accepted for non-primary-key, non-vector fields. +- Primary-key fields reject `local_format=vortex`. +- Vector fields reject `local_format=vortex`. +- Unknown values are rejected. The supported values are `raw` and `vortex`. + +### Storage V3 Relationship + +`local_format` is only effective for Storage V3 sealed segments. + +For write-time column group planning, fields marked `local_format=vortex` are +partitioned away from fields using the raw local format. A column group whose +remaining fields all use Vortex local format is written with physical format +`vortex`; other column groups use the normal fallback storage format. + +For read-time loading, Vortex local format is used only when both conditions are +true: + +- all fields in the physical column group have `local_format=vortex`; +- the Storage V3 manifest says the physical column group file is Vortex. + +If either condition is false, the segment uses the existing raw loading path for +that column group. + +## Design Details + +### High-Level Architecture + +The design extends `ChunkedColumnInterface` with column-oriented scan and +positional access for sealed scalar fields. This is the main interface shift for +local format: Vortex data is not naturally owned as Milvus raw chunks, so the +new path moves callers from `ChunkedBase` chunk access to column-level +operations. + +`Scan` is one operation under `ChunkedColumnInterface`, used by expression +evaluation. Positional take/output operations are also part of the same +column-based abstraction and are used by retrieve and requery. The Vortex reader +consumes a sparse local file view behind these column-level operations. + +Milvus is in a transition state where two access families coexist: + +- `ChunkedBase` remains the raw chunk-oriented path for the existing raw local + format and existing chunk consumers. +- `ChunkedColumnInterface` is the local-format-aware path used by Vortex and by + scan/take code that should not depend on physical chunk ownership. + +Filter scan path: + +```text +Expr + -> ChunkedColumnInterface::Scan(...) + -> VortexColumn + -> VortexPlanner + -> VortexColumnGroup cache slot pin + -> VortexFormatReader::read_with_plan / read_row_ids_with_plan + -> Vortex scan builder +``` + +Retrieve/requery output path: + +```text +Retrieve output / bulk_subscript + -> ChunkedColumnInterface positional take + -> VortexColumn::Take... + -> VortexPlanner::PlanForOffsets + -> VortexColumnGroup cache slot pin + -> VortexFormatReader::read_with_plan(row indices) +``` + +The key ownership split is: + +- `milvus-storage` understands the Vortex file layout and maps row ranges, + offsets, and predicates to Vortex read plans. +- Milvus QueryNode owns cache pinning, sparse-file lifecycle, expression cursor + consumption, and output conversion. + +### Column Group Splitting + +Column group splitting partitions fields by `local_format` before subsequent +split policies finalize physical groups. This prevents raw and Vortex local +format fields from sharing one physical column group. + +The split policy behavior is: + +1. Partition pending fields by `local_format`. +2. Keep the partition's format metadata through later split policies. +3. Emit physical column groups with `Format=vortex` only for Vortex local format + groups. +4. Leave raw groups with an empty format override so they use the configured + fallback storage format. + +System, vector, text, average-size, and remanent-short split policies still +apply after the local-format partitioning. They split within the current local +format partition instead of mixing formats. + +### Cell Semantics + +A cell is the cache and loading unit for Vortex local format. Cells are defined +by the Vortex file layout and exposed through `VortexPlanner`. + +Vortex V1: + +- There is no stable row-group boundary. +- A cell corresponds to a full flat physical unit. + +Vortex V2: + +- Row groups are available. +- A cell corresponds to a complete row group and its physical segments. +- Row-group boundaries must align for fields in the same physical column group. + +General cell invariants: + +- Cell ids are contiguous and start at zero within a file. +- Cell row ranges are contiguous and cover the file. +- Cells do not share physical segments. +- All fields in the same physical column group share a `VortexColumnGroup`; pinning + a cell loads the underlying bytes once for all fields in that group. + +### Storage-Side Vortex Interfaces + +#### `VortexFooterReader` + +`VortexFooterReader` reads Vortex file metadata. It is not responsible for data +scan, take, Milvus cache pinning, or cache lifetime. + +Responsibilities: + +- Open a Vortex file through a filesystem. +- Materialize the footer into the sparse local file. +- Optionally materialize V1/V2 zone-map segments. +- Expose schema, row count, footer size, field layout, row-group ranges, and + physical byte ranges. +- Prune row groups using zone maps when they are loaded. + +Lifecycle: + +- `Open(fs, load_zonemap)` succeeds at most once per reader instance. +- `Open(false)` loads footer metadata only; pruning conservatively keeps all + candidate row groups. +- `Open(true)` loads footer metadata, materializes zone-map bytes, and then + reopens the final Vortex file view so Vortex's internal initial-read cache + cannot retain sparse zero-filled zone-map bytes. + +#### `VortexPlanner` + +`VortexPlanner` converts logical Milvus access requests into two outputs: + +```cpp +struct VortexPlan { + std::vector cell_ids; + VortexReadPlan read_plan; +}; +``` + +- `cell_ids` are used by Milvus to pin Vortex cells through the cache layer. +- `read_plan` is passed to `VortexFormatReader` for execution. + +Supported planning modes: + +- `PlanForRowRange(row_start, row_end, predicate)` +- `PlanForOffsets(offsets)` + +For V2 row-group cells and supported predicates, the planner may use zone maps +to prune cells. For V1 files or unsupported predicates, it returns all candidate +cells conservatively. + +#### `VortexFormatReader` + +The common `FormatReader` interface remains compatible with existing callers. +Vortex local format uses Vortex-specific extensions: + +- `read_with_plan(const VortexReadPlan&)` +- `read_row_ids_with_plan(const VortexReadPlan&)` + +`read_with_plan` returns data as an Arrow stream. `read_row_ids_with_plan` +returns file-local row ids satisfying the predicate in the plan. Predicate state +is carried by `VortexReadPlan`, not by long-lived reader state. Existing +`set_predicate` behavior remains for compatibility but is not the local format +path. + +### Milvus-Side Components + +#### `FieldMeta` + +`FieldMeta` parses `type_params["local_format"]` and defaults to `raw`. It also +serializes non-default local format back to the field schema. + +#### `ChunkedColumnInterface` + +`ChunkedColumnInterface` is the shared access contract for column-oriented scalar +data. It lets callers express the operation they need without assuming the data +is backed by raw Milvus chunks. + +The interface covers two operation groups: + +- scan operations for expression evaluation; +- positional take/output operations for retrieve, requery, and bulk_subscript. + +`Scan` returns a cursor of `ScanBatch` values. + +Scan outputs: + +| Output | Payload | +|--------|---------| +| `ScanOutput::RowIds` | Sparse row ids that satisfy, or may satisfy, a pushed predicate. | +| `ScanOutput::Data` | Dense values over a row range, plus validity when needed. | + +Data scan supports: + +- row range; +- value kind (`FixedWidth`, `StringView`, `JsonView`, `ArrayView`); +- validity; +- validity-only projection. + +Row-id scan supports: + +- unary predicates; +- binary range predicates; +- sparse row-id batches. + +If a column implementation cannot support a scan mode, it falls back to the +raw-compatible behavior through the existing chunked path. + +#### `VortexColumnGroup` + +`VortexColumnGroup` owns shared state for one physical Vortex column group. + +Each file state contains: + +- source filesystem and resolved source path; +- sparse filesystem and sparse path; +- `VortexFooterReader`; +- group-level `VortexPlanner`; +- cache slot and translator; +- row count and memory accounting. + +All fields in the same physical group share the same `VortexColumnGroup`. + +#### `VortexColumn` + +`VortexColumn` is a field-level `ChunkedColumnInterface` implementation over a +shared `VortexColumnGroup`. + +Responsibilities: + +- Resolve the Vortex field name. External fields use the external column name; + internal fields use the field id string. +- Build a field-level projected Arrow schema. +- Create field-level planner/reader state. +- Implement `Scan`. +- Implement positional take helpers for retrieve output. + +### Filter Scan + +#### Predicate Pushdown + +For supported unary and binary range expressions, expression execution requests +`ScanOutput::RowIds`. + +Example: + +```text +UnaryExpr / BinaryRangeExpr + -> ChunkedColumnInterface::Scan(ScanOutput::RowIds) + -> VortexColumn::Scan + -> VortexRowIdScanCursor + -> VortexPlanner::PlanForRowRange(predicate) + -> pin planned cells + -> VortexFormatReader::read_row_ids_with_plan + -> bitmap assembly in expression execution +``` + +The initial implementation supports a narrow set of predicate strings that can +be represented safely for the Vortex reader. Unsupported expressions fall back +to data scan. This keeps correctness independent of pushdown coverage. + +#### Data Scan + +Unsupported predicates, complex expressions, and expressions that need raw value +inspection use `ScanOutput::Data`. + +Example: + +```text +Expr data path + -> ChunkedColumnInterface::Scan(ScanOutput::Data) + -> VortexDataScanCursor + -> VortexPlanner::PlanForRowRange(no predicate) + -> pin planned cells + -> VortexFormatReader::read_with_plan + -> expression layer evaluates predicate +``` + +This is the current path for examples such as `LIKE`, `IN`, JSON path +expressions, and array predicates when they cannot be represented as a Vortex +predicate. + +### Offset Input Execution + +Offset-input execution is used when expression evaluation is restricted to a +known set of segment offsets. + +The initial Vortex local format implementation handles dense sorted offsets by +scanning one continuous range: + +```text +ProcessDataByOffsets + -> ProcessSortedDataByOffsetsByScan + -> scan [min_offset, max_offset + 1) + -> expression layer checks the offset bitmap +``` + +Bitmap or selection pushdown into `ChunkedColumnInterface::Scan` is left as +future work. The current strategy avoids many small reads while keeping +semantics simple. + +### Retrieve and Requery + +Retrieve/requery output is not filter scan. It reads requested output fields at +selected row offsets. + +```text +FillTargetEntry / Retrieve output + -> bulk_subscript + -> ChunkedColumnInterface positional take + -> VortexColumn::BulkPrimitiveValueAt / BulkRawStringAt / BulkArrayAt + -> TakeOwn / TakeStringLikeViews + -> VortexPlanner::PlanForOffsets + -> pin planned cells + -> VortexFormatReader::read_with_plan(row indices) + -> restore requested output order +``` + +The planner disables predicate semantics for take because retrieve output is +positional. Random requery over long strings can still touch many cells; this is +tracked as a separate performance area from filter pushdown. + +### Nullable and Validity + +The `ChunkedColumnInterface` scan API uses `ValidityView` to present nullability +uniformly. + +Rules: + +- Non-nullable fields may return all-valid validity. +- Nullable dense data scans must return validity aligned with the dense row + range. +- Row-id scans may return validity aligned with sparse row ids. +- Validity-only projection is part of the scan model so callers that only need + nullability do not need to materialize full values. + +The raw path may adapt its existing `bool*` validity representation into this +model. Vortex uses Arrow bitmap/null-buffer semantics. + +### Sparse Local File and Cache Loading + +Vortex local format uses a sparse local file as the file view consumed by the +Vortex reader. + +Flow: + +```text +VortexFooterReader + -> materialize footer and optional zone-map bytes into sparse file + +VortexPlanner + -> choose cell ids and read plan + +Milvus cache layer + -> pin cells + -> Vortex translator loads cell byte ranges into sparse file + +VortexFormatReader + -> reads the sparse file as a normal file +``` + +Properties: + +- Loaded byte ranges are written to the sparse file. +- Missing ranges remain sparse holes and read as zero-filled bytes if an + over-wide read crosses them. +- Footer bytes are always materialized before planning. +- Zone-map bytes are materialized when pruning is enabled. +- Cell lifecycle remains controlled by Milvus cache pin/unpin. + +### Warmup and Eviction + +Vortex local format reuses Milvus cache warmup policy. Warmed scalar fields can +load their cells during segment load, reducing the first-query penalty. Manual +eviction and warmup cancellation are implemented at the `VortexColumnGroup` +level so all field proxies in the same physical group share the same state. + +## Compatibility, Deprecation, and Migration Plan + +- Backward compatible by default: fields without `local_format` behave as + `raw`. +- Existing non-Vortex segments continue to load through the raw path. +- A schema can contain both raw and Vortex local format fields; column group + splitting keeps them physically separate. +- During the transition, QueryNode keeps both access paths: raw fields continue + to use the `ChunkedBase` chunk-oriented path, while Vortex fields use the + `ChunkedColumnInterface` column-oriented path. +- Vortex local format is only used for Storage V3 sealed segments. +- Rolling upgrades must ensure QueryNodes understand Vortex local format before + new Vortex column groups are loaded. Older readers cannot load Vortex physical + column groups. + +## Test Plan + +System and integration validation: + +- Create collections with raw fields, Vortex local format fields, and mixed + fields; verify insert, flush, load, search, query, and retrieve. +- Verify `local_format=vortex` is rejected for primary-key and vector fields. +- Verify Storage V3 manifests with Vortex physical column groups load as + `VortexColumnGroup + VortexColumn`. +- Verify non-Vortex physical files continue to load through the raw path. +- Run scalar filter benchmark cases for primitive predicates, complex + expressions, offset-input execution, and retrieve/requery output. + +Unit and component validation: + +- `FieldMeta` parse/serialize of `local_format`. +- Column group split policy keeps raw and Vortex fields separate. +- `ChunkedColumnInterface` scan and positional take behavior. +- Raw `ChunkedBase` path and Vortex `ChunkedColumnInterface` path coexist without + changing raw field behavior. +- `VortexColumn` row-id scan, data scan, validity, and take paths. +- `VortexFooterReader` footer and optional zone-map lifecycle. +- `VortexPlanner` row range, offset, and predicate pruning plans. +- `VortexFormatReader::read_with_plan` and `read_row_ids_with_plan`. + +Performance validation: + +- Compare Vortex and raw local format for cold and hot retrieve. +- Compare Vortex and raw local format for primitive filter scan. +- Track complex data scan cases such as JSON, ARRAY, and `LIKE`. +- Track random retrieve/requery over long VARCHAR because it exercises take and + output conversion rather than filter scan. + +## Future Work + +- Push offset bitmaps or row selections into `ChunkedColumnInterface::Scan`. +- Expand Vortex predicate construction for more scalar types and expression + forms. +- Optimize long string, JSON, and ARRAY retrieve/take conversion paths. +- Use validity-only scan in paths that only need nullability. + +## References + +- [Milvus PR: support vortex local format](https://github.com/milvus-io/milvus/pull/49908) +- [Milvus issue: vortex local format](https://github.com/milvus-io/milvus/issues/50304) +- [milvus-storage](https://github.com/milvus-io/milvus-storage) +- [Vortex project](https://github.com/vortex-data/vortex) diff --git a/internal/core/src/common/Consts.h b/internal/core/src/common/Consts.h index 57183e8ef5..974c65ed6d 100644 --- a/internal/core/src/common/Consts.h +++ b/internal/core/src/common/Consts.h @@ -154,6 +154,10 @@ const std::string ELEMENT_TYPE_KEY_FOR_ARROW = "elementType"; const float EPSILON = 0.0000000119; const std::string NAMESPACE_FIELD_NAME = "$namespace_id"; const std::string MMAP_ENABLED_KEY = "mmap.enabled"; +constexpr const char* LOCAL_FORMAT_KEY = "local_format"; +constexpr const char* LOCAL_FORMAT_RAW = "raw"; +constexpr const char* LOCAL_FORMAT_VORTEX = "vortex"; +constexpr const char* STORAGE_FORMAT_VORTEX = "vortex"; const int64_t LOGICAL_BITS = 18; // Warmup policy keys diff --git a/internal/core/src/common/FieldMeta.cpp b/internal/core/src/common/FieldMeta.cpp index e7fe44f914..92e17b1de3 100644 --- a/internal/core/src/common/FieldMeta.cpp +++ b/internal/core/src/common/FieldMeta.cpp @@ -122,6 +122,7 @@ FieldMeta::ToProto() const { if (string_info_.has_value()) { params = string_info_->params; } + params.erase(LOCAL_FORMAT_KEY); params[MAX_LENGTH] = std::to_string(get_max_len()); params["enable_match"] = enable_match() ? "true" : "false"; params["enable_analyzer"] = enable_analyzer() ? "true" : "false"; @@ -132,6 +133,10 @@ FieldMeta::ToProto() const { // element_type already populated above } + if (local_format_ != LOCAL_FORMAT_RAW) { + add_type_param(LOCAL_FORMAT_KEY, local_format_); + } + return proto; } @@ -166,10 +171,17 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { return schema_proto.default_value(); }(); + auto type_map = RepeatedKeyValToMap(schema_proto.type_params()); + auto local_format = [&]() -> std::string { + if (auto it = type_map.find(LOCAL_FORMAT_KEY); it != type_map.end()) { + return it->second; + } + return LOCAL_FORMAT_RAW; + }; + if (data_type == DataType::VECTOR_ARRAY) { // todo(SpadeA): revisit the code when index build for vector array is ready int64_t dim = 0; - auto type_map = RepeatedKeyValToMap(schema_proto.type_params()); AssertInfo(type_map.count("dim"), "dim not found"); dim = boost::lexical_cast(type_map.at("dim")); @@ -180,13 +192,13 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { dim, std::nullopt, nullable, - external_field_mapping}; + external_field_mapping, + local_format()}; } if (IsVectorDataType(data_type)) { AssertInfo(!default_value.has_value(), "vector fields do not support default values"); - auto type_map = RepeatedKeyValToMap(schema_proto.type_params()); auto index_map = RepeatedKeyValToMap(schema_proto.index_params()); int64_t dim = 0; @@ -203,7 +215,8 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { std::nullopt, nullable, default_value, - external_field_mapping}; + external_field_mapping, + local_format()}; } auto metric_type = index_map.at("metric_type"); return FieldMeta{name, @@ -213,11 +226,11 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { metric_type, nullable, default_value, - external_field_mapping}; + external_field_mapping, + local_format()}; } if (IsStringDataType(data_type)) { - auto type_map = RepeatedKeyValToMap(schema_proto.type_params()); int64_t max_len = 0; if (type_map.count(MAX_LENGTH)) { max_len = boost::lexical_cast(type_map.at(MAX_LENGTH)); @@ -243,6 +256,8 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { bool enable_analyzer = get_bool_value("enable_analyzer"); bool enable_match = get_bool_value("enable_match"); + auto string_params = type_map; + string_params.erase(LOCAL_FORMAT_KEY); return FieldMeta{name, field_id, @@ -251,9 +266,10 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { nullable, enable_match, enable_analyzer, - type_map, + string_params, default_value, - external_field_mapping}; + external_field_mapping, + local_format()}; } if (IsArrayDataType(data_type)) { @@ -263,7 +279,8 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { DataType(schema_proto.element_type()), nullable, default_value, - external_field_mapping}; + external_field_mapping, + local_format()}; } return FieldMeta{name, @@ -271,7 +288,8 @@ FieldMeta::ParseFrom(const milvus::proto::schema::FieldSchema& schema_proto) { data_type, nullable, default_value, - external_field_mapping}; + external_field_mapping, + local_format()}; } } // namespace milvus diff --git a/internal/core/src/common/FieldMeta.h b/internal/core/src/common/FieldMeta.h index 32b7f5e6b3..10c8f35019 100644 --- a/internal/core/src/common/FieldMeta.h +++ b/internal/core/src/common/FieldMeta.h @@ -51,13 +51,15 @@ class FieldMeta { DataType type, bool nullable, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), nullable_(nullable), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(!IsVectorDataType(type_)); } @@ -67,14 +69,16 @@ class FieldMeta { int64_t max_length, bool nullable, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), nullable_(nullable), string_info_(StringInfo{max_length}), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(IsStringDataType(type_)); } @@ -87,7 +91,8 @@ class FieldMeta { bool enable_analyzer, std::map& params, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), @@ -99,7 +104,8 @@ class FieldMeta { std::move(params), }), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(IsStringDataType(type_)); } @@ -109,14 +115,16 @@ class FieldMeta { DataType element_type, bool nullable, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), element_type_(element_type), nullable_(nullable), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(IsArrayDataType(type_)); } @@ -129,14 +137,16 @@ class FieldMeta { std::optional metric_type, bool nullable, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), nullable_(nullable), vector_info_(VectorInfo{dim, std::move(metric_type)}), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(IsVectorDataType(type_)); Assert(!default_value_.has_value() && "vector fields do not support default values"); @@ -150,14 +160,16 @@ class FieldMeta { int64_t dim, std::optional metric_type, bool nullable, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), type_(type), nullable_(nullable), element_type_(element_type), vector_info_(VectorInfo{dim, std::move(metric_type)}), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(type_ == DataType::VECTOR_ARRAY); Assert(IsVectorDataType(element_type_)); } @@ -170,14 +182,16 @@ class FieldMeta { DataType type, bool nullable, std::optional default_value, - std::string external_field_mapping = "") + std::string external_field_mapping = "", + std::string local_format = LOCAL_FORMAT_RAW) : name_(std::move(name)), id_(id), main_field_id_(main_field_id), type_(type), nullable_(nullable), default_value_(std::move(default_value)), - external_field_mapping_(std::move(external_field_mapping)) { + external_field_mapping_(std::move(external_field_mapping)), + local_format_(std::move(local_format)) { Assert(!IsVectorDataType(type_)); } @@ -296,6 +310,11 @@ class FieldMeta { external_field_mapping_ = external_field; } + const std::string& + get_local_format() const { + return local_format_; + } + milvus::proto::schema::FieldSchema ToProto() const; @@ -349,6 +368,7 @@ class FieldMeta { // of collection schema, the field id is the json shredding field id int64_t main_field_id_ = INVALID_FIELD_ID; std::string external_field_mapping_; + std::string local_format_ = LOCAL_FORMAT_RAW; }; } // namespace milvus diff --git a/internal/core/src/common/FieldMetaTest.cpp b/internal/core/src/common/FieldMetaTest.cpp index abcb42368e..9c43d3b846 100644 --- a/internal/core/src/common/FieldMetaTest.cpp +++ b/internal/core/src/common/FieldMetaTest.cpp @@ -84,6 +84,52 @@ TEST(FieldMetaTest, ParseFromWithoutExternalField) { EXPECT_TRUE(field.get_external_field_mapping().empty()); } +TEST(FieldMetaTest, LocalFormatRoundTrip) { + milvus::proto::schema::FieldSchema proto; + proto.set_fieldid(202); + proto.set_name("vortex_varchar"); + proto.set_data_type(milvus::proto::schema::DataType::VarChar); + proto.set_nullable(true); + auto* max_length = proto.add_type_params(); + max_length->set_key(MAX_LENGTH); + max_length->set_value("128"); + auto* local_format = proto.add_type_params(); + local_format->set_key(LOCAL_FORMAT_KEY); + local_format->set_value(LOCAL_FORMAT_VORTEX); + + auto field = FieldMeta::ParseFrom(proto); + EXPECT_EQ(field.get_local_format(), LOCAL_FORMAT_VORTEX); + + auto serialized = field.ToProto(); + int local_format_count = 0; + for (const auto& param : serialized.type_params()) { + if (param.key() == LOCAL_FORMAT_KEY) { + ++local_format_count; + EXPECT_EQ(param.value(), LOCAL_FORMAT_VORTEX); + } + } + EXPECT_EQ(local_format_count, 1); + + auto reparsed = FieldMeta::ParseFrom(serialized); + EXPECT_EQ(reparsed.get_local_format(), LOCAL_FORMAT_VORTEX); + EXPECT_EQ(reparsed.get_max_len(), 128); +} + +TEST(FieldMetaTest, RawLocalFormatIsDefaultAndNotSerialized) { + milvus::proto::schema::FieldSchema proto; + proto.set_fieldid(203); + proto.set_name("raw_scalar"); + proto.set_data_type(milvus::proto::schema::DataType::Int64); + + auto field = FieldMeta::ParseFrom(proto); + EXPECT_EQ(field.get_local_format(), LOCAL_FORMAT_RAW); + + auto serialized = field.ToProto(); + for (const auto& param : serialized.type_params()) { + EXPECT_NE(param.key(), LOCAL_FORMAT_KEY); + } +} + TEST(FieldMetaTest, ShouldLoadFieldReturnsFalseForExternalField) { auto schema = std::make_shared(); diff --git a/internal/flushcommon/syncmgr/task.go b/internal/flushcommon/syncmgr/task.go index b3f0e85277..35c434178a 100644 --- a/internal/flushcommon/syncmgr/task.go +++ b/internal/flushcommon/syncmgr/task.go @@ -234,7 +234,7 @@ func resolveColumnGroups(segmentInfo *metacache.SegmentInfo, schema *schemapb.Co for _, cg := range currentSplit { // legacy split found, use legacy policy if len(cg.Fields) == 0 { - result := storagecommon.SplitColumns(allFields, map[int64]storagecommon.ColumnStats{}, storagecommon.NewSelectedDataTypePolicy(), storagecommon.NewRemanentShortPolicy(-1)) + result := storagecommon.SplitColumns(allFields, map[int64]storagecommon.ColumnStats{}, storagecommon.NewLocalFormatPolicy(), storagecommon.NewSelectedDataTypePolicy(), storagecommon.NewRemanentShortPolicy(-1)) result = storagecommon.FillColumnGroupFormats(result, paramtable.Get().DataNodeCfg.StorageFormat.GetValue()) mlog.Info(context.TODO(), "use legacy split policy", mlog.FieldSegmentID(segmentID), mlog.Stringers("columnGroups", result)) return result diff --git a/internal/rootcoord/create_collection_task_test.go b/internal/rootcoord/create_collection_task_test.go index b42cad1ba2..1b45a50dff 100644 --- a/internal/rootcoord/create_collection_task_test.go +++ b/internal/rootcoord/create_collection_task_test.go @@ -349,6 +349,32 @@ func Test_createCollectionTask_validateSchema(t *testing.T) { assert.Error(t, err) }) + t.Run("primary field rejects vortex local format", func(t *testing.T) { + collectionName := funcutil.GenRandomStr() + task := createCollectionTask{ + Req: &milvuspb.CreateCollectionRequest{ + Base: &commonpb.MsgBase{MsgType: commonpb.MsgType_CreateCollection}, + CollectionName: collectionName, + }, + } + schema := &schemapb.CollectionSchema{ + Name: collectionName, + Fields: []*schemapb.FieldSchema{ + { + Name: "pk", + DataType: schemapb.DataType_Int64, + IsPrimaryKey: true, + TypeParams: []*commonpb.KeyValuePair{ + {Key: common.LocalFormatKey, Value: common.LocalFormatVortex}, + }, + }, + }, + } + err := task.validateSchema(context.TODO(), schema) + assert.Error(t, err) + assert.Contains(t, err.Error(), "local_format vortex is not supported for primary key field") + }) + t.Run("has system fields", func(t *testing.T) { collectionName := funcutil.GenRandomStr() task := createCollectionTask{ diff --git a/internal/rootcoord/util.go b/internal/rootcoord/util.go index e2a78c04e4..a3252691dd 100644 --- a/internal/rootcoord/util.go +++ b/internal/rootcoord/util.go @@ -446,6 +446,9 @@ func checkFieldSchema(fieldSchemas []*schemapb.FieldSchema) error { if err := checkDupKvPairs(fieldSchema.GetTypeParams(), "type"); err != nil { return err } + if err := validateLocalFormat(fieldSchema); err != nil { + return err + } if err := checkDupKvPairs(fieldSchema.GetIndexParams(), "index"); err != nil { return err } @@ -479,6 +482,9 @@ func checkStructArrayFieldSchema(schemas []*schemapb.StructArrayFieldSchema) err if err := checkDupKvPairs(field.GetTypeParams(), "type"); err != nil { return err } + if err := validateLocalFormat(field); err != nil { + return err + } if err := checkDupKvPairs(field.GetIndexParams(), "index"); err != nil { return err } @@ -547,6 +553,34 @@ func checkDupKvPairs(params []*commonpb.KeyValuePair, paramType string) error { return nil } +func validateLocalFormat(fieldSchema *schemapb.FieldSchema) error { + for _, kv := range fieldSchema.GetTypeParams() { + if kv.GetKey() == common.LocalFormatKey { + switch kv.GetValue() { + case common.LocalFormatRaw: + // valid + case common.LocalFormatVortex: + if fieldSchema.GetIsPrimaryKey() { + return merr.WrapErrParameterInvalidMsg( + "local_format vortex is not supported for primary key field '%s'", + fieldSchema.GetName()) + } + if typeutil.IsVectorType(fieldSchema.GetDataType()) { + return merr.WrapErrParameterInvalidMsg( + "local_format vortex is not supported for vector field '%s'", + fieldSchema.GetName()) + } + default: + return merr.WrapErrParameterInvalidMsg( + "invalid local_format '%s' for field '%s', supported: raw, vortex", + kv.GetValue(), fieldSchema.GetName()) + } + break + } + } + return nil +} + func validateFieldDataType(fieldSchemas []*schemapb.FieldSchema) error { for _, field := range fieldSchemas { if _, ok := schemapb.DataType_name[int32(field.GetDataType())]; !ok || field.GetDataType() == schemapb.DataType_None { diff --git a/internal/rootcoord/util_test.go b/internal/rootcoord/util_test.go index bb3226f9ef..fd2feb798b 100644 --- a/internal/rootcoord/util_test.go +++ b/internal/rootcoord/util_test.go @@ -760,3 +760,67 @@ func Test_updateMaxFieldIDProperty(t *testing.T) { assert.Equal(t, "103", result[1].Value) }) } + +func TestValidateLocalFormat(t *testing.T) { + tests := []struct { + name string + field *schemapb.FieldSchema + errorText string + }{ + { + name: "raw scalar", + field: &schemapb.FieldSchema{ + Name: "raw_scalar", + DataType: schemapb.DataType_Int64, + TypeParams: []*commonpb.KeyValuePair{{Key: common.LocalFormatKey, Value: common.LocalFormatRaw}}, + }, + }, + { + name: "vortex scalar", + field: &schemapb.FieldSchema{ + Name: "vortex_scalar", + DataType: schemapb.DataType_VarChar, + TypeParams: []*commonpb.KeyValuePair{{Key: common.LocalFormatKey, Value: common.LocalFormatVortex}}, + }, + }, + { + name: "invalid value", + field: &schemapb.FieldSchema{ + Name: "invalid", + DataType: schemapb.DataType_Int64, + TypeParams: []*commonpb.KeyValuePair{{Key: common.LocalFormatKey, Value: "unknown"}}, + }, + errorText: "invalid local_format", + }, + { + name: "primary key", + field: &schemapb.FieldSchema{ + Name: "pk", + DataType: schemapb.DataType_Int64, + IsPrimaryKey: true, + TypeParams: []*commonpb.KeyValuePair{{Key: common.LocalFormatKey, Value: common.LocalFormatVortex}}, + }, + errorText: "not supported for primary key", + }, + { + name: "vector field", + field: &schemapb.FieldSchema{ + Name: "vector", + DataType: schemapb.DataType_FloatVector, + TypeParams: []*commonpb.KeyValuePair{{Key: common.LocalFormatKey, Value: common.LocalFormatVortex}}, + }, + errorText: "not supported for vector field", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := validateLocalFormat(test.field) + if test.errorText == "" { + assert.NoError(t, err) + return + } + assert.ErrorContains(t, err, test.errorText) + }) + } +} diff --git a/internal/storagecommon/column_group_splitter.go b/internal/storagecommon/column_group_splitter.go index 05fb171ea5..8265c858d6 100644 --- a/internal/storagecommon/column_group_splitter.go +++ b/internal/storagecommon/column_group_splitter.go @@ -79,7 +79,8 @@ func SplitColumns(fields []*schemapb.FieldSchema, stats map[int64]ColumnStats, p func DefaultPolicies() []ColumnGroupSplitPolicy { paramtable.Init() - result := make([]ColumnGroupSplitPolicy, 0, 4) + result := make([]ColumnGroupSplitPolicy, 0, 5) + result = append(result, NewLocalFormatPolicy()) if paramtable.Get().CommonCfg.Stv2SplitSystemColumn.GetAsBool() { result = append(result, NewSystemColumnPolicy(paramtable.Get().CommonCfg.Stv2SystemColumnIncludePK.GetAsBool(), paramtable.Get().CommonCfg.Stv2SystemColumnIncludePartitionKey.GetAsBool(), diff --git a/internal/storagecommon/split_policy.go b/internal/storagecommon/split_policy.go index 3939217b70..b9cdf510d2 100644 --- a/internal/storagecommon/split_policy.go +++ b/internal/storagecommon/split_policy.go @@ -37,19 +37,50 @@ type currentSplit struct { nextGroupID int64 outputGroups []ColumnGroup processFields typeutil.Set[int64] + pendingGroups []localFormatGroup } func newCurrentSplit(fields []*schemapb.FieldSchema, stats map[int64]ColumnStats) *currentSplit { + pendingGroup := localFormatGroup{ + fields: make([]int64, 0, len(fields)), + indices: make([]int, 0, len(fields)), + localFormat: "", + } + for idx, field := range fields { + pendingGroup.fields = append(pendingGroup.fields, field.GetFieldID()) + pendingGroup.indices = append(pendingGroup.indices, idx) + } return ¤tSplit{ fields: fields, stats: stats, processFields: typeutil.NewSet[int64](), + pendingGroups: []localFormatGroup{pendingGroup}, } } func (c *currentSplit) SplitFields(groupID int64, fields []int64, indices []int) { + c.SplitFieldsWithFormat(groupID, fields, indices, c.columnGroupFormat(indices)) +} + +func (c *currentSplit) SplitFieldsWithFormat(groupID int64, fields []int64, indices []int, format string) { c.processFields.Insert(fields...) - c.outputGroups = append(c.outputGroups, ColumnGroup{Columns: indices, GroupID: groupID, Fields: fields}) + c.outputGroups = append(c.outputGroups, ColumnGroup{Columns: indices, GroupID: groupID, Fields: fields, Format: format}) +} + +func (c *currentSplit) columnGroupFormat(indices []int) string { + if len(indices) == 0 { + return "" + } + format := fieldLocalFormat(c.fields[indices[0]]) + if format == common.LocalFormatRaw { + return "" + } + for _, idx := range indices[1:] { + if fieldLocalFormat(c.fields[idx]) != format { + return "" + } + } + return storageFormatForLocalFormat(format) } func (c *currentSplit) NextGroupID() int64 { @@ -63,20 +94,78 @@ func (c *currentSplit) Processed(field int64) bool { } func (c *currentSplit) Range(f func(idx int, field *schemapb.FieldSchema)) { - for idx, field := range c.fields { - if c.Processed(field.GetFieldID()) { - continue + for _, group := range c.RangeGroups(nil) { + for _, idx := range group.indices { + f(idx, c.fields[idx]) } - f(idx, field) } } +func (c *currentSplit) RangeGroups(match func(*schemapb.FieldSchema) bool) []localFormatGroup { + pendingGroups := c.pendingGroups + if len(pendingGroups) == 0 { + pendingGroups = []localFormatGroup{{}} + for idx, field := range c.fields { + pendingGroups[0].fields = append(pendingGroups[0].fields, field.GetFieldID()) + pendingGroups[0].indices = append(pendingGroups[0].indices, idx) + } + } + + groups := make([]localFormatGroup, 0, len(pendingGroups)) + for _, pendingGroup := range pendingGroups { + group := localFormatGroup{ + fields: make([]int64, 0, len(pendingGroup.fields)), + indices: make([]int, 0, len(pendingGroup.indices)), + localFormat: pendingGroup.localFormat, + } + for _, idx := range pendingGroup.indices { + field := c.fields[idx] + if c.Processed(field.GetFieldID()) { + continue + } + if match != nil && !match(field) { + continue + } + group.fields = append(group.fields, field.GetFieldID()) + group.indices = append(group.indices, idx) + } + if len(group.fields) > 0 { + groups = append(groups, group) + } + } + return groups +} + +func (c *currentSplit) PartitionRemainingByLocalFormat() { + nextGroups := make([]localFormatGroup, 0, len(c.pendingGroups)) + for _, pendingGroup := range c.RangeGroups(nil) { + groupsByFormat := make(map[string]*localFormatGroup) + formats := make([]string, 0, 2) + for _, idx := range pendingGroup.indices { + field := c.fields[idx] + format := fieldLocalFormat(field) + group := groupsByFormat[format] + if group == nil { + formats = append(formats, format) + group = &localFormatGroup{localFormat: format} + groupsByFormat[format] = group + } + group.fields = append(group.fields, field.GetFieldID()) + group.indices = append(group.indices, idx) + } + for _, format := range formats { + nextGroups = append(nextGroups, *groupsByFormat[format]) + } + } + c.pendingGroups = nextGroups +} + // ColumnGroupSplitPolicy interface for column group split policy. type ColumnGroupSplitPolicy interface { Split(currentSplit *currentSplit) *currentSplit } -// selectedDataTypePolicy split widt datatype (vector, text) to a new column group. +// selectedDataTypePolicy splits wide data types (vector, text) to new column groups. type selectedDataTypePolicy struct{} func (p *selectedDataTypePolicy) Split(currentSplit *currentSplit) *currentSplit { @@ -93,6 +182,39 @@ func NewSelectedDataTypePolicy() ColumnGroupSplitPolicy { return &selectedDataTypePolicy{} } +type localFormatPolicy struct{} + +type localFormatGroup struct { + fields []int64 + indices []int + localFormat string +} + +func fieldLocalFormat(field *schemapb.FieldSchema) string { + for _, kv := range field.GetTypeParams() { + if kv.GetKey() == common.LocalFormatKey { + return kv.GetValue() + } + } + return common.LocalFormatRaw +} + +func storageFormatForLocalFormat(format string) string { + if format == common.LocalFormatVortex { + return common.LocalFormatVortex + } + return "" +} + +func (p *localFormatPolicy) Split(currentSplit *currentSplit) *currentSplit { + currentSplit.PartitionRemainingByLocalFormat() + return currentSplit +} + +func NewLocalFormatPolicy() ColumnGroupSplitPolicy { + return &localFormatPolicy{} +} + // systemColumnPolicy split system columns to a new column group // if includePK is true, system columns include primary key column. type systemColumnPolicy struct { @@ -110,20 +232,16 @@ func NewSystemColumnPolicy(includePK bool, includePartKey bool, includeClusterin } func (p *systemColumnPolicy) Split(currentSplit *currentSplit) *currentSplit { - systemFields := make([]int64, 0, 3) - systemFieldIndices := make([]int, 0, 3) - - currentSplit.Range(func(idx int, field *schemapb.FieldSchema) { - if field.GetFieldID() < common.StartOfUserFieldID || + groups := currentSplit.RangeGroups(func(field *schemapb.FieldSchema) bool { + return field.GetFieldID() < common.StartOfUserFieldID || (p.includePrimaryKey && field.GetIsPrimaryKey()) || (p.includePartitionKey && field.GetIsPartitionKey()) || - (p.includeClusteringKey && field.GetIsClusteringKey()) { - systemFields = append(systemFields, field.GetFieldID()) - systemFieldIndices = append(systemFieldIndices, idx) - } + (p.includeClusteringKey && field.GetIsClusteringKey()) }) - currentSplit.SplitFields(currentSplit.NextGroupID(), systemFields, systemFieldIndices) + for _, group := range groups { + currentSplit.SplitFields(currentSplit.NextGroupID(), group.fields, group.indices) + } return currentSplit } @@ -137,21 +255,22 @@ func NewRemanentShortPolicy(maxGroupSize int) ColumnGroupSplitPolicy { } func (p *remanentShortPolicy) Split(currentSplit *currentSplit) *currentSplit { - var shortFields []int64 - var shortFieldIndices []int - - currentSplit.Range(func(idx int, field *schemapb.FieldSchema) { - shortFields = append(shortFields, field.GetFieldID()) - shortFieldIndices = append(shortFieldIndices, idx) - if p.maxGroupSize > 0 && len(shortFields) >= p.maxGroupSize { - currentSplit.SplitFields(currentSplit.NextGroupID(), shortFields, shortFieldIndices) - shortFields = make([]int64, 0, p.maxGroupSize) - shortFieldIndices = make([]int, 0, p.maxGroupSize) + for _, group := range currentSplit.RangeGroups(nil) { + shortFields := make([]int64, 0, len(group.fields)) + shortFieldIndices := make([]int, 0, len(group.indices)) + for i, fieldID := range group.fields { + shortFields = append(shortFields, fieldID) + shortFieldIndices = append(shortFieldIndices, group.indices[i]) + if p.maxGroupSize > 0 && len(shortFields) >= p.maxGroupSize { + currentSplit.SplitFields(currentSplit.NextGroupID(), shortFields, shortFieldIndices) + shortFields = make([]int64, 0, p.maxGroupSize) + shortFieldIndices = make([]int, 0, p.maxGroupSize) + } } - }) - if len(shortFields) > 0 { - currentSplit.SplitFields(currentSplit.NextGroupID(), shortFields, shortFieldIndices) + if len(shortFields) > 0 { + currentSplit.SplitFields(currentSplit.NextGroupID(), shortFields, shortFieldIndices) + } } return currentSplit diff --git a/internal/storagecommon/split_policy_test.go b/internal/storagecommon/split_policy_test.go index 9d5c7841ee..2750567595 100644 --- a/internal/storagecommon/split_policy_test.go +++ b/internal/storagecommon/split_policy_test.go @@ -21,7 +21,9 @@ import ( "github.com/stretchr/testify/assert" + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" + "github.com/milvus-io/milvus/pkg/v3/common" "github.com/milvus-io/milvus/pkg/v3/util/typeutil" ) @@ -39,6 +41,17 @@ func AssertSplitEqual(t *testing.T, expect, actual *currentSplit) { assert.Equal(t, expect.outputGroups[i].GroupID, actual.outputGroups[i].GroupID) assert.Equal(t, expect.outputGroups[i].Columns, actual.outputGroups[i].Columns) assert.Equal(t, expect.outputGroups[i].Fields, actual.outputGroups[i].Fields) + assert.Equal(t, expect.outputGroups[i].Format, actual.outputGroups[i].Format) + } +} + +func AssertPendingGroupsEqual(t *testing.T, expect []ColumnGroup, actual *currentSplit) { + groups := actual.RangeGroups(nil) + assert.Equal(t, len(expect), len(groups)) + for i := range expect { + assert.Equal(t, expect[i].Columns, groups[i].indices) + assert.Equal(t, expect[i].Fields, groups[i].fields) + assert.Equal(t, expect[i].Format, storageFormatForLocalFormat(groups[i].localFormat)) } } @@ -49,6 +62,15 @@ func TestWideDataTypePolicy(t *testing.T) { expect *currentSplit } + localFormatParam := func(format string) []*commonpb.KeyValuePair { + return []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: format, + }, + } + } + cases := []testCase{ { tag: "float_vector", @@ -77,6 +99,31 @@ func TestWideDataTypePolicy(t *testing.T) { }, }, }, + { + tag: "text_with_vortex_local_format", + input: newCurrentSplit([]*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 101, + DataType: schemapb.DataType_Text, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + }, nil), + expect: ¤tSplit{ + processFields: typeutil.NewSet[int64](101), + outputGroups: []ColumnGroup{ + { + GroupID: 101, + Columns: []int{1}, + Fields: []int64{101}, + Format: common.LocalFormatVortex, + }, + }, + }, + }, { tag: "text_with_processed_group", input: ¤tSplit{ @@ -137,6 +184,224 @@ func TestWideDataTypePolicy(t *testing.T) { } } +func TestLocalFormatPolicy(t *testing.T) { + type testCase struct { + tag string + input *currentSplit + expect *currentSplit + } + + localFormatParam := func(format string) []*commonpb.KeyValuePair { + return []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: format, + }, + } + } + + cases := []testCase{ + { + tag: "mixed_local_formats", + input: newCurrentSplit([]*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 101, + DataType: schemapb.DataType_VarChar, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + { + FieldID: 102, + DataType: schemapb.DataType_Double, + }, + { + FieldID: 103, + DataType: schemapb.DataType_Int64, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + }, nil), + expect: ¤tSplit{ + processFields: typeutil.NewSet[int64](), + }, + }, + { + tag: "single_vortex_local_format_partitions_without_output", + input: newCurrentSplit([]*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + { + FieldID: 101, + DataType: schemapb.DataType_Double, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + }, nil), + expect: ¤tSplit{ + processFields: typeutil.NewSet[int64](), + }, + }, + } + + policy := NewLocalFormatPolicy() + for _, tc := range cases { + t.Run(tc.tag, func(t *testing.T) { + result := policy.Split(tc.input) + + AssertSplitEqual(t, tc.expect, result) + switch tc.tag { + case "mixed_local_formats": + AssertPendingGroupsEqual(t, []ColumnGroup{ + { + Columns: []int{0, 2}, + Fields: []int64{100, 102}, + }, + { + Columns: []int{1, 3}, + Fields: []int64{101, 103}, + Format: common.LocalFormatVortex, + }, + }, result) + case "single_vortex_local_format_partitions_without_output": + AssertPendingGroupsEqual(t, []ColumnGroup{ + { + Columns: []int{0, 1}, + Fields: []int64{100, 101}, + Format: common.LocalFormatVortex, + }, + }, result) + } + }) + } +} + +func TestSplitColumnsSeparatesLocalFormatsBeforeRemanent(t *testing.T) { + localFormatParam := func(format string) []*commonpb.KeyValuePair { + return []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: format, + }, + } + } + + fields := []*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 101, + DataType: schemapb.DataType_Int64, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + { + FieldID: 102, + DataType: schemapb.DataType_FloatVector, + }, + { + FieldID: 103, + DataType: schemapb.DataType_Double, + }, + { + FieldID: 104, + DataType: schemapb.DataType_Int64, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + } + + result := SplitColumns(fields, + map[int64]ColumnStats{}, + NewLocalFormatPolicy(), + NewSelectedDataTypePolicy(), + NewRemanentShortPolicy(-1)) + + assert.Equal(t, []ColumnGroup{ + { + GroupID: 0, + Columns: []int{0, 3}, + Fields: []int64{100, 103}, + }, + { + GroupID: 1, + Columns: []int{1, 4}, + Fields: []int64{101, 104}, + Format: common.LocalFormatVortex, + }, + { + GroupID: 102, + Columns: []int{2}, + Fields: []int64{102}, + }, + }, result) +} + +func TestLocalFormatPolicyKeepsLaterSplitsWithinFormat(t *testing.T) { + localFormatParam := func(format string) []*commonpb.KeyValuePair { + return []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: format, + }, + } + } + + fields := []*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 101, + DataType: schemapb.DataType_Int64, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + { + FieldID: 102, + DataType: schemapb.DataType_Double, + }, + { + FieldID: 103, + DataType: schemapb.DataType_Double, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + } + + result := SplitColumns(fields, + map[int64]ColumnStats{}, + NewLocalFormatPolicy(), + NewRemanentShortPolicy(1)) + + assert.Equal(t, []ColumnGroup{ + { + GroupID: 0, + Columns: []int{0}, + Fields: []int64{100}, + }, + { + GroupID: 1, + Columns: []int{2}, + Fields: []int64{102}, + }, + { + GroupID: 2, + Columns: []int{1}, + Fields: []int64{101}, + Format: common.LocalFormatVortex, + }, + { + GroupID: 3, + Columns: []int{3}, + Fields: []int64{103}, + Format: common.LocalFormatVortex, + }, + }, result) +} + func TestSystemColumnPolicy(t *testing.T) { type testCase struct { tag string @@ -147,6 +412,15 @@ func TestSystemColumnPolicy(t *testing.T) { expect *currentSplit } + localFormatParam := func(format string) []*commonpb.KeyValuePair { + return []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: format, + }, + } + } + cases := []testCase{ { tag: "normal_include_pk", @@ -181,6 +455,50 @@ func TestSystemColumnPolicy(t *testing.T) { }, }, }, + { + tag: "include_pk_respects_local_format_partitions", + input: func() *currentSplit { + split := newCurrentSplit([]*schemapb.FieldSchema{ + { + FieldID: 0, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 1, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + IsPrimaryKey: true, + TypeParams: localFormatParam(common.LocalFormatVortex), + }, + { + FieldID: 101, + DataType: schemapb.DataType_FloatVector, + }, + }, nil) + split.PartitionRemainingByLocalFormat() + return split + }(), + includePK: true, + expect: ¤tSplit{ + processFields: typeutil.NewSet[int64](0, 1, 100), + outputGroups: []ColumnGroup{ + { + GroupID: 0, + Columns: []int{0, 1}, + Fields: []int64{0, 1}, + }, + { + GroupID: 1, + Columns: []int{2}, + Fields: []int64{100}, + Format: common.LocalFormatVortex, + }, + }, + }, + }, { tag: "normal_include_partition_key", input: newCurrentSplit([]*schemapb.FieldSchema{ @@ -537,6 +855,42 @@ func TestAvgSizePolicy(t *testing.T) { }, }, }, + { + tag: "over_threshold_preserves_vortex_local_format", + input: newCurrentSplit([]*schemapb.FieldSchema{ + { + FieldID: 100, + DataType: schemapb.DataType_Int64, + }, + { + FieldID: 101, + DataType: schemapb.DataType_VarChar, + TypeParams: []*commonpb.KeyValuePair{ + { + Key: common.LocalFormatKey, + Value: common.LocalFormatVortex, + }, + }, + }, + }, map[int64]ColumnStats{ + 101: { + AvgSize: 512, + MaxSize: 1024, + }, + }), + sizeThreshold: 500, + expect: ¤tSplit{ + processFields: typeutil.NewSet[int64](101), + outputGroups: []ColumnGroup{ + { + GroupID: 101, + Columns: []int{1}, + Fields: []int64{101}, + Format: common.LocalFormatVortex, + }, + }, + }, + }, } for _, tc := range cases { diff --git a/internal/storagev2/packed/ffi_common.go b/internal/storagev2/packed/ffi_common.go index 2d2a799087..1160a808dd 100644 --- a/internal/storagev2/packed/ffi_common.go +++ b/internal/storagev2/packed/ffi_common.go @@ -60,7 +60,7 @@ var ( PropertyFSUseCRC32CChecksum = C.GoString(C.loon_properties_fs_use_crc32c_checksum) PropertyWriterPolicy = C.GoString(C.loon_properties_writer_policy) - PropertyWriterFormat = "writer.format" + PropertyWriterFormat = C.GoString(C.loon_properties_writer_format) PropertyWriterSchemaBasedPattern = C.GoString(C.loon_properties_writer_schema_base_patterns) PropertyWriterSchemaBasedFormats = "writer.split.schema_based.formats" diff --git a/pkg/common/common.go b/pkg/common/common.go index 875b8ff139..bce57d75cc 100644 --- a/pkg/common/common.go +++ b/pkg/common/common.go @@ -315,9 +315,16 @@ const ( FieldDescriptionKey = "field.description" ) +// local format type +const ( + LocalFormatRaw = "raw" + LocalFormatVortex = "vortex" +) + // common properties const ( MmapEnabledKey = "mmap.enabled" + LocalFormatKey = "local_format" LoadPriorityKey = "load_priority" PartitionKeyIsolationKey = "partitionkey.isolation" FieldSkipLoadKey = "field.skipLoad"