// Licensed to the LF AI & Data foundation under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you under the Apache License, Version 2.0 (the // "License"); you may not use this file except in compliance // with the License. You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package segments import ( "context" "github.com/milvus-io/milvus-proto/go-api/v3/msgpb" "github.com/milvus-io/milvus-proto/go-api/v3/schemapb" pkoracle "github.com/milvus-io/milvus/internal/querynodev2/pkoracle" "github.com/milvus-io/milvus/internal/storage" "github.com/milvus-io/milvus/internal/storagecommon" "github.com/milvus-io/milvus/internal/util/segcore" "github.com/milvus-io/milvus/pkg/v3/proto/datapb" "github.com/milvus-io/milvus/pkg/v3/proto/querypb" "github.com/milvus-io/milvus/pkg/v3/proto/segcorepb" "github.com/milvus-io/milvus/pkg/v3/util/metautil" "github.com/milvus-io/milvus/pkg/v3/util/typeutil" ) // BinlogSaver is a minimal interface for saving binlog paths to DataCoord. // This avoids depending on the full broker.Broker interface. type BinlogSaver interface { SaveBinlogPaths(ctx context.Context, req *datapb.SaveBinlogPathsRequest) error } // ResourceUsage is used to estimate the resource usage of a sealed segment. type ResourceUsage struct { MemorySize uint64 DiskSize uint64 MmapFieldCount int FieldGpuMemorySize []uint64 } // Segment is the interface of a segment implementation. // Some methods can not apply to all segment types,such as LoadInfo, ResourceUsageEstimate. // Add more interface to represent different segment types is a better implementation. type Segment interface { // ResourceUsageEstimate() ResourceUsage // Properties ID() int64 DatabaseName() string ResourceGroup() string Collection() int64 Partition() int64 Shard() metautil.Channel Version() int64 CASVersion(int64, int64) bool StartPosition() *msgpb.MsgPosition Type() SegmentType Level() datapb.SegmentLevel IsSorted() bool LoadInfo() *querypb.SegmentLoadInfo // PinIfNotReleased the segment to prevent it from being released PinIfNotReleased() error // Unpin the segment to allow it to be released Unpin() // Stats related // InsertCount returns the number of inserted rows, not effected by deletion InsertCount() int64 // RowNum returns the number of rows, it's slow, so DO NOT call it in a loop RowNum() int64 MemSize() int64 // ResourceUsageEstimate returns the estimated resource usage of the segment ResourceUsageEstimate() ResourceUsage // Index related GetIndexByID(indexID int64) *IndexedFieldInfo GetIndex(fieldID int64) []*IndexedFieldInfo ExistIndex(fieldID int64) bool Indexes() []*IndexedFieldInfo HasRawData(fieldID int64) bool DropIndex(ctx context.Context, indexID int64) error // Modification related Insert(ctx context.Context, rowIDs []int64, timestamps []typeutil.Timestamp, record *segcorepb.InsertRecord) error Delete(ctx context.Context, primaryKeys storage.PrimaryKeys, timestamps []typeutil.Timestamp) error LoadDeltaData(ctx context.Context, deltaData *storage.DeltaData) error LastDeltaTimestamp() uint64 Load(ctx context.Context) error Release(ctx context.Context, opts ...releaseOption) Reopen(ctx context.Context, newLoadInfo *querypb.SegmentLoadInfo) error // PK candidate related (BloomFilterSet for regular segments, ExternalSegmentCandidate for external) // Segment implements pkoracle.Candidate: MayPkExist, BatchPkExist, ID, Partition, Type, // PkCandidateExist, UpdatePkCandidate, Stats, Charge, Refund — with protective guards (e.g. skipGrowingBF). SetPKCandidate(candidate pkoracle.Candidate) PkCandidateExist() bool UpdatePkCandidate(pks []storage.PrimaryKey) Stats() *storage.PkStatistics Charge() Refund() MayPkExist(lc *storage.LocationsCache) bool BatchPkExist(lc *storage.BatchLocationsCache) []bool // Get min/max GetMinPk() *storage.PrimaryKey GetMaxPk() *storage.PrimaryKey // BM25 stats UpdateBM25Stats(stats map[int64]*storage.BM25Stats) GetBM25Stats() map[int64]*storage.BM25Stats // Read operations // Search executes a search on the segment. // If searchReq.FilterOnly() is true, only executes the filter and returns valid_count (Stage 1 of two-stage search). Search(ctx context.Context, searchReq *segcore.SearchRequest) (*segcore.SearchResult, error) Retrieve(ctx context.Context, plan *segcore.RetrievePlan) (*segcorepb.RetrieveResults, error) RetrieveByOffsets(ctx context.Context, plan *segcore.RetrievePlanWithOffsets) (*segcorepb.RetrieveResults, error) // FlushData flushes data from segment memory directly to storage via C++ milvus-storage. // This is a unified interface that combines data extraction and writing: // - C++ side extracts data directly from ConcurrentVector (no query engine overhead) // - C++ side writes data to storage (TEXT fields via TextColumnWriter, others via PackedWriter) // - Returns binlog paths and metadata (all processing in C++ side) // Go layer only provides thin wrapper for FFI call - no business logic. // TODO: Implement C++ FlushData interface (Phase 1.3, 1.4, 2.3) FlushData(ctx context.Context, startOffset, endOffset int64, config *FlushConfig) (*FlushResult, error) IsLazyLoad() bool ResetIndexesLazyLoad(lazyState bool) // lazy load related NeedUpdatedVersion() int64 RemoveUnusedFieldFiles() error GetFieldJSONIndexStats() map[int64]*querypb.JsonStatsInfo } // FlushConfig contains configuration for flushing segment data. // All paths and settings are passed to C++ side via FFI. type FlushConfig struct { // Segment base path for binlog storage SegmentBasePath string // Partition base path for LOB storage (TEXT fields) PartitionBasePath string // Collection ID CollectionID int64 // Partition ID PartitionID int64 // Schema is the flush-task schema for the offset range being flushed. // It must not be inferred from the mutable growing segment runtime schema. Schema *schemapb.CollectionSchema // TEXT column field IDs TextFieldIDs []int64 // LOB base paths for each TEXT field (same order as TextFieldIDs) // Format: {partition_path}/lobs/{field_id} TextLobPaths []string // TEXT LOB writer thresholds, aligned with DataNode writer settings. TextInlineThreshold int64 TextMaxLobFileBytes int64 TextFlushThresholdBytes int64 // BM25 output sparse vector field IDs whose stats should be collected for // the flushed offset range. BM25FieldIDs []int64 // BM25 stats log IDs, in the same order as BM25FieldIDs. BM25StatsLogIDs []int64 // WriteMergedBM25Stats writes a compound BM25 stats file in the same // manifest transaction. It should be true only for the final flush. WriteMergedBM25Stats bool // ReadVersion is the manifest version to read from. // Must be set to the last version acknowledged by DataCoord (via SaveBinlogPaths). // Use ManifestEarliest for the first flush so retries never append to latest. ReadVersion int64 // WriterFormat is passed as writer.format. For incremental growing flushes, // it may be resolved from the acknowledged manifest to keep appends // compatible with existing column groups. WriterFormat string // SchemaBasedPattern is passed as writer.split.schema_based.patterns. // When set, C++ uses schema_based writer policy instead of single. SchemaBasedPattern string // SchemaBasedFormats is passed as writer.split.schema_based.formats. // It preserves per-column-group formats when appending to existing manifests. SchemaBasedFormats string // AllowedFieldIDs limits growing flush output to fields compatible with // the target segment layout. AllowedFieldIDs []int64 ColumnGroups []storagecommon.ColumnGroup } // FlushResult contains the result of flushing segment data. // All data is returned from C++ side via FFI. In Storage V3 FFI mode, // ManifestPath points to the physical files while the summary fields are used // to build DataCoord binlog metadata. type FlushResult struct { // Manifest path (Storage V3 - contains all file information). // The committed version is encoded in the path and can be extracted // via packed.UnmarshalManifestPath when needed. ManifestPath string // Number of rows flushed NumRows int64 TimestampFrom uint64 TimestampTo uint64 // FlushedFieldIDs is the authoritative set of columns the flush actually // wrote; non-materialized function-output columns are skipped and absent. FlushedFieldIDs []int64 ColumnGroupMemorySizes map[int64]int64 FieldNullCounts map[int64]int64 BM25Stats map[int64]*storage.BM25Stats }