wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
2370 lines
74 KiB
Go
2370 lines
74 KiB
Go
// 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 importv2
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"testing"
|
|
|
|
"github.com/bytedance/mockey"
|
|
"github.com/cockroachdb/errors"
|
|
"github.com/stretchr/testify/assert"
|
|
|
|
"github.com/milvus-io/milvus/internal/compaction"
|
|
"github.com/milvus-io/milvus/internal/mocks"
|
|
"github.com/milvus-io/milvus/internal/storage"
|
|
"github.com/milvus-io/milvus/internal/storagev2/packed"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
|
|
)
|
|
|
|
func TestGenerateTargetPath(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
sourcePath string
|
|
wantPath string
|
|
wantErr bool
|
|
}{
|
|
{
|
|
name: "insert binlog path",
|
|
sourcePath: "files/insert_log/111/222/333/100/log1.log",
|
|
wantPath: "files/insert_log/444/555/666/100/log1.log",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "delta binlog path",
|
|
sourcePath: "files/delta_log/111/222/333/100/log1.log",
|
|
wantPath: "files/delta_log/444/555/666/100/log1.log",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "stats binlog path",
|
|
sourcePath: "files/stats_log/111/222/333/100/log1.log",
|
|
wantPath: "files/stats_log/444/555/666/100/log1.log",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "bm25 binlog path",
|
|
sourcePath: "files/bm25_stats/111/222/333/100/log1.log",
|
|
wantPath: "files/bm25_stats/444/555/666/100/log1.log",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "v3 bm25 stats numeric log id path",
|
|
sourcePath: "files/insert_log/111/222/333/_stats/bm25.100/12345",
|
|
wantPath: "files/insert_log/444/555/666/_stats/bm25.100/12345",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "external component in regular binlog root is not external table",
|
|
sourcePath: "files/external/insert_log/111/222/333/100/log1.log",
|
|
wantPath: "files/external/insert_log/444/555/666/100/log1.log",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "invalid path - no log type",
|
|
sourcePath: "files/111/222/333/100/log1.log",
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "invalid path - too short",
|
|
sourcePath: "files/insert_log/111",
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
gotPath, err := generateTargetPath(tt.sourcePath, source, target)
|
|
if tt.wantErr {
|
|
assert.Error(t, err)
|
|
} else {
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, tt.wantPath, gotPath)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGenerateMappingsFromFiles_ExternalTable(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: packed.MarshalManifestPath("files/insert_log/111/222/333", 1),
|
|
IsExternalCollection: true,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
files := &SegmentFiles{
|
|
InsertBinlogs: []string{"files/insert_log/111/222/333/_metadata/manifest.json"},
|
|
}
|
|
|
|
mappings, err := generateMappingsFromFiles(files, source, target)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t,
|
|
"files/insert_log/444/555/666/_metadata/manifest.json",
|
|
mappings["files/insert_log/111/222/333/_metadata/manifest.json"])
|
|
}
|
|
|
|
func TestGenerateTargetLOBPath(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
sourcePath string
|
|
wantPath string
|
|
wantErr bool
|
|
}{
|
|
{
|
|
name: "standard LOB file path",
|
|
sourcePath: "files/insert_log/111/222/lobs/100/_data/abc123.vx",
|
|
wantPath: "files/insert_log/444/555/lobs/100/_data/abc123.vx",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "LOB file with nested field path",
|
|
sourcePath: "root/data/insert_log/111/222/lobs/200/_data/xyz.vx",
|
|
wantPath: "root/data/insert_log/444/555/lobs/200/_data/xyz.vx",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "invalid path - no insert_log",
|
|
sourcePath: "files/other/111/222/lobs/100/_data/abc.vx",
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "invalid path - too short after insert_log",
|
|
sourcePath: "files/insert_log/111",
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
gotPath, err := generateTargetLOBPath(tt.sourcePath, source, target)
|
|
if tt.wantErr {
|
|
assert.Error(t, err)
|
|
} else {
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, tt.wantPath, gotPath)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestLobFileInfosToPaths(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
infos []packed.LobFileInfo
|
|
expected []string
|
|
}{
|
|
{
|
|
name: "empty input",
|
|
infos: nil,
|
|
expected: []string{},
|
|
},
|
|
{
|
|
name: "single LOB file - absolute path preserved",
|
|
infos: []packed.LobFileInfo{
|
|
{Path: "root/insert_log/100/200/lobs/300/_data/abc.vx", FieldID: 300, TotalRows: 1000, ValidRows: 900},
|
|
},
|
|
expected: []string{"root/insert_log/100/200/lobs/300/_data/abc.vx"},
|
|
},
|
|
{
|
|
name: "multiple LOB files from same segment",
|
|
infos: []packed.LobFileInfo{
|
|
{Path: "root/insert_log/100/200/lobs/300/_data/file1.vx", FieldID: 300, TotalRows: 500, ValidRows: 500},
|
|
{Path: "root/insert_log/100/200/lobs/300/_data/file2.vx", FieldID: 300, TotalRows: 500, ValidRows: 400},
|
|
{Path: "root/insert_log/100/200/lobs/301/_data/file3.vx", FieldID: 301, TotalRows: 200, ValidRows: 200},
|
|
},
|
|
expected: []string{
|
|
"root/insert_log/100/200/lobs/300/_data/file1.vx",
|
|
"root/insert_log/100/200/lobs/300/_data/file2.vx",
|
|
"root/insert_log/100/200/lobs/301/_data/file3.vx",
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
result := lobFileInfosToPaths(tt.infos)
|
|
if len(tt.expected) == 0 {
|
|
assert.Empty(t, result)
|
|
} else {
|
|
assert.Equal(t, tt.expected, result)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestCollectSegmentFiles_LOBFromManifest verifies that LOB file collection
|
|
// reads from the segment manifest (not the partition directory), ensuring
|
|
// only this segment's LOB files are collected.
|
|
func TestCollectSegmentFiles_LOBFromManifest(t *testing.T) {
|
|
// This segment owns only 2 LOB files
|
|
segmentLobFiles := []packed.LobFileInfo{
|
|
{Path: "root/insert_log/100/200/lobs/300/_data/seg1_file1.vx", FieldID: 300, TotalRows: 1000, ValidRows: 1000},
|
|
{Path: "root/insert_log/100/200/lobs/300/_data/seg1_file2.vx", FieldID: 300, TotalRows: 500, ValidRows: 500},
|
|
}
|
|
|
|
// Mock CreateStorageConfig to avoid paramtable dependency
|
|
mocker0 := mockey.Mock(compaction.CreateStorageConfig).Return(&indexpb.StorageConfig{}).Build()
|
|
defer mocker0.UnPatch()
|
|
|
|
// Mock GetManifestLobFiles to return only this segment's LOB files
|
|
mocker1 := mockey.Mock(packed.GetManifestLobFiles).Return(segmentLobFiles, nil).Build()
|
|
defer mocker1.UnPatch()
|
|
|
|
// Mock UnmarshalManifestPath
|
|
mocker2 := mockey.Mock(packed.UnmarshalManifestPath).Return("root/insert_log/100/200/1001", int64(1), nil).Build()
|
|
defer mocker2.UnPatch()
|
|
|
|
// Mock listAllFiles for insert binlogs (via WalkWithPrefix)
|
|
mocker3 := mockey.Mock(listAllFiles).Return(
|
|
[]string{"root/insert_log/100/200/1001/_data/cg0.parquet"}, nil).Build()
|
|
defer mocker3.UnPatch()
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
SegmentId: 1001,
|
|
CollectionId: 100,
|
|
PartitionId: 200,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: `{"basePath":"root/insert_log/100/200/1001","version":1}`,
|
|
}
|
|
|
|
files, err := collectSegmentFiles(context.Background(), nil, source)
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, files)
|
|
|
|
// Verify LOB files come from manifest, not from directory listing.
|
|
// GetManifestLobFiles returns absolute paths (after ToAbsolutePaths in C++),
|
|
// so lobFileInfosToPaths uses them directly.
|
|
assert.Equal(t, 2, len(files.LobFiles))
|
|
assert.Equal(t, "root/insert_log/100/200/lobs/300/_data/seg1_file1.vx", files.LobFiles[0])
|
|
assert.Equal(t, "root/insert_log/100/200/lobs/300/_data/seg1_file2.vx", files.LobFiles[1])
|
|
|
|
// Key invariant: only this segment's files are returned.
|
|
// The old code used listAllFiles(partition/lobs/) which would have also
|
|
// returned seg2_file3.vx, seg2_file4.vx from other segments.
|
|
}
|
|
|
|
func TestGenerateTargetIndexPath(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
sourcePath string
|
|
indexType string
|
|
pathVersion indexpb.IndexStorePathVersion
|
|
wantPath string
|
|
wantErr bool
|
|
}{
|
|
{
|
|
name: "vector scalar index path v0 format",
|
|
sourcePath: "files/index_files/1001/1/222/333/scalar_index",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
pathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED,
|
|
wantPath: "files/index_files/1001/1/555/666/scalar_index",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "text index path",
|
|
sourcePath: "files/text_log/123/1/111/222/333/100/index_file",
|
|
indexType: IndexTypeText,
|
|
wantPath: "files/text_log/123/1/444/555/666/100/index_file",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "json key index path",
|
|
sourcePath: "files/json_key_index_log/123/1/111/222/333/100/index_file",
|
|
indexType: IndexTypeJSONKey,
|
|
wantPath: "files/json_key_index_log/123/1/444/555/666/100/index_file",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "json stats path - shared_key_index",
|
|
sourcePath: "files/json_stats/2/123/1/111/222/333/100/shared_key_index/index_file",
|
|
indexType: IndexTypeJSONStats,
|
|
wantPath: "files/json_stats/2/123/1/444/555/666/100/shared_key_index/index_file",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "json stats path - shredding_data",
|
|
sourcePath: "files/json_stats/2/123/1/111/222/333/100/shredding_data/data_file",
|
|
indexType: IndexTypeJSONStats,
|
|
wantPath: "files/json_stats/2/123/1/444/555/666/100/shredding_data/data_file",
|
|
wantErr: false,
|
|
},
|
|
{
|
|
name: "invalid - keyword not found",
|
|
sourcePath: "files/other_index/111/222/333/100/index",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "invalid - path too short",
|
|
sourcePath: "files/index_files/111",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "invalid - unsupported index type",
|
|
sourcePath: "files/unknown_index/111/222/333/100/index",
|
|
indexType: "unknown_type",
|
|
wantPath: "",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "vector scalar index path v1 format",
|
|
sourcePath: "files/index_v1/111/222/333/1001/1/scalar_index",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
pathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED,
|
|
wantPath: "files/index_v1/444/555/666/1001/1/scalar_index",
|
|
wantErr: false,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
gotPath, err := generateTargetIndexPath(tt.sourcePath, source, target, tt.indexType, tt.pathVersion)
|
|
if tt.wantErr {
|
|
assert.Error(t, err)
|
|
} else {
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, tt.wantPath, gotPath)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestGenerateTargetIndexPath_BuildIDMapping verifies that NewBuildIds mapping
|
|
// is applied at the correct offset for each index path format.
|
|
// Regression test: v1 VectorScalar places buildID at offset 4 (not 1).
|
|
func TestGenerateTargetIndexPath_BuildIDMapping(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
NewBuildIds: map[int64]int64{1001: 2001},
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
sourcePath string
|
|
indexType string
|
|
pathVersion indexpb.IndexStorePathVersion
|
|
wantPath string
|
|
}{
|
|
{
|
|
name: "v0 vector scalar maps buildID at offset 1",
|
|
sourcePath: "files/index_files/1001/1/222/333/scalar_index",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
pathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED,
|
|
wantPath: "files/index_files/2001/1/555/666/scalar_index",
|
|
},
|
|
{
|
|
name: "v1 vector scalar maps buildID at offset 4",
|
|
sourcePath: "files/index_v1/111/222/333/1001/1/scalar_index",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
pathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED,
|
|
wantPath: "files/index_v1/444/555/666/2001/1/scalar_index",
|
|
},
|
|
{
|
|
name: "text index maps buildID at offset 1",
|
|
sourcePath: "files/text_log/1001/1/111/222/333/100/idx",
|
|
indexType: IndexTypeText,
|
|
wantPath: "files/text_log/2001/1/444/555/666/100/idx",
|
|
},
|
|
{
|
|
name: "json stats maps buildID at offset 2",
|
|
sourcePath: "files/json_stats/2/1001/1/111/222/333/100/shared_key_index/idx",
|
|
indexType: IndexTypeJSONStats,
|
|
wantPath: "files/json_stats/2/2001/1/444/555/666/100/shared_key_index/idx",
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got, err := generateTargetIndexPath(tt.sourcePath, source, target, tt.indexType, tt.pathVersion)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, tt.wantPath, got)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGenerateMappingsFromFiles_VectorScalarUsesSourcePathVersion(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
BuildID: 1001,
|
|
IndexVersion: 1,
|
|
IndexStorePathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED,
|
|
IndexFilePaths: []string{"files/index_files/1001/1/222/333/v0_file"},
|
|
},
|
|
{
|
|
BuildID: 1002,
|
|
IndexVersion: 1,
|
|
IndexStorePathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED,
|
|
IndexFilePaths: []string{"files/index_v1/111/222/333/1002/1/v1_file"},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
NewBuildIds: map[int64]int64{1001: 2001, 1002: 2002},
|
|
}
|
|
files := &SegmentFiles{
|
|
VectorScalarIndex: []string{
|
|
"files/index_files/1001/1/222/333/v0_file",
|
|
"files/index_v1/111/222/333/1002/1/v1_file",
|
|
},
|
|
}
|
|
|
|
mappings, err := generateMappingsFromFiles(files, source, target)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, "files/index_files/2001/1/555/666/v0_file", mappings["files/index_files/1001/1/222/333/v0_file"])
|
|
assert.Equal(t, "files/index_v1/444/555/666/2002/1/v1_file", mappings["files/index_v1/111/222/333/1002/1/v1_file"])
|
|
}
|
|
|
|
func TestTransformFieldBinlogs(t *testing.T) {
|
|
mappings := map[string]string{
|
|
"files/insert_log/111/222/333/100/log1.log": "files/insert_log/444/555/666/100/log1.log",
|
|
"files/insert_log/111/222/333/101/log2.log": "files/insert_log/444/555/666/101/log2.log",
|
|
}
|
|
|
|
srcFieldBinlogs := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 1000,
|
|
TimestampFrom: 100,
|
|
TimestampTo: 200,
|
|
LogPath: "files/insert_log/111/222/333/100/log1.log",
|
|
LogSize: 1024,
|
|
MemorySize: 1024,
|
|
},
|
|
},
|
|
},
|
|
{
|
|
FieldID: 101,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 2000,
|
|
TimestampFrom: 150,
|
|
TimestampTo: 250,
|
|
LogPath: "files/insert_log/111/222/333/101/log2.log",
|
|
LogSize: 2048,
|
|
MemorySize: 2048,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
t.Run("count rows for insert logs", func(t *testing.T) {
|
|
result, totalRows, err := transformFieldBinlogs(srcFieldBinlogs, mappings, true, false)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(3000), totalRows)
|
|
assert.Equal(t, 2, len(result))
|
|
|
|
// Verify first field binlog
|
|
assert.Equal(t, int64(100), result[0].FieldID)
|
|
assert.Equal(t, 1, len(result[0].Binlogs))
|
|
assert.Equal(t, int64(1000), result[0].Binlogs[0].EntriesNum)
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log1.log", result[0].Binlogs[0].LogPath)
|
|
assert.Equal(t, int64(1024), result[0].Binlogs[0].LogSize)
|
|
assert.Equal(t, int64(1024), result[0].Binlogs[0].MemorySize)
|
|
|
|
// Verify second field binlog
|
|
assert.Equal(t, int64(101), result[1].FieldID)
|
|
assert.Equal(t, 1, len(result[1].Binlogs))
|
|
assert.Equal(t, int64(2000), result[1].Binlogs[0].EntriesNum)
|
|
})
|
|
|
|
t.Run("no row counting for stats logs", func(t *testing.T) {
|
|
result, totalRows, err := transformFieldBinlogs(srcFieldBinlogs, mappings, false, false)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(0), totalRows)
|
|
assert.Equal(t, 2, len(result))
|
|
})
|
|
|
|
t.Run("skip binlogs with empty path", func(t *testing.T) {
|
|
srcWithEmpty := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 1000,
|
|
LogPath: "",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
result, _, err := transformFieldBinlogs(srcWithEmpty, mappings, false, false)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, 0, len(result))
|
|
})
|
|
|
|
t.Run("error on missing mapping", func(t *testing.T) {
|
|
srcWithUnmapped := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 1000,
|
|
LogPath: "files/insert_log/999/888/777/100/unmapped.log",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
result, totalRows, err := transformFieldBinlogs(srcWithUnmapped, mappings, true, false)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found")
|
|
assert.Nil(t, result)
|
|
assert.Equal(t, int64(0), totalRows)
|
|
})
|
|
}
|
|
|
|
func TestCopySegmentAndIndexFiles(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 1000,
|
|
LogPath: "files/insert_log/111/222/333/100/12345",
|
|
LogSize: 1024,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
FieldID: 100,
|
|
IndexID: 1001,
|
|
BuildID: 1002,
|
|
IndexFilePaths: []string{"files/index_files/1002/1/222/333/index1"},
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
t.Run("successful copy", func(t *testing.T) {
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
assert.Equal(t, int64(666), result.SegmentId)
|
|
assert.Equal(t, int64(1000), result.ImportedRows)
|
|
assert.Equal(t, 1, len(result.Binlogs))
|
|
assert.Equal(t, 1, len(result.IndexInfos))
|
|
assert.Len(t, copiedFiles, 2)
|
|
})
|
|
|
|
t.Run("copy failure", func(t *testing.T) {
|
|
mCopy := mockey.Mock(copyFile).Return(errors.New("copy failed")).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, result)
|
|
assert.Contains(t, err.Error(), "failed to copy file")
|
|
assert.Empty(t, copiedFiles)
|
|
})
|
|
}
|
|
|
|
func TestGenerateSegmentInfoFromSource(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
EntriesNum: 1000,
|
|
TimestampFrom: 100,
|
|
TimestampTo: 200,
|
|
LogPath: "files/insert_log/111/222/333/100/log1.log",
|
|
LogSize: 1024,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
StatsBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
LogPath: "files/stats_log/111/222/333/100/stats1.log",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mappings := map[string]string{
|
|
"files/insert_log/111/222/333/100/log1.log": "files/insert_log/444/555/666/100/log1.log",
|
|
"files/stats_log/111/222/333/100/stats1.log": "files/stats_log/444/555/666/100/stats1.log",
|
|
}
|
|
|
|
t.Run("generate segment info", func(t *testing.T) {
|
|
segmentInfo, err := generateSegmentInfoFromSource(source, target, mappings)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, segmentInfo)
|
|
assert.Equal(t, int64(666), segmentInfo.SegmentID)
|
|
assert.Equal(t, int64(1000), segmentInfo.ImportedRows)
|
|
assert.Equal(t, 1, len(segmentInfo.Binlogs))
|
|
assert.Equal(t, 1, len(segmentInfo.Statslogs))
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log1.log", segmentInfo.Binlogs[0].Binlogs[0].LogPath)
|
|
})
|
|
}
|
|
|
|
func TestBuildIndexInfoFromSource(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
FieldID: 100,
|
|
IndexID: 1001,
|
|
BuildID: 1002,
|
|
IndexFilePaths: []string{"files/index_files/1002/1/222/333/index1"},
|
|
SerializedSize: 5000,
|
|
},
|
|
},
|
|
TextIndexFiles: map[int64]*datapb.TextIndexStats{
|
|
100: {
|
|
FieldID: 100,
|
|
Version: 1,
|
|
BuildID: 2001,
|
|
Files: []string{"files/text_log/123/1/111/222/333/100/text1"},
|
|
LogSize: 2048,
|
|
MemorySize: 4096,
|
|
},
|
|
},
|
|
JsonKeyIndexFiles: map[int64]*datapb.JsonKeyStats{
|
|
101: {
|
|
FieldID: 101,
|
|
Version: 1,
|
|
BuildID: 3001,
|
|
JsonKeyStatsDataFormat: 1, // Legacy format
|
|
Files: []string{"files/json_key_index_log/123/1/111/222/333/101/json1"},
|
|
MemorySize: 3072,
|
|
},
|
|
102: {
|
|
FieldID: 102,
|
|
Version: 1,
|
|
BuildID: 3002,
|
|
JsonKeyStatsDataFormat: 2, // New format
|
|
Files: []string{
|
|
"files/json_stats/2/3002/1/111/222/333/102/shared_key_index/index1",
|
|
"files/json_stats/2/3002/1/111/222/333/102/shredding_data/data1",
|
|
},
|
|
MemorySize: 4096,
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mappings := map[string]string{
|
|
"files/index_files/1002/1/222/333/index1": "files/index_files/1002/1/555/666/index1",
|
|
"files/text_log/123/1/111/222/333/100/text1": "files/text_log/123/1/444/555/666/100/text1",
|
|
"files/json_key_index_log/123/1/111/222/333/101/json1": "files/json_key_index_log/123/1/444/555/666/101/json1",
|
|
"files/json_stats/2/3002/1/111/222/333/102/shared_key_index/index1": "files/json_stats/2/3002/1/444/555/666/102/shared_key_index/index1",
|
|
"files/json_stats/2/3002/1/111/222/333/102/shredding_data/data1": "files/json_stats/2/3002/1/444/555/666/102/shredding_data/data1",
|
|
}
|
|
|
|
t.Run("error on missing vector index mapping", func(t *testing.T) {
|
|
emptyMappings := map[string]string{}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, emptyMappings)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for index file")
|
|
})
|
|
|
|
t.Run("error on missing text index mapping", func(t *testing.T) {
|
|
partialMappings := map[string]string{
|
|
"files/index_files/1002/1/222/333/index1": "files/index_files/1002/1/555/666/index1",
|
|
}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, partialMappings)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for text index file")
|
|
})
|
|
|
|
t.Run("error on missing json index mapping", func(t *testing.T) {
|
|
partialMappings := map[string]string{
|
|
"files/index_files/1002/1/222/333/index1": "files/index_files/1002/1/555/666/index1",
|
|
"files/text_log/123/1/111/222/333/100/text1": "files/text_log/123/1/444/555/666/100/text1",
|
|
}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, partialMappings)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for JSON index file")
|
|
})
|
|
|
|
t.Run("build all index info", func(t *testing.T) {
|
|
indexInfos, textIndexInfos, jsonKeyIndexInfos, err := buildIndexInfoFromSource(source, target, mappings)
|
|
assert.NoError(t, err)
|
|
|
|
// Verify vector/scalar index info (keyed by buildID, not fieldID)
|
|
assert.Equal(t, 1, len(indexInfos))
|
|
assert.NotNil(t, indexInfos[1002])
|
|
assert.Equal(t, int64(100), indexInfos[1002].FieldId)
|
|
assert.Equal(t, int64(1001), indexInfos[1002].IndexId)
|
|
assert.Equal(t, int64(1002), indexInfos[1002].BuildId)
|
|
assert.Equal(t, int64(5000), indexInfos[1002].IndexSize)
|
|
assert.Equal(t, "files/index_files/1002/1/555/666/index1", indexInfos[1002].IndexFilePaths[0])
|
|
|
|
// Verify text index info
|
|
assert.Equal(t, 1, len(textIndexInfos))
|
|
assert.NotNil(t, textIndexInfos[100])
|
|
assert.Equal(t, int64(100), textIndexInfos[100].FieldID)
|
|
assert.Equal(t, int64(2001), textIndexInfos[100].BuildID)
|
|
assert.Equal(t, "files/text_log/123/1/444/555/666/100/text1", textIndexInfos[100].Files[0])
|
|
|
|
// Verify JSON key index info (legacy and new formats)
|
|
assert.Equal(t, 2, len(jsonKeyIndexInfos))
|
|
|
|
// Legacy format (data_format = 1)
|
|
assert.NotNil(t, jsonKeyIndexInfos[101])
|
|
assert.Equal(t, int64(101), jsonKeyIndexInfos[101].FieldID)
|
|
assert.Equal(t, int64(3001), jsonKeyIndexInfos[101].BuildID)
|
|
assert.Equal(t, int64(1), jsonKeyIndexInfos[101].JsonKeyStatsDataFormat)
|
|
assert.Equal(t, "files/json_key_index_log/123/1/444/555/666/101/json1", jsonKeyIndexInfos[101].Files[0])
|
|
|
|
// New format (data_format = 2)
|
|
assert.NotNil(t, jsonKeyIndexInfos[102])
|
|
assert.Equal(t, int64(102), jsonKeyIndexInfos[102].FieldID)
|
|
assert.Equal(t, int64(3002), jsonKeyIndexInfos[102].BuildID)
|
|
assert.Equal(t, int64(2), jsonKeyIndexInfos[102].JsonKeyStatsDataFormat)
|
|
assert.Equal(t, 2, len(jsonKeyIndexInfos[102].Files))
|
|
assert.Equal(t, "files/json_stats/2/3002/1/444/555/666/102/shared_key_index/index1", jsonKeyIndexInfos[102].Files[0])
|
|
assert.Equal(t, "files/json_stats/2/3002/1/444/555/666/102/shredding_data/data1", jsonKeyIndexInfos[102].Files[1])
|
|
})
|
|
}
|
|
|
|
func TestBuildIndexInfoFromSource_PreservesIndexStorePathVersion(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
FieldID: 100,
|
|
IndexID: 10,
|
|
BuildID: 1001,
|
|
IndexName: "idx_v0",
|
|
IndexVersion: 1,
|
|
SerializedSize: 1024,
|
|
IndexStorePathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED,
|
|
IndexFilePaths: []string{"files/index_files/1001/1/222/333/v0_file"},
|
|
},
|
|
{
|
|
FieldID: 101,
|
|
IndexID: 11,
|
|
BuildID: 1002,
|
|
IndexName: "idx_v1",
|
|
IndexVersion: 1,
|
|
SerializedSize: 2048,
|
|
IndexStorePathVersion: indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED,
|
|
IndexFilePaths: []string{"files/index_v1/111/222/333/1002/1/v1_file"},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
NewBuildIds: map[int64]int64{1001: 2001, 1002: 2002},
|
|
}
|
|
mappings := map[string]string{
|
|
"files/index_files/1001/1/222/333/v0_file": "files/index_files/2001/1/555/666/v0_file",
|
|
"files/index_v1/111/222/333/1002/1/v1_file": "files/index_v1/444/555/666/2002/1/v1_file",
|
|
}
|
|
|
|
indexInfos, _, _, err := buildIndexInfoFromSource(source, target, mappings)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED, indexInfos[2001].GetIndexStorePathVersion())
|
|
assert.Equal(t, indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_COLLECTION_ROOTED, indexInfos[2002].GetIndexStorePathVersion())
|
|
}
|
|
|
|
func TestCopySegmentAndIndexFiles_ReturnsFileList(t *testing.T) {
|
|
t.Run("success returns all copied files", func(t *testing.T) {
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 1,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/insert_log/111/222/333/1/10001", LogSize: 100},
|
|
{LogPath: "files/insert_log/111/222/333/1/10002", LogSize: 200},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
assert.Len(t, copiedFiles, 2)
|
|
assert.Contains(t, copiedFiles, "files/insert_log/444/555/666/1/10001")
|
|
assert.Contains(t, copiedFiles, "files/insert_log/444/555/666/1/10002")
|
|
})
|
|
|
|
t.Run("failure returns partial file list", func(t *testing.T) {
|
|
callCount := 0
|
|
mCopy := mockey.Mock(copyFile).
|
|
To(func(_ context.Context, _ storage.ChunkManager, src, dst string) error {
|
|
callCount++
|
|
if callCount > 1 {
|
|
return errors.New("copy failed")
|
|
}
|
|
return nil
|
|
}).Build()
|
|
defer mCopy.UnPatch()
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 1,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/insert_log/111/222/333/1/10001", LogSize: 100},
|
|
{LogPath: "files/insert_log/111/222/333/1/10002", LogSize: 200},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, result)
|
|
assert.True(t, len(copiedFiles) <= 1, "should return files copied before failure")
|
|
})
|
|
}
|
|
|
|
func TestGenerateTargetIndexPath_PathTooShort(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
path string
|
|
indexType string
|
|
}{
|
|
{
|
|
name: "text index path too short",
|
|
path: "files/text_log/123/1/111",
|
|
indexType: IndexTypeText,
|
|
},
|
|
{
|
|
name: "json key index path too short",
|
|
path: "files/json_key_index_log/123",
|
|
indexType: IndexTypeJSONKey,
|
|
},
|
|
{
|
|
name: "json stats path too short",
|
|
path: "files/json_stats/2/123/1/111",
|
|
indexType: IndexTypeJSONStats,
|
|
},
|
|
{
|
|
name: "vector scalar path exactly at keyword",
|
|
path: "files/index_files",
|
|
indexType: IndexTypeVectorScalarV0,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
_, err := generateTargetIndexPath(tt.path, source, target, tt.indexType, indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "invalid")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGenerateTargetIndexPath_VectorScalarPreservesIDs(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
// Verify buildID (1001) and indexVersion (1) are preserved, NOT replaced
|
|
// partitionID (222->555) and segmentID (333->666) ARE replaced
|
|
// collectionID is NOT in path at all
|
|
result, err := generateTargetIndexPath("files/index_files/1001/1/222/333/HNSW_SQ_3", source, target, IndexTypeVectorScalarV0, indexpb.IndexStorePathVersion_INDEX_STORE_PATH_VERSION_BUILD_ROOTED)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, "files/index_files/1001/1/555/666/HNSW_SQ_3", result)
|
|
|
|
// Verify collectionID (111) does NOT appear in target path
|
|
assert.NotContains(t, result, "111")
|
|
assert.NotContains(t, result, "444") // collectionID should NOT be injected
|
|
}
|
|
|
|
func TestGenerateTargetPath_AllConstants(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
// Verify all constants match the correct storage path names
|
|
tests := []struct {
|
|
name string
|
|
logType string
|
|
expected string
|
|
}{
|
|
{"insert uses insert_log", BinlogTypeInsert, "insert_log"},
|
|
{"delta uses delta_log", BinlogTypeDelta, "delta_log"},
|
|
{"stats uses stats_log", BinlogTypeStats, "stats_log"},
|
|
{"bm25 uses bm25_stats", BinlogTypeBM25, "bm25_stats"},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
path := fmt.Sprintf("files/%s/111/222/333/100/log1", tt.expected)
|
|
result, err := generateTargetPath(path, source, target)
|
|
assert.NoError(t, err)
|
|
assert.Contains(t, result, tt.expected)
|
|
assert.Contains(t, result, "444")
|
|
assert.Contains(t, result, "555")
|
|
assert.Contains(t, result, "666")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestGenerateSegmentInfoFromSource_AllBinlogTypes(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 500, LogPath: "files/insert_log/111/222/333/100/log1"},
|
|
},
|
|
},
|
|
},
|
|
StatsBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/stats_log/111/222/333/100/stats1"},
|
|
},
|
|
},
|
|
},
|
|
DeltaBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 10, LogPath: "files/delta_log/111/222/333/100/delta1"},
|
|
},
|
|
},
|
|
},
|
|
Bm25Binlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/bm25_stats/111/222/333/100/bm25_1"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mappings := map[string]string{
|
|
"files/insert_log/111/222/333/100/log1": "files/insert_log/444/555/666/100/log1",
|
|
"files/stats_log/111/222/333/100/stats1": "files/stats_log/444/555/666/100/stats1",
|
|
"files/delta_log/111/222/333/100/delta1": "files/delta_log/444/555/666/100/delta1",
|
|
"files/bm25_stats/111/222/333/100/bm25_1": "files/bm25_stats/444/555/666/100/bm25_1",
|
|
}
|
|
|
|
segmentInfo, err := generateSegmentInfoFromSource(source, target, mappings)
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, segmentInfo)
|
|
assert.Equal(t, int64(666), segmentInfo.SegmentID)
|
|
assert.Equal(t, int64(500), segmentInfo.ImportedRows) // Only insert rows counted
|
|
assert.Equal(t, 1, len(segmentInfo.Binlogs))
|
|
assert.Equal(t, 1, len(segmentInfo.Statslogs))
|
|
assert.Equal(t, 1, len(segmentInfo.Deltalogs))
|
|
assert.Equal(t, 1, len(segmentInfo.Bm25Logs))
|
|
assert.Equal(t, "files/delta_log/444/555/666/100/delta1", segmentInfo.Deltalogs[0].Binlogs[0].LogPath)
|
|
assert.Equal(t, "files/bm25_stats/444/555/666/100/bm25_1", segmentInfo.Bm25Logs[0].Binlogs[0].LogPath)
|
|
}
|
|
|
|
func TestGenerateSegmentInfoFromSource_EmptySource(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
segmentInfo, err := generateSegmentInfoFromSource(source, target, map[string]string{})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, segmentInfo)
|
|
assert.Equal(t, int64(666), segmentInfo.SegmentID)
|
|
assert.Equal(t, int64(0), segmentInfo.ImportedRows)
|
|
assert.Equal(t, 0, len(segmentInfo.Binlogs))
|
|
assert.Equal(t, 0, len(segmentInfo.Statslogs))
|
|
assert.Equal(t, 0, len(segmentInfo.Deltalogs))
|
|
assert.Equal(t, 0, len(segmentInfo.Bm25Logs))
|
|
}
|
|
|
|
// TestBuildIndexInfoFromSource_MultipleIndexesPerField verifies that multiple indexes
|
|
// on the same field (e.g., JSON path indexes) are all preserved in the result map.
|
|
// This is the core scenario for the fix: using buildID as map key instead of fieldID.
|
|
func TestBuildIndexInfoFromSource_MultipleIndexesPerField(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
FieldID: 100, // Same field
|
|
IndexID: 1001,
|
|
BuildID: 2001,
|
|
IndexName: "idx_category",
|
|
IndexFilePaths: []string{"files/index_files/2001/1/222/333/index1"},
|
|
SerializedSize: 3000,
|
|
},
|
|
{
|
|
FieldID: 100, // Same field, different path index
|
|
IndexID: 1002,
|
|
BuildID: 2002,
|
|
IndexName: "idx_price",
|
|
IndexFilePaths: []string{"files/index_files/2002/1/222/333/index2"},
|
|
SerializedSize: 4000,
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
NewBuildIds: map[int64]int64{
|
|
2001: 3001, // old buildID -> new buildID
|
|
2002: 3002,
|
|
},
|
|
}
|
|
|
|
mappings := map[string]string{
|
|
"files/index_files/2001/1/222/333/index1": "files/index_files/3001/1/555/666/index1",
|
|
"files/index_files/2002/1/222/333/index2": "files/index_files/3002/1/555/666/index2",
|
|
}
|
|
|
|
indexInfos, _, _, err := buildIndexInfoFromSource(source, target, mappings)
|
|
assert.NoError(t, err)
|
|
|
|
// Both indexes should be present (keyed by new buildID)
|
|
assert.Equal(t, 2, len(indexInfos), "Both JSON path indexes should be in the result")
|
|
|
|
// Verify first index (idx_category)
|
|
assert.NotNil(t, indexInfos[3001])
|
|
assert.Equal(t, int64(100), indexInfos[3001].FieldId)
|
|
assert.Equal(t, int64(1001), indexInfos[3001].IndexId)
|
|
assert.Equal(t, int64(3001), indexInfos[3001].BuildId)
|
|
assert.Equal(t, "idx_category", indexInfos[3001].IndexName)
|
|
|
|
// Verify second index (idx_price)
|
|
assert.NotNil(t, indexInfos[3002])
|
|
assert.Equal(t, int64(100), indexInfos[3002].FieldId)
|
|
assert.Equal(t, int64(1002), indexInfos[3002].IndexId)
|
|
assert.Equal(t, int64(3002), indexInfos[3002].BuildId)
|
|
assert.Equal(t, "idx_price", indexInfos[3002].IndexName)
|
|
}
|
|
|
|
func TestBuildIndexInfoFromSource_EmptySource(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
indexInfos, textIndexInfos, jsonKeyIndexInfos, err := buildIndexInfoFromSource(source, target, map[string]string{})
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, 0, len(indexInfos))
|
|
assert.Equal(t, 0, len(textIndexInfos))
|
|
assert.Equal(t, 0, len(jsonKeyIndexInfos))
|
|
}
|
|
|
|
func TestBuildIndexInfoFromSource_UnmappedPaths(t *testing.T) {
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
t.Run("unmapped vector index path returns error", func(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
IndexFiles: []*indexpb.IndexFilePathInfo{
|
|
{
|
|
FieldID: 100,
|
|
IndexID: 1001,
|
|
BuildID: 1002,
|
|
IndexFilePaths: []string{"unmapped/path"},
|
|
SerializedSize: 5000,
|
|
},
|
|
},
|
|
}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, map[string]string{})
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for index file")
|
|
})
|
|
|
|
t.Run("unmapped text index path returns error", func(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
TextIndexFiles: map[int64]*datapb.TextIndexStats{
|
|
200: {
|
|
FieldID: 200,
|
|
Files: []string{"unmapped/text/path"},
|
|
},
|
|
},
|
|
}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, map[string]string{})
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for text index file")
|
|
})
|
|
|
|
t.Run("unmapped json index path returns error", func(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
JsonKeyIndexFiles: map[int64]*datapb.JsonKeyStats{
|
|
300: {
|
|
FieldID: 300,
|
|
Files: []string{"unmapped/json/path"},
|
|
},
|
|
},
|
|
}
|
|
_, _, _, err := buildIndexInfoFromSource(source, target, map[string]string{})
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for JSON index file")
|
|
})
|
|
}
|
|
|
|
func TestTransformFieldBinlogs_NilInput(t *testing.T) {
|
|
result, totalRows, err := transformFieldBinlogs(nil, map[string]string{}, true, false)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(0), totalRows)
|
|
assert.Equal(t, 0, len(result))
|
|
}
|
|
|
|
func TestTransformFieldBinlogs_MultipleBinlogsPerField(t *testing.T) {
|
|
mappings := map[string]string{
|
|
"files/insert_log/111/222/333/100/log1": "files/insert_log/444/555/666/100/log1",
|
|
"files/insert_log/111/222/333/100/log2": "files/insert_log/444/555/666/100/log2",
|
|
"files/insert_log/111/222/333/100/log3": "files/insert_log/444/555/666/100/log3",
|
|
}
|
|
|
|
srcFieldBinlogs := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 100, LogPath: "files/insert_log/111/222/333/100/log1", LogSize: 512},
|
|
{EntriesNum: 200, LogPath: "files/insert_log/111/222/333/100/log2", LogSize: 1024},
|
|
{EntriesNum: 300, LogPath: "files/insert_log/111/222/333/100/log3", LogSize: 2048},
|
|
},
|
|
},
|
|
}
|
|
|
|
result, totalRows, err := transformFieldBinlogs(srcFieldBinlogs, mappings, true, false)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(600), totalRows)
|
|
assert.Equal(t, 1, len(result))
|
|
assert.Equal(t, 3, len(result[0].Binlogs))
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log1", result[0].Binlogs[0].LogPath)
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log2", result[0].Binlogs[1].LogPath)
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log3", result[0].Binlogs[2].LogPath)
|
|
}
|
|
|
|
func TestTransformFieldBinlogs_UnmappedPath(t *testing.T) {
|
|
// Path not in mappings -> returns error (fail-fast)
|
|
mappings := map[string]string{}
|
|
|
|
srcFieldBinlogs := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 100, LogPath: "files/insert_log/111/222/333/100/log1"},
|
|
},
|
|
},
|
|
}
|
|
|
|
_, _, err := transformFieldBinlogs(srcFieldBinlogs, mappings, true, false)
|
|
assert.Error(t, err)
|
|
assert.Contains(t, err.Error(), "no mapping found for source path")
|
|
}
|
|
|
|
func TestCopySegmentAndIndexFiles_CreateFileMappingsError(t *testing.T) {
|
|
mockCM := mocks.NewChunkManager(t)
|
|
// Source with invalid path that will cause generateMappingsFromFiles to fail
|
|
source := &datapb.CopySegmentSource{
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "invalid/path/without/log/type"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), mockCM, source, target, nil)
|
|
assert.Error(t, err)
|
|
assert.Nil(t, result)
|
|
assert.Nil(t, copiedFiles)
|
|
assert.Contains(t, err.Error(), "failed to generate file mappings")
|
|
}
|
|
|
|
func TestShortenIndexFilePaths(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
fullPaths []string
|
|
expected []string
|
|
}{
|
|
{
|
|
name: "vector/scalar index paths",
|
|
fullPaths: []string{
|
|
"files/index_files/1001/1/555/666/scalar_index",
|
|
"files/index_files/1001/1/555/666/vector_index",
|
|
},
|
|
expected: []string{"scalar_index", "vector_index"},
|
|
},
|
|
{
|
|
name: "empty path list",
|
|
fullPaths: []string{},
|
|
expected: []string{},
|
|
},
|
|
{
|
|
name: "single file name",
|
|
fullPaths: []string{
|
|
"index_file",
|
|
},
|
|
expected: []string{"index_file"},
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
result := shortenIndexFilePaths(tt.fullPaths)
|
|
assert.Equal(t, tt.expected, result)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestShortenSingleJsonStatsPath(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
inputPath string
|
|
expectedPath string
|
|
}{
|
|
{
|
|
name: "shared_key_index path",
|
|
inputPath: "files/json_stats/2/123/1/444/555/666/100/shared_key_index/inverted_index_0",
|
|
expectedPath: "shared_key_index/inverted_index_0",
|
|
},
|
|
{
|
|
name: "shredding_data path",
|
|
inputPath: "files/json_stats/2/123/1/444/555/666/100/shredding_data/parquet_data_0",
|
|
expectedPath: "shredding_data/parquet_data_0",
|
|
},
|
|
{
|
|
name: "already shortened - shared_key_index",
|
|
inputPath: "shared_key_index/inverted_index_0",
|
|
expectedPath: "shared_key_index/inverted_index_0",
|
|
},
|
|
{
|
|
name: "already shortened - shredding_data",
|
|
inputPath: "shredding_data/parquet_data_0",
|
|
expectedPath: "shredding_data/parquet_data_0",
|
|
},
|
|
{
|
|
name: "no keyword - keep as-is",
|
|
inputPath: "files/other/path/file.json",
|
|
expectedPath: "files/other/path/file.json",
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
result := shortenSingleJSONStatsPath(tt.inputPath)
|
|
assert.Equal(t, tt.expectedPath, result)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestShortenJsonStatsPath(t *testing.T) {
|
|
jsonStats := map[int64]*datapb.JsonKeyStats{
|
|
100: {
|
|
FieldID: 100,
|
|
Version: 1,
|
|
BuildID: 123,
|
|
Files: []string{
|
|
"files/json_stats/2/123/1/444/555/666/100/shared_key_index/inverted_index_0",
|
|
"files/json_stats/2/123/1/444/555/666/100/shared_key_index/inverted_index_1",
|
|
},
|
|
JsonKeyStatsDataFormat: 2,
|
|
MemorySize: 1024,
|
|
LogSize: 2048,
|
|
},
|
|
200: {
|
|
FieldID: 200,
|
|
Version: 1,
|
|
BuildID: 456,
|
|
Files: []string{
|
|
"files/json_stats/2/456/1/444/555/666/200/shredding_data/parquet_data_0",
|
|
},
|
|
JsonKeyStatsDataFormat: 2,
|
|
MemorySize: 512,
|
|
},
|
|
}
|
|
|
|
result := shortenJSONStatsPath(jsonStats)
|
|
|
|
assert.Equal(t, 2, len(result))
|
|
|
|
// Check field 100
|
|
assert.NotNil(t, result[100])
|
|
assert.Equal(t, int64(100), result[100].FieldID)
|
|
assert.Equal(t, int64(1), result[100].Version)
|
|
assert.Equal(t, int64(123), result[100].BuildID)
|
|
assert.Equal(t, int64(2), result[100].JsonKeyStatsDataFormat)
|
|
assert.Equal(t, int64(1024), result[100].MemorySize)
|
|
assert.Equal(t, int64(2048), result[100].LogSize)
|
|
assert.Equal(t, 2, len(result[100].Files))
|
|
assert.Equal(t, "shared_key_index/inverted_index_0", result[100].Files[0])
|
|
assert.Equal(t, "shared_key_index/inverted_index_1", result[100].Files[1])
|
|
|
|
// Check field 200
|
|
assert.NotNil(t, result[200])
|
|
assert.Equal(t, int64(200), result[200].FieldID)
|
|
assert.Equal(t, int64(1), result[200].Version)
|
|
assert.Equal(t, int64(456), result[200].BuildID)
|
|
assert.Equal(t, int64(2), result[200].JsonKeyStatsDataFormat)
|
|
assert.Equal(t, int64(512), result[200].MemorySize)
|
|
assert.Equal(t, 1, len(result[200].Files))
|
|
assert.Equal(t, "shredding_data/parquet_data_0", result[200].Files[0])
|
|
}
|
|
|
|
func TestShortenJsonStatsPath_MetaJson(t *testing.T) {
|
|
jsonStats := map[int64]*datapb.JsonKeyStats{
|
|
102: {
|
|
FieldID: 102,
|
|
Version: 1,
|
|
BuildID: 462930163709949539,
|
|
Files: []string{
|
|
"files/json_stats/2/462930163709949539/1/462930163710600038/462930163710600039/462930163710600046/102/meta.json",
|
|
},
|
|
JsonKeyStatsDataFormat: 2,
|
|
MemorySize: 256,
|
|
},
|
|
}
|
|
|
|
result := shortenJSONStatsPath(jsonStats)
|
|
|
|
assert.Equal(t, 1, len(result))
|
|
assert.NotNil(t, result[102])
|
|
assert.Equal(t, int64(102), result[102].FieldID)
|
|
assert.Equal(t, 1, len(result[102].Files))
|
|
assert.Equal(t, "meta.json", result[102].Files[0])
|
|
}
|
|
|
|
func TestShortenSingleJsonStatsPath_EdgeCases(t *testing.T) {
|
|
t.Run("already_shortened_meta", func(t *testing.T) {
|
|
result := shortenSingleJSONStatsPath("meta.json")
|
|
assert.Equal(t, "meta.json", result)
|
|
})
|
|
|
|
t.Run("already_shortened_shared_key", func(t *testing.T) {
|
|
result := shortenSingleJSONStatsPath("shared_key_index/inverted_index_0")
|
|
assert.Equal(t, "shared_key_index/inverted_index_0", result)
|
|
})
|
|
|
|
t.Run("full_path_meta_json", func(t *testing.T) {
|
|
fullPath := "files/json_stats/2/123/1/444/555/666/100/meta.json"
|
|
result := shortenSingleJSONStatsPath(fullPath)
|
|
assert.Equal(t, "meta.json", result)
|
|
})
|
|
|
|
t.Run("full_path_nested_file", func(t *testing.T) {
|
|
fullPath := "files/json_stats/2/123/1/444/555/666/100/subdir/file.dat"
|
|
result := shortenSingleJSONStatsPath(fullPath)
|
|
assert.Equal(t, "subdir/file.dat", result)
|
|
})
|
|
}
|
|
|
|
func TestTransformManifestPath(t *testing.T) {
|
|
target := &datapb.CopySegmentTarget{
|
|
SegmentId: 2001,
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
}
|
|
|
|
sourceManifest := packed.MarshalManifestPath(
|
|
"files/insert_log/449104612037410004/449104621518610066/449104621518610065",
|
|
2,
|
|
)
|
|
|
|
source := &datapb.CopySegmentSource{}
|
|
targetManifest, err := transformManifestPath(sourceManifest, source, target)
|
|
assert.NoError(t, err)
|
|
|
|
basePath, version, err := packed.UnmarshalManifestPath(targetManifest)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(2), version)
|
|
assert.Contains(t, basePath, "111")
|
|
assert.Contains(t, basePath, "222")
|
|
assert.Contains(t, basePath, "2001")
|
|
}
|
|
|
|
func TestTransformManifestPath_ExternalTable(t *testing.T) {
|
|
target := &datapb.CopySegmentTarget{
|
|
SegmentId: 666,
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
}
|
|
|
|
sourceManifest := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
source := &datapb.CopySegmentSource{IsExternalCollection: true}
|
|
targetManifest, err := transformManifestPath(sourceManifest, source, target)
|
|
assert.NoError(t, err)
|
|
|
|
basePath, version, err := packed.UnmarshalManifestPath(targetManifest)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(2), version)
|
|
assert.Equal(t, "files/insert_log/444/555/666", basePath)
|
|
}
|
|
|
|
func TestTransformManifestPath_LegacyExternalTableLayoutUnsupported(t *testing.T) {
|
|
target := &datapb.CopySegmentTarget{
|
|
SegmentId: 666,
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
}
|
|
|
|
sourceManifest := packed.MarshalManifestPath("external/111/segments/333", 2)
|
|
source := &datapb.CopySegmentSource{IsExternalCollection: true}
|
|
targetManifest, err := transformManifestPath(sourceManifest, source, target)
|
|
assert.Error(t, err)
|
|
assert.Empty(t, targetManifest)
|
|
assert.Contains(t, err.Error(), "invalid binlog path structure")
|
|
}
|
|
|
|
func TestTransformFieldBinlogs_SkipsPathMappingForExternalTable(t *testing.T) {
|
|
src := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{
|
|
LogID: 10,
|
|
EntriesNum: 123,
|
|
MemorySize: 456,
|
|
LogSize: 456,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
got, totalRows, err := transformFieldBinlogs(src, nil, true, true)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(123), totalRows)
|
|
assert.Len(t, got, 1)
|
|
assert.Equal(t, int64(100), got[0].GetFieldID())
|
|
assert.Len(t, got[0].GetBinlogs(), 1)
|
|
assert.Empty(t, got[0].GetBinlogs()[0].GetLogPath())
|
|
assert.Equal(t, int64(10), got[0].GetBinlogs()[0].GetLogID())
|
|
assert.Equal(t, int64(123), got[0].GetBinlogs()[0].GetEntriesNum())
|
|
assert.Equal(t, int64(456), got[0].GetBinlogs()[0].GetMemorySize())
|
|
}
|
|
|
|
func TestTransformFieldBinlogs_SkipsEmptyPathForInternalTable(t *testing.T) {
|
|
src := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogID: 10, EntriesNum: 123},
|
|
},
|
|
},
|
|
}
|
|
|
|
got, totalRows, err := transformFieldBinlogs(src, nil, false, false)
|
|
assert.NoError(t, err)
|
|
assert.Zero(t, totalRows)
|
|
assert.Empty(t, got)
|
|
}
|
|
|
|
func TestGenerateSegmentInfoFromSource_PreservesExternalTableBinlogMetadata(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: packed.MarshalManifestPath("files/insert_log/111/222/333", 2),
|
|
IsExternalCollection: true,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 0,
|
|
ChildFields: []int64{100, 101, 102},
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogID: 10, EntriesNum: 123, MemorySize: 456},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
segmentInfo, err := generateSegmentInfoFromSource(source, target, nil)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(123), segmentInfo.GetImportedRows())
|
|
assert.Len(t, segmentInfo.GetBinlogs(), 1)
|
|
assert.Empty(t, segmentInfo.GetBinlogs()[0].GetBinlogs()[0].GetLogPath())
|
|
}
|
|
|
|
func TestCopySegmentAndIndexFiles_ExternalTable(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: packed.MarshalManifestPath("files/insert_log/111/222/333", 2),
|
|
IsExternalCollection: true,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 0,
|
|
ChildFields: []int64{100, 101, 102},
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogID: 10, EntriesNum: 123, MemorySize: 456},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mList := mockey.Mock(listAllFiles).Return([]string{
|
|
"files/insert_log/111/222/333/_data/0",
|
|
"files/insert_log/111/222/333/_metadata/manifest.json",
|
|
}, nil).Build()
|
|
defer mList.UnPatch()
|
|
|
|
copied := make(map[string]string)
|
|
mCopy := mockey.Mock(copyFile).
|
|
To(func(_ context.Context, _ storage.ChunkManager, src, dst string) error {
|
|
copied[src] = dst
|
|
return nil
|
|
}).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, int64(123), result.GetImportedRows())
|
|
assert.Equal(t, packed.MarshalManifestPath("files/insert_log/444/555/666", 2), result.GetManifestPath())
|
|
assert.Len(t, result.GetBinlogs(), 1)
|
|
assert.Empty(t, result.GetBinlogs()[0].GetBinlogs()[0].GetLogPath())
|
|
assert.Equal(t, "files/insert_log/444/555/666/_data/0", copied["files/insert_log/111/222/333/_data/0"])
|
|
assert.Equal(t, "files/insert_log/444/555/666/_metadata/manifest.json", copied["files/insert_log/111/222/333/_metadata/manifest.json"])
|
|
assert.ElementsMatch(t, []string{
|
|
"files/insert_log/444/555/666/_data/0",
|
|
"files/insert_log/444/555/666/_metadata/manifest.json",
|
|
}, copiedFiles)
|
|
}
|
|
|
|
func TestCollectSegmentFiles_WithManifest(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
mList := mockey.Mock(listAllFiles).Return([]string{
|
|
"files/insert_log/111/222/333/_data/100/file1.log",
|
|
"files/insert_log/111/222/333/_data/101/file2.log",
|
|
"files/insert_log/111/222/333/_metadata/manifest.json",
|
|
}, nil).Build()
|
|
defer mList.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{},
|
|
}
|
|
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.NoError(t, err)
|
|
assert.Len(t, files.InsertBinlogs, 3)
|
|
assert.Contains(t, files.InsertBinlogs, "files/insert_log/111/222/333/_data/100/file1.log")
|
|
assert.Contains(t, files.InsertBinlogs, "files/insert_log/111/222/333/_data/101/file2.log")
|
|
assert.Contains(t, files.InsertBinlogs, "files/insert_log/111/222/333/_metadata/manifest.json")
|
|
}
|
|
|
|
func TestCollectSegmentFiles_V3MissingManifestPath(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: "", // V3 but no manifest
|
|
}
|
|
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, files)
|
|
assert.Contains(t, err.Error(), "requires manifest_path but it is empty")
|
|
}
|
|
|
|
func TestCollectSegmentFiles_ManifestUnmarshalFailure(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: "invalid-json-not-a-manifest",
|
|
}
|
|
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, files)
|
|
assert.Contains(t, err.Error(), "failed to unmarshal manifest path")
|
|
}
|
|
|
|
func TestCollectSegmentFiles_ManifestListFailure(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
mList := mockey.Mock(listAllFiles).Return(nil, errors.New("storage unavailable")).Build()
|
|
defer mList.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
}
|
|
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, files)
|
|
assert.Contains(t, err.Error(), "failed to list files from manifest base path")
|
|
}
|
|
|
|
func TestCollectSegmentFiles_ManifestEmptyFiles(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
mList := mockey.Mock(listAllFiles).Return([]string{}, nil).Build()
|
|
defer mList.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{}, // empty pb — should NOT be used
|
|
}
|
|
|
|
// Empty file list is OK for V3 — segment may have only deltas
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, files)
|
|
assert.Empty(t, files.InsertBinlogs)
|
|
}
|
|
|
|
func TestCollectSegmentFiles_NoManifest(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV1, // V1: use pb paths
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/insert_log/111/222/333/100/file1.log"},
|
|
{LogPath: "files/insert_log/111/222/333/100/file2.log"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
files, err := collectSegmentFiles(ctx, cm, source)
|
|
|
|
assert.NoError(t, err)
|
|
assert.Len(t, files.InsertBinlogs, 2)
|
|
assert.Contains(t, files.InsertBinlogs, "files/insert_log/111/222/333/100/file1.log")
|
|
assert.Contains(t, files.InsertBinlogs, "files/insert_log/111/222/333/100/file2.log")
|
|
}
|
|
|
|
func TestExtractFromPb(t *testing.T) {
|
|
t.Run("extract paths from field binlogs", func(t *testing.T) {
|
|
fieldBinlogs := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "path/a"},
|
|
{LogPath: "path/b"},
|
|
},
|
|
},
|
|
{
|
|
FieldID: 101,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "path/c"},
|
|
},
|
|
},
|
|
}
|
|
paths := extractFromPb(fieldBinlogs)
|
|
assert.Equal(t, []string{"path/a", "path/b", "path/c"}, paths)
|
|
})
|
|
|
|
t.Run("skip empty paths", func(t *testing.T) {
|
|
fieldBinlogs := []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "path/a"},
|
|
{LogPath: ""},
|
|
},
|
|
},
|
|
}
|
|
paths := extractFromPb(fieldBinlogs)
|
|
assert.Equal(t, []string{"path/a"}, paths)
|
|
})
|
|
|
|
t.Run("empty input", func(t *testing.T) {
|
|
paths := extractFromPb(nil)
|
|
assert.Empty(t, paths)
|
|
})
|
|
}
|
|
|
|
func TestExtractIndexFiles(t *testing.T) {
|
|
t.Run("extract index file paths", func(t *testing.T) {
|
|
indexInfos := []*indexpb.IndexFilePathInfo{
|
|
{IndexFilePaths: []string{"idx/a", "idx/b"}},
|
|
{IndexFilePaths: []string{"idx/c"}},
|
|
}
|
|
paths := extractIndexFiles(indexInfos)
|
|
assert.Equal(t, []string{"idx/a", "idx/b", "idx/c"}, paths)
|
|
})
|
|
|
|
t.Run("empty input", func(t *testing.T) {
|
|
paths := extractIndexFiles(nil)
|
|
assert.Empty(t, paths)
|
|
})
|
|
}
|
|
|
|
func TestExtractTextIndexFiles(t *testing.T) {
|
|
t.Run("extract text index file paths", func(t *testing.T) {
|
|
textInfos := map[int64]*datapb.TextIndexStats{
|
|
100: {Files: []string{"text/a", "text/b"}},
|
|
}
|
|
paths := extractTextIndexFiles(textInfos)
|
|
assert.Equal(t, []string{"text/a", "text/b"}, paths)
|
|
})
|
|
|
|
t.Run("empty input", func(t *testing.T) {
|
|
paths := extractTextIndexFiles(nil)
|
|
assert.Empty(t, paths)
|
|
})
|
|
}
|
|
|
|
func TestExtractJsonFiles(t *testing.T) {
|
|
t.Run("separate by data format", func(t *testing.T) {
|
|
jsonInfos := map[int64]*datapb.JsonKeyStats{
|
|
100: {
|
|
JsonKeyStatsDataFormat: 1,
|
|
Files: []string{"legacy/a"},
|
|
},
|
|
101: {
|
|
JsonKeyStatsDataFormat: 2,
|
|
Files: []string{"new/b", "new/c"},
|
|
},
|
|
}
|
|
jsonKeyFiles, jsonStatsFiles := extractJSONFiles(jsonInfos)
|
|
assert.Equal(t, []string{"legacy/a"}, jsonKeyFiles)
|
|
assert.ElementsMatch(t, []string{"new/b", "new/c"}, jsonStatsFiles)
|
|
})
|
|
|
|
t.Run("empty input", func(t *testing.T) {
|
|
jsonKeyFiles, jsonStatsFiles := extractJSONFiles(nil)
|
|
assert.Empty(t, jsonKeyFiles)
|
|
assert.Empty(t, jsonStatsFiles)
|
|
})
|
|
}
|
|
|
|
func TestGenerateMappingsFromFiles(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
}
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
t.Run("generate mappings for all file types", func(t *testing.T) {
|
|
files := &SegmentFiles{
|
|
InsertBinlogs: []string{"files/insert_log/111/222/333/100/log1"},
|
|
DeltaBinlogs: []string{"files/delta_log/111/222/333/100/delta1"},
|
|
StatsBinlogs: []string{"files/stats_log/111/222/333/100/stats1"},
|
|
Bm25Binlogs: []string{"files/bm25_stats/111/222/333/100/bm25_1"},
|
|
VectorScalarIndex: []string{"files/index_files/1002/1/222/333/idx1"},
|
|
TextIndex: []string{"files/text_log/123/1/111/222/333/100/text1"},
|
|
JSONKeyIndex: []string{"files/json_key_index_log/123/1/111/222/333/101/json1"},
|
|
JSONStats: []string{"files/json_stats/2/3002/1/111/222/333/102/shared_key_index/idx1"},
|
|
}
|
|
|
|
mappings, err := generateMappingsFromFiles(files, source, target)
|
|
assert.NoError(t, err)
|
|
assert.Len(t, mappings, 8)
|
|
|
|
assert.Equal(t, "files/insert_log/444/555/666/100/log1",
|
|
mappings["files/insert_log/111/222/333/100/log1"])
|
|
assert.Equal(t, "files/delta_log/444/555/666/100/delta1",
|
|
mappings["files/delta_log/111/222/333/100/delta1"])
|
|
assert.Equal(t, "files/bm25_stats/444/555/666/100/bm25_1",
|
|
mappings["files/bm25_stats/111/222/333/100/bm25_1"])
|
|
assert.Equal(t, "files/index_files/1002/1/555/666/idx1",
|
|
mappings["files/index_files/1002/1/222/333/idx1"])
|
|
})
|
|
|
|
t.Run("empty files", func(t *testing.T) {
|
|
files := &SegmentFiles{}
|
|
mappings, err := generateMappingsFromFiles(files, source, target)
|
|
assert.NoError(t, err)
|
|
assert.Empty(t, mappings)
|
|
})
|
|
|
|
t.Run("invalid path returns error", func(t *testing.T) {
|
|
files := &SegmentFiles{
|
|
InsertBinlogs: []string{"invalid/path/no/log_type"},
|
|
}
|
|
_, err := generateMappingsFromFiles(files, source, target)
|
|
assert.Error(t, err)
|
|
})
|
|
}
|
|
|
|
// TestCopySegmentAndIndexFiles_WithManifest tests the complete copy flow for StorageV3 segments.
|
|
// When storage_version >= StorageV3, insert binlogs are stored in packed format.
|
|
// Physical files (under _data/_metadata) differ from logical paths in protobuf.
|
|
// Step 3.5 must add logical pb path mappings AFTER file copying for metadata generation.
|
|
func TestCopySegmentAndIndexFiles_WithManifest(t *testing.T) {
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 500, LogPath: "files/insert_log/111/222/333/100/10001", LogSize: 1024},
|
|
},
|
|
},
|
|
{
|
|
FieldID: 101,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 500, LogPath: "files/insert_log/111/222/333/101/10002", LogSize: 2048},
|
|
},
|
|
},
|
|
},
|
|
StatsBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/stats_log/111/222/333/100/20001"},
|
|
},
|
|
},
|
|
},
|
|
DeltaBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 0,
|
|
Binlogs: []*datapb.Binlog{
|
|
{LogPath: "files/delta_log/111/222/333/0/30001"},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
t.Run("successful copy with manifest", func(t *testing.T) {
|
|
mList := mockey.Mock(listAllFiles).To(func(_ context.Context, _ storage.ChunkManager, basePath string) ([]string, error) {
|
|
return []string{
|
|
"files/insert_log/111/222/333/_data/0",
|
|
"files/insert_log/111/222/333/_metadata/manifest.json",
|
|
}, nil
|
|
}).Build()
|
|
defer mList.UnPatch()
|
|
|
|
copiedSrcPaths := make([]string, 0)
|
|
mCopy := mockey.Mock(copyFile).
|
|
To(func(_ context.Context, _ storage.ChunkManager, src, dst string) error {
|
|
copiedSrcPaths = append(copiedSrcPaths, src)
|
|
return nil
|
|
}).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
|
|
assert.Equal(t, int64(666), result.SegmentId)
|
|
assert.Equal(t, int64(1000), result.ImportedRows)
|
|
|
|
assert.Equal(t, 2, len(result.Binlogs))
|
|
assert.Equal(t, 1, len(result.Statslogs))
|
|
assert.Equal(t, 1, len(result.Deltalogs))
|
|
|
|
expectedManifestPath := packed.MarshalManifestPath("files/insert_log/444/555/666", 2)
|
|
assert.Equal(t, expectedManifestPath, result.ManifestPath)
|
|
|
|
assert.Len(t, copiedFiles, 4)
|
|
|
|
for _, src := range copiedSrcPaths {
|
|
assert.NotEqual(t, "files/insert_log/111/222/333/100/10001", src)
|
|
assert.NotEqual(t, "files/insert_log/111/222/333/101/10002", src)
|
|
}
|
|
|
|
assert.Contains(t, copiedSrcPaths, "files/insert_log/111/222/333/_data/0")
|
|
assert.Contains(t, copiedSrcPaths, "files/insert_log/111/222/333/_metadata/manifest.json")
|
|
assert.Contains(t, copiedSrcPaths, "files/stats_log/111/222/333/100/20001")
|
|
assert.Contains(t, copiedSrcPaths, "files/delta_log/111/222/333/0/30001")
|
|
})
|
|
|
|
t.Run("copy failure on physical file", func(t *testing.T) {
|
|
mList := mockey.Mock(listAllFiles).To(func(_ context.Context, _ storage.ChunkManager, basePath string) ([]string, error) {
|
|
return []string{"files/insert_log/111/222/333/_data/0"}, nil
|
|
}).Build()
|
|
defer mList.UnPatch()
|
|
|
|
mCopy := mockey.Mock(copyFile).Return(errors.New("storage unavailable")).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, result)
|
|
assert.Contains(t, err.Error(), "failed to copy file")
|
|
assert.Empty(t, copiedFiles)
|
|
})
|
|
|
|
t.Run("step 3.5 fails on invalid pb path format", func(t *testing.T) {
|
|
badPbSource := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 500, LogPath: "invalid/path/no/log_type/10001", LogSize: 1024},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
mList := mockey.Mock(listAllFiles).Return([]string{"files/insert_log/111/222/333/_data/0"}, nil).Build()
|
|
defer mList.UnPatch()
|
|
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, _, err := CopySegmentAndIndexFiles(context.Background(), cm, badPbSource, target, nil)
|
|
|
|
assert.Error(t, err)
|
|
assert.Nil(t, result)
|
|
assert.Contains(t, err.Error(), "failed to generate target path for pb insert binlog")
|
|
})
|
|
}
|
|
|
|
// TestCopySegmentAndIndexFiles_WithManifest_NoPbBinlogs tests StorageV3 segment with no pb insert binlogs.
|
|
// When InsertBinlogs in pb is empty but manifest has files, Step 3.5 has nothing to add (no logical paths).
|
|
func TestCopySegmentAndIndexFiles_WithManifest_NoPbBinlogs(t *testing.T) {
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mList := mockey.Mock(listAllFiles).Return([]string{"files/insert_log/111/222/333/_data/0"}, nil).Build()
|
|
defer mList.UnPatch()
|
|
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
assert.Equal(t, int64(0), result.ImportedRows)
|
|
assert.Len(t, copiedFiles, 1)
|
|
assert.NotEmpty(t, result.ManifestPath)
|
|
}
|
|
|
|
// TestCopySegmentAndIndexFiles_WithoutManifest_Unchanged verifies that StorageV1 segments
|
|
// use pb paths and do NOT trigger manifest-related logic (Step 3.5, Step 8).
|
|
func TestCopySegmentAndIndexFiles_WithoutManifest_Unchanged(t *testing.T) {
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV1,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 1000, LogPath: "files/insert_log/111/222/333/100/10001", LogSize: 1024},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
}
|
|
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
assert.Equal(t, int64(1000), result.ImportedRows)
|
|
assert.Equal(t, int64(666), result.SegmentId)
|
|
assert.Len(t, copiedFiles, 1)
|
|
assert.Empty(t, result.ManifestPath)
|
|
}
|
|
|
|
func TestListAllFiles(t *testing.T) {
|
|
t.Run("list files successfully", func(t *testing.T) {
|
|
mWalk := mockey.Mock((*storage.LocalChunkManager).WalkWithPrefix).
|
|
To(func(_ *storage.LocalChunkManager, _ context.Context, prefix string, recursive bool, walkFunc storage.ChunkObjectWalkFunc) error {
|
|
walkFunc(&storage.ChunkObjectInfo{FilePath: "base/path/file1"})
|
|
walkFunc(&storage.ChunkObjectInfo{FilePath: "base/path/file2"})
|
|
return nil
|
|
}).Build()
|
|
defer mWalk.UnPatch()
|
|
|
|
cm := &storage.LocalChunkManager{}
|
|
files, err := listAllFiles(context.Background(), cm, "base/path")
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, []string{"base/path/file1", "base/path/file2"}, files)
|
|
})
|
|
|
|
t.Run("walk error returns error", func(t *testing.T) {
|
|
mWalk := mockey.Mock((*storage.LocalChunkManager).WalkWithPrefix).
|
|
Return(errors.New("walk failed")).Build()
|
|
defer mWalk.UnPatch()
|
|
|
|
cm := &storage.LocalChunkManager{}
|
|
files, err := listAllFiles(context.Background(), cm, "bad/path")
|
|
assert.Error(t, err)
|
|
assert.Nil(t, files)
|
|
})
|
|
}
|
|
|
|
// TestCopySegmentAndIndexFiles_V3WithTextAndJsonStats verifies that V3 segments
|
|
// with text index and JSON key stats succeed during copy. Before the fix, V3
|
|
// segments had wrong-format paths in TextIndexFiles/JsonKeyIndexFiles (etcd
|
|
// metadata), causing "no mapping found" errors during buildIndexInfoFromSource.
|
|
// The fix skips pb path extraction for V3 (files already copied via manifest
|
|
// basePath/_stats/) and passes metadata as placeholders.
|
|
func TestCopySegmentAndIndexFiles_V3WithTextAndJsonStats(t *testing.T) {
|
|
manifestPath := packed.MarshalManifestPath("files/insert_log/111/222/333", 2)
|
|
|
|
source := &datapb.CopySegmentSource{
|
|
CollectionId: 111,
|
|
PartitionId: 222,
|
|
SegmentId: 333,
|
|
StorageVersion: storage.StorageV3,
|
|
ManifestPath: manifestPath,
|
|
InsertBinlogs: []*datapb.FieldBinlog{
|
|
{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{
|
|
{EntriesNum: 500, LogPath: "files/insert_log/111/222/333/100/10001", LogSize: 1024},
|
|
},
|
|
},
|
|
},
|
|
// Text index with relative paths (as stored in etcd after dual-write fix)
|
|
TextIndexFiles: map[int64]*datapb.TextIndexStats{
|
|
101: {
|
|
FieldID: 101,
|
|
BuildID: 7000,
|
|
Version: 1,
|
|
Files: []string{"tokenizer.json", "index.data"},
|
|
},
|
|
},
|
|
// JSON key stats with relative paths (as stored in etcd after dual-write fix)
|
|
JsonKeyIndexFiles: map[int64]*datapb.JsonKeyStats{
|
|
102: {
|
|
FieldID: 102,
|
|
BuildID: 8000,
|
|
Version: 1,
|
|
Files: []string{"shared_key_index/.managed.json_0", "shared_key_index/.managed.json_1"},
|
|
JsonKeyStatsDataFormat: 3,
|
|
},
|
|
},
|
|
}
|
|
|
|
target := &datapb.CopySegmentTarget{
|
|
CollectionId: 444,
|
|
PartitionId: 555,
|
|
SegmentId: 666,
|
|
NewBuildIds: map[int64]int64{7000: 7777, 8000: 9000},
|
|
}
|
|
|
|
mList := mockey.Mock(listAllFiles).To(func(_ context.Context, _ storage.ChunkManager, basePath string) ([]string, error) {
|
|
// Simulate listing all files under basePath including _stats/
|
|
return []string{
|
|
"files/insert_log/111/222/333/_data/0",
|
|
"files/insert_log/111/222/333/_metadata/manifest.json",
|
|
"files/insert_log/111/222/333/_stats/text_index.101/tokenizer.json",
|
|
"files/insert_log/111/222/333/_stats/text_index.101/index.data",
|
|
"files/insert_log/111/222/333/_stats/json_stats.102/shared_key_index/.managed.json_0",
|
|
"files/insert_log/111/222/333/_stats/json_stats.102/shared_key_index/.managed.json_1",
|
|
}, nil
|
|
}).Build()
|
|
defer mList.UnPatch()
|
|
|
|
mCopy := mockey.Mock(copyFile).Return(nil).Build()
|
|
defer mCopy.UnPatch()
|
|
|
|
cm := &struct{ storage.ChunkManager }{}
|
|
result, copiedFiles, err := CopySegmentAndIndexFiles(context.Background(), cm, source, target, nil)
|
|
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, result)
|
|
assert.Equal(t, int64(666), result.SegmentId)
|
|
|
|
// Verify text index metadata is passed through as placeholder
|
|
assert.Len(t, result.TextIndexInfos, 1)
|
|
assert.Contains(t, result.TextIndexInfos, int64(101))
|
|
// Files are passed through as-is (placeholder, not mapped)
|
|
assert.Equal(t, []string{"tokenizer.json", "index.data"}, result.TextIndexInfos[101].GetFiles())
|
|
assert.Equal(t, int64(7777), result.TextIndexInfos[101].GetBuildID(), "text index buildID should be remapped")
|
|
|
|
// Verify JSON key stats metadata is passed through as placeholder with new buildID
|
|
assert.Len(t, result.JsonKeyIndexInfos, 1)
|
|
assert.Contains(t, result.JsonKeyIndexInfos, int64(102))
|
|
assert.Equal(t, int64(9000), result.JsonKeyIndexInfos[102].GetBuildID(), "buildID should be remapped")
|
|
// Files are passed through as-is (placeholder, not mapped)
|
|
assert.Equal(t, []string{"shared_key_index/.managed.json_0", "shared_key_index/.managed.json_1"},
|
|
result.JsonKeyIndexInfos[102].GetFiles())
|
|
|
|
// Verify manifest path is transformed
|
|
expectedManifestPath := packed.MarshalManifestPath("files/insert_log/444/555/666", 2)
|
|
assert.Equal(t, expectedManifestPath, result.ManifestPath)
|
|
|
|
// Verify _stats files are included in the copy (from listAllFiles, part of InsertBinlogs)
|
|
assert.True(t, len(copiedFiles) >= 6, "should copy all files including _stats/")
|
|
}
|