mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 02:05:41 +00:00
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 <zhicheng.yue@zilliz.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
aoiasd
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
71b53d0a1f
commit
f662cec16d
@@ -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<uint64_t> 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)
|
||||
@@ -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
|
||||
|
||||
@@ -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<int64_t>(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<int64_t>(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
|
||||
|
||||
@@ -51,13 +51,15 @@ class FieldMeta {
|
||||
DataType type,
|
||||
bool nullable,
|
||||
std::optional<DefaultValueType> 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<DefaultValueType> 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<std::string, std::string>& params,
|
||||
std::optional<DefaultValueType> 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<DefaultValueType> 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<knowhere::MetricType> metric_type,
|
||||
bool nullable,
|
||||
std::optional<DefaultValueType> 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<knowhere::MetricType> 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<DefaultValueType> 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
|
||||
|
||||
@@ -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<Schema>();
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user