Files
milvus/internal/compaction/lob_compaction_test.go
b16c821b36 enhance: carry TEXT LOB refs through schema-bump full-rewrite compaction (#51125)
### What & why

`bump_schema_version`'s full-rewrite path (`runFullSchemaRewrite`,
reached only for a **drop-field** schema bump) builds a from-scratch
output manifest via `NewBinlogRecordWriter` and did **not** carry the
source segment's TEXT LOB references — leaving dangling refs for an
out-of-line (`>=64KB`) TEXT column. This was the `TODO(#50021)`.

### What this does

Wire **REUSE_ALL** LOB handling into `runFullSchemaRewrite`, mirroring
sort compaction. Both are `1->1` with a single output manifest, so
REUSE_ALL is always correct: a schema bump never changes the existing
TEXT LOB data — only its references are carried.

- `internal/compaction/lob_compaction.go`: `GetForcedStrategy` forces
REUSE_ALL for `BumpSchemaVersionCompaction`.
- `internal/datanode/compactor/bump_schema_version_compactor.go`:
collect the source LOB files into a `LOBCompactionContext`, pass
`storage.WithTextRefsAsBinary()`, update per-LOB-file `valid_rows`
(`SetSegmentRowStats`), and merge the LOB references into the output
manifest (`applyLOBCompaction`) **before** building the text-match index
— `createTextIndex` reads the TEXT column through this manifest, so the
carried LOB files must already be referenced (matches sort/mix
ordering). Removed the `TODO(#50021)`.
- Unit tests: forced strategy, init/apply LOB wiring, the
applyLOBCompaction-before-createTextIndex ordering
(`TestFullRewriteMergesLOBRefsBeforeBuildingTextIndex`), and a real
add-nullable-TEXT + drop-field full-rewrite regression driving the
actual packed writer
(`TestFullRewriteFillsMissingNullableTextAsBinary`).

### Also: guard add_function_field behind StorageV3 (#51167)

Adding a function to an existing collection must backfill the new output
field into **pre-existing** sealed segments; that backfill runs through
`bump_schema_version` compaction, which only works on a StorageV3
segment. A pre-existing V2 segment is only backfilled after the
storage-version upgrade compaction rewrites it to V3, and that upgrade
runs only when **both** `common.storage.useLoonFFI` and
`dataCoord.compaction.storageVersion.enabled` are on. The proxy now
rejects add-function unless both are enabled
(`validateAddFunctionRequiresStorageV3`, wired into
`addCollectionFunctionTask` and `alterCollectionSchemaTask`).
`create_collection` with a function is unaffected. Full reasoning in
#51167.

### Notes

- Reuses the LOB machinery from **#50784**.
- The `GenerateEmptyArrayFromSchema` binary-TEXT production fix landed
independently on master via **#51124**; this PR keeps the regression
test guarding the bump path.

issue: #50021, #51167

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-09 17:20:34 +08:00

895 lines
32 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 compaction
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
"github.com/milvus-io/milvus/internal/storagev2/packed"
"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
)
func TestGetForcedStrategy_AllTypes(t *testing.T) {
tests := []struct {
name string
compType datapb.CompactionType
sourceCount int
targetCount int
wantStrategy LOBCompactionStrategy
wantForced bool
}{
{
name: "clustering compaction",
compType: datapb.CompactionType_ClusteringCompaction,
sourceCount: 3, targetCount: 3,
wantStrategy: LOBStrategyRewriteAll, wantForced: true,
},
{
name: "sort compaction",
compType: datapb.CompactionType_SortCompaction,
sourceCount: 1, targetCount: 1,
wantStrategy: LOBStrategyReuseAll, wantForced: true,
},
{
name: "L0 delete compaction",
compType: datapb.CompactionType_Level0DeleteCompaction,
sourceCount: 1, targetCount: 1,
wantStrategy: LOBStrategySkip, wantForced: true,
},
{
name: "mix compaction split 1->N",
compType: datapb.CompactionType_MixCompaction,
sourceCount: 1, targetCount: 3,
wantStrategy: LOBStrategyRewriteAll, wantForced: true,
},
{
name: "mix compaction N->M forced",
compType: datapb.CompactionType_MixCompaction,
sourceCount: 3, targetCount: 2,
wantStrategy: LOBStrategyRewriteAll, wantForced: true,
},
{
name: "mix compaction N->1 not forced",
compType: datapb.CompactionType_MixCompaction,
sourceCount: 3, targetCount: 1,
wantStrategy: 0, wantForced: false,
},
{
name: "mix compaction 1->1 not forced",
compType: datapb.CompactionType_MixCompaction,
sourceCount: 1, targetCount: 1,
wantStrategy: 0, wantForced: false,
},
{
// schema-bump is 1->1 and never changes existing TEXT LOB data, so it
// must force REUSE_ALL (never fall through to hole-ratio REWRITE_ALL).
name: "schema bump compaction",
compType: datapb.CompactionType_BumpSchemaVersionCompaction,
sourceCount: 1, targetCount: 1,
wantStrategy: LOBStrategyReuseAll, wantForced: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
strategy, forced := GetForcedStrategy(tt.compType, tt.sourceCount, tt.targetCount)
assert.Equal(t, tt.wantForced, forced)
if forced {
assert.Equal(t, tt.wantStrategy, strategy)
}
})
}
}
func TestLOBCompactionContext_GetTextColumnConfigs_EdgeCases(t *testing.T) {
t.Run("empty partition path", func(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/f.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 200},
})
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
configs := ctx.GetTextColumnConfigs("", DefaultTextInlineThreshold, DefaultTextMaxLobFileBytes, DefaultTextFlushThresholdBytes)
assert.Len(t, configs, 1)
assert.Equal(t, "/lobs/101", configs[0].LobBasePath)
})
t.Run("zero thresholds", func(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/f.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 200},
})
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
configs := ctx.GetTextColumnConfigs("/base", 0, 0, 0)
assert.Len(t, configs, 1)
assert.Equal(t, int64(0), configs[0].InlineThreshold)
assert.Equal(t, int64(0), configs[0].MaxLobFileBytes)
})
t.Run("multiple REWRITE_ALL fields", func(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/f1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 200},
{Path: "/lob/f2.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 100},
{Path: "/lob/f3.vortex", FieldID: 103, TotalRows: 1000, ValidRows: 300},
})
ctx.ComputeStrategies([]int64{101, 102, 103}, DefaultLOBHoleRatioThreshold)
configs := ctx.GetTextColumnConfigs("/base", DefaultTextInlineThreshold, DefaultTextMaxLobFileBytes, DefaultTextFlushThresholdBytes)
assert.Len(t, configs, 3)
fieldIDs := make(map[int64]bool)
for _, c := range configs {
fieldIDs[c.FieldID] = true
}
assert.True(t, fieldIDs[101])
assert.True(t, fieldIDs[102])
assert.True(t, fieldIDs[103])
})
}
func TestDecideLOBStrategyFromManifest_ReuseAll(t *testing.T) {
// Test case: hole ratio < 30%, should return REUSE_ALL
// file_1001: total_rows=1000, valid_rows=900 (shared by two segments)
// hole_ratio = 1 - 900/1000 = 10% -> REUSE_ALL
fieldID := int64(101)
lobFiles := []packed.LobFileInfo{
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 500},
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 400}, // same file, different segment
}
decision := DecideLOBStrategyFromManifest(lobFiles, fieldID, DefaultLOBHoleRatioThreshold)
assert.Equal(t, fieldID, decision.FieldID)
assert.Equal(t, LOBStrategyReuseAll, decision.Strategy)
assert.Equal(t, int64(900), decision.TotalValidRows)
assert.Equal(t, int64(1000), decision.TotalRows) // file counted once
assert.InDelta(t, 0.1, decision.OverallHoleRatio, 0.001)
}
func TestDecideLOBStrategyFromManifest_RewriteAll(t *testing.T) {
// Test case: hole ratio >= 30%, should return REWRITE_ALL
// file_1001: total_rows=1000, valid_rows=700 (from two segments: 500+200)
// file_1002: total_rows=800, valid_rows=300
// file_1003: total_rows=500, valid_rows=100
// total_valid_rows = 700 + 300 + 100 = 1100
// total_rows = 1000 + 800 + 500 = 2300
// hole_ratio = 1 - 1100/2300 = 52.2% -> REWRITE_ALL
fieldID := int64(101)
lobFiles := []packed.LobFileInfo{
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 500},
{Path: "/lob/file_1002.vortex", FieldID: fieldID, TotalRows: 800, ValidRows: 300},
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 200}, // same file
{Path: "/lob/file_1003.vortex", FieldID: fieldID, TotalRows: 500, ValidRows: 100},
}
decision := DecideLOBStrategyFromManifest(lobFiles, fieldID, DefaultLOBHoleRatioThreshold)
assert.Equal(t, fieldID, decision.FieldID)
assert.Equal(t, LOBStrategyRewriteAll, decision.Strategy)
assert.Equal(t, int64(1100), decision.TotalValidRows)
assert.Equal(t, int64(2300), decision.TotalRows)
assert.InDelta(t, 0.522, decision.OverallHoleRatio, 0.01)
}
func TestDecideLOBStrategyFromManifest_EmptyFiles(t *testing.T) {
fieldID := int64(101)
// Test with empty LOB files
decision := DecideLOBStrategyFromManifest([]packed.LobFileInfo{}, fieldID, DefaultLOBHoleRatioThreshold)
assert.Equal(t, LOBStrategyReuseAll, decision.Strategy)
assert.Equal(t, int64(0), decision.TotalValidRows)
assert.Equal(t, int64(0), decision.TotalRows)
assert.Equal(t, float64(0), decision.OverallHoleRatio)
}
func TestDecideLOBStrategyFromManifest_NoMatchingField(t *testing.T) {
fieldID := int64(101)
// Test with LOB files for different field
lobFiles := []packed.LobFileInfo{
{Path: "/lob/file_1001.vortex", FieldID: 999, TotalRows: 1000, ValidRows: 500},
}
decision := DecideLOBStrategyFromManifest(lobFiles, fieldID, DefaultLOBHoleRatioThreshold)
assert.Equal(t, LOBStrategyReuseAll, decision.Strategy)
assert.Equal(t, int64(0), decision.TotalValidRows)
assert.Equal(t, int64(0), decision.TotalRows)
}
func TestDecideLOBStrategyFromManifest_CustomThreshold(t *testing.T) {
fieldID := int64(101)
lobFiles := []packed.LobFileInfo{
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 850}, // 15% hole
}
// With default threshold (30%), should be REUSE_ALL
decision := DecideLOBStrategyFromManifest(lobFiles, fieldID, DefaultLOBHoleRatioThreshold)
assert.Equal(t, LOBStrategyReuseAll, decision.Strategy)
// With lower threshold (10%), should be REWRITE_ALL
decision = DecideLOBStrategyFromManifest(lobFiles, fieldID, 0.10)
assert.Equal(t, LOBStrategyRewriteAll, decision.Strategy)
}
func TestMergeLOBFiles_Basic(t *testing.T) {
fieldID := int64(101)
allLobFiles := [][]packed.LobFileInfo{
{
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 500},
{Path: "/lob/file_1002.vortex", FieldID: fieldID, TotalRows: 800, ValidRows: 300},
},
{
{Path: "/lob/file_1001.vortex", FieldID: fieldID, TotalRows: 1000, ValidRows: 200},
{Path: "/lob/file_1003.vortex", FieldID: fieldID, TotalRows: 500, ValidRows: 100},
},
}
merged := MergeLOBFiles(allLobFiles, fieldID)
assert.Len(t, merged, 3)
// Build a map for easier assertion
fileMap := make(map[string]packed.LobFileInfo)
for _, f := range merged {
fileMap[f.Path] = f
}
// file_1001: valid_rows should be accumulated (500 + 200 = 700)
assert.Equal(t, int64(1000), fileMap["/lob/file_1001.vortex"].TotalRows)
assert.Equal(t, int64(700), fileMap["/lob/file_1001.vortex"].ValidRows)
// file_1002: valid_rows = 300
assert.Equal(t, int64(800), fileMap["/lob/file_1002.vortex"].TotalRows)
assert.Equal(t, int64(300), fileMap["/lob/file_1002.vortex"].ValidRows)
// file_1003: valid_rows = 100
assert.Equal(t, int64(500), fileMap["/lob/file_1003.vortex"].TotalRows)
assert.Equal(t, int64(100), fileMap["/lob/file_1003.vortex"].ValidRows)
}
func TestMergeLOBFiles_EmptySlices(t *testing.T) {
fieldID := int64(101)
merged := MergeLOBFiles([][]packed.LobFileInfo{}, fieldID)
assert.Len(t, merged, 0)
merged = MergeLOBFiles([][]packed.LobFileInfo{{}, {}}, fieldID)
assert.Len(t, merged, 0)
}
func TestMergeLOBFiles_NoMatchingField(t *testing.T) {
fieldID := int64(101)
allLobFiles := [][]packed.LobFileInfo{
{
{Path: "/lob/file_1001.vortex", FieldID: 999, TotalRows: 1000, ValidRows: 500},
},
}
merged := MergeLOBFiles(allLobFiles, fieldID)
assert.Len(t, merged, 0)
}
func TestGetTEXTFieldIDsFromLOBFiles(t *testing.T) {
lobFiles := []packed.LobFileInfo{
{Path: "/lob/file_1.vortex", FieldID: 101, TotalRows: 100, ValidRows: 100},
{Path: "/lob/file_2.vortex", FieldID: 102, TotalRows: 100, ValidRows: 100},
{Path: "/lob/file_3.vortex", FieldID: 101, TotalRows: 100, ValidRows: 100}, // same field
{Path: "/lob/file_4.vortex", FieldID: 103, TotalRows: 100, ValidRows: 100},
}
fieldIDs := GetTEXTFieldIDsFromLOBFiles(lobFiles)
assert.Len(t, fieldIDs, 3)
fieldIDMap := make(map[int64]bool)
for _, id := range fieldIDs {
fieldIDMap[id] = true
}
assert.True(t, fieldIDMap[101])
assert.True(t, fieldIDMap[102])
assert.True(t, fieldIDMap[103])
}
func TestGetTEXTFieldIDsFromLOBFiles_Empty(t *testing.T) {
fieldIDs := GetTEXTFieldIDsFromLOBFiles([]packed.LobFileInfo{})
assert.Len(t, fieldIDs, 0)
}
func TestGetTEXTFieldIDsFromSchema(t *testing.T) {
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64},
{FieldID: 101, Name: "text1", DataType: schemapb.DataType_Text},
{FieldID: 102, Name: "varchar", DataType: schemapb.DataType_VarChar},
{FieldID: 103, Name: "text2", DataType: schemapb.DataType_Text},
},
}
fieldIDs := GetTEXTFieldIDsFromSchema(schema)
assert.Len(t, fieldIDs, 2)
assert.Contains(t, fieldIDs, int64(101))
assert.Contains(t, fieldIDs, int64(103))
}
func TestGetTEXTFieldIDsFromSchema_NoTextFields(t *testing.T) {
schema := &schemapb.CollectionSchema{
Fields: []*schemapb.FieldSchema{
{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64},
{FieldID: 101, Name: "varchar", DataType: schemapb.DataType_VarChar},
},
}
fieldIDs := GetTEXTFieldIDsFromSchema(schema)
assert.Len(t, fieldIDs, 0)
}
func TestLOBCompactionContext_Basic(t *testing.T) {
ctx := NewLOBCompactionContext()
assert.NotNil(t, ctx)
assert.NotNil(t, ctx.Decisions)
assert.NotNil(t, ctx.AllLobFiles)
assert.NotNil(t, ctx.MergedLobFiles)
}
func TestLOBCompactionContext_AddSegmentLobFiles(t *testing.T) {
ctx := NewLOBCompactionContext()
files1 := []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 900},
}
files2 := []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 100},
}
ctx.AddSegmentLobFiles(1, files1)
ctx.AddSegmentLobFiles(2, files2)
assert.Len(t, ctx.AllLobFiles, 2)
assert.Equal(t, files1, ctx.AllLobFiles[1])
assert.Equal(t, files2, ctx.AllLobFiles[2])
}
func TestLOBCompactionContext_ComputeStrategies_ReuseAll(t *testing.T) {
ctx := NewLOBCompactionContext()
// Segment 1: 900 valid rows from file
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 900},
})
// hole_ratio = 1 - 900/1000 = 10% < 30% -> REUSE_ALL
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
assert.True(t, ctx.HasTEXTFields())
assert.True(t, ctx.IsReuseAll(101))
assert.False(t, ctx.ShouldRewriteAnyField())
merged := ctx.GetMergedLobFiles(101)
assert.Len(t, merged, 1)
assert.Equal(t, int64(900), merged[0].ValidRows)
}
func TestLOBCompactionContext_ComputeStrategies_RewriteAll(t *testing.T) {
ctx := NewLOBCompactionContext()
// Segment 1: 200 valid rows
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 200},
})
// hole_ratio = 1 - 200/1000 = 80% >= 30% -> REWRITE_ALL
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
assert.True(t, ctx.HasTEXTFields())
assert.False(t, ctx.IsReuseAll(101))
assert.Equal(t, LOBStrategyRewriteAll, ctx.GetStrategy(101))
assert.True(t, ctx.ShouldRewriteAnyField())
// For REWRITE_ALL, GetReuseAllLobFilesWithUpdatedStats should return empty
// because there are no REUSE_ALL fields
reuseAllFiles := ctx.GetReuseAllLobFilesWithUpdatedStats()
assert.Len(t, reuseAllFiles, 0)
// GetMergedLobFiles still returns all files (used for reference)
merged := ctx.GetMergedLobFiles(101)
assert.Len(t, merged, 1)
}
func TestLOBCompactionContext_ComputeStrategies_MultipleFields(t *testing.T) {
ctx := NewLOBCompactionContext()
// Field 101: 90% valid -> REUSE_ALL
// Field 102: 50% valid -> REWRITE_ALL
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101_file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 900},
{Path: "/lob/field102_file1.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 500},
})
ctx.ComputeStrategies([]int64{101, 102}, DefaultLOBHoleRatioThreshold)
assert.True(t, ctx.IsReuseAll(101))
assert.False(t, ctx.IsReuseAll(102))
assert.True(t, ctx.ShouldRewriteAnyField()) // field 102 needs rewrite
}
func TestLOBCompactionContext_GetAllMergedLobFiles(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101_file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 900},
{Path: "/lob/field102_file1.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 800},
})
ctx.ComputeStrategies([]int64{101, 102}, DefaultLOBHoleRatioThreshold)
allMerged := ctx.GetAllMergedLobFiles()
assert.Len(t, allMerged, 2) // both fields are REUSE_ALL
}
// Tests for GetForcedStrategy
func TestGetForcedStrategy_ClusterCompaction(t *testing.T) {
// Cluster compaction always forces REWRITE_ALL
strategy, forced := GetForcedStrategy(datapb.CompactionType_ClusteringCompaction, 3, 5)
assert.True(t, forced)
assert.Equal(t, LOBStrategyRewriteAll, strategy)
// ClusteringPartitionKeySortCompaction also forces REWRITE_ALL
strategy, forced = GetForcedStrategy(datapb.CompactionType_ClusteringPartitionKeySortCompaction, 3, 5)
assert.True(t, forced)
assert.Equal(t, LOBStrategyRewriteAll, strategy)
}
func TestGetForcedStrategy_SortCompaction(t *testing.T) {
// Sort compaction always forces REUSE_ALL
strategy, forced := GetForcedStrategy(datapb.CompactionType_SortCompaction, 1, 1)
assert.True(t, forced)
assert.Equal(t, LOBStrategyReuseAll, strategy)
// PartitionKeySortCompaction also forces REUSE_ALL
strategy, forced = GetForcedStrategy(datapb.CompactionType_PartitionKeySortCompaction, 1, 1)
assert.True(t, forced)
assert.Equal(t, LOBStrategyReuseAll, strategy)
}
func TestGetForcedStrategy_MixCompaction_MultiOutput(t *testing.T) {
// 1->N forces REWRITE_ALL
strategy, forced := GetForcedStrategy(datapb.CompactionType_MixCompaction, 1, 3)
assert.True(t, forced)
assert.Equal(t, LOBStrategyRewriteAll, strategy)
// N->M (M>1) also forces REWRITE_ALL
strategy, forced = GetForcedStrategy(datapb.CompactionType_MixCompaction, 3, 2)
assert.True(t, forced)
assert.Equal(t, LOBStrategyRewriteAll, strategy)
}
func TestGetForcedStrategy_MixCompaction_SingleOutput(t *testing.T) {
// N->1 uses hole ratio (not forced)
strategy, forced := GetForcedStrategy(datapb.CompactionType_MixCompaction, 3, 1)
assert.False(t, forced)
assert.Equal(t, LOBStrategyReuseAll, strategy) // default when not forced
}
func TestGetForcedStrategy_Level0DeleteCompaction(t *testing.T) {
// L0 delete compaction forces SKIP (only applies delete logs, segment LOB refs unchanged)
strategy, forced := GetForcedStrategy(datapb.CompactionType_Level0DeleteCompaction, 1, 1)
assert.True(t, forced)
assert.Equal(t, LOBStrategySkip, strategy)
}
func TestLOBCompactionContext_ComputeStrategies_Skip(t *testing.T) {
ctx := NewLOBCompactionContext()
// Add LOB files
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
})
// Set L0 delete compaction which forces SKIP
ctx.SetCompactionType(datapb.CompactionType_Level0DeleteCompaction, 1, 1)
// Compute strategies - should result in no decisions (SKIP)
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
// SKIP means no LOB processing: no decisions, no TEXT fields to handle
assert.False(t, ctx.HasTEXTFields())
assert.False(t, ctx.ShouldRewriteAnyField())
assert.False(t, ctx.HasReuseAllFields())
}
// Tests for SetCompactionType
func TestLOBCompactionContext_SetCompactionType_Forced(t *testing.T) {
ctx := NewLOBCompactionContext()
// Cluster compaction should set forced strategy
ctx.SetCompactionType(datapb.CompactionType_ClusteringCompaction, 3, 5)
assert.True(t, ctx.IsForced)
assert.Equal(t, LOBStrategyRewriteAll, ctx.ForcedStrategy)
assert.Equal(t, datapb.CompactionType_ClusteringCompaction, ctx.CompactionType)
}
func TestLOBCompactionContext_SetCompactionType_NotForced(t *testing.T) {
ctx := NewLOBCompactionContext()
// Normal mix compaction should not set forced strategy
ctx.SetCompactionType(datapb.CompactionType_MixCompaction, 3, 1)
assert.False(t, ctx.IsForced)
assert.Equal(t, datapb.CompactionType_MixCompaction, ctx.CompactionType)
}
func TestLOBCompactionContext_ComputeStrategies_WithForcedStrategy(t *testing.T) {
ctx := NewLOBCompactionContext()
// Add LOB files with high valid ratio (would normally be REUSE_ALL)
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
})
// Set cluster compaction which forces REWRITE_ALL
ctx.SetCompactionType(datapb.CompactionType_ClusteringCompaction, 1, 3)
// Compute strategies - should use forced strategy
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
// Even with high valid ratio, strategy should be REWRITE_ALL due to forced
assert.Equal(t, LOBStrategyRewriteAll, ctx.GetStrategy(101))
assert.True(t, ctx.ShouldRewriteAnyField())
}
// Tests for SegmentRowStats
func TestLOBCompactionContext_SetSegmentRowStats(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.SetSegmentRowStats(1, 1000, 100)
ctx.SetSegmentRowStats(2, 2000, 300)
assert.Equal(t, int64(1000), ctx.SegmentRowStats[1].TotalRows)
assert.Equal(t, int64(100), ctx.SegmentRowStats[1].DeletedRows)
assert.Equal(t, int64(2000), ctx.SegmentRowStats[2].TotalRows)
assert.Equal(t, int64(300), ctx.SegmentRowStats[2].DeletedRows)
}
func TestLOBCompactionContext_IncrementSegmentDeletedRows(t *testing.T) {
ctx := NewLOBCompactionContext()
// Initialize stats first
ctx.SetSegmentRowStats(1, 1000, 0)
// Increment deleted rows
ctx.IncrementSegmentDeletedRows(1, 50)
ctx.IncrementSegmentDeletedRows(1, 30)
assert.Equal(t, int64(80), ctx.SegmentRowStats[1].DeletedRows)
// Incrementing non-existent segment should not panic
ctx.IncrementSegmentDeletedRows(999, 10) // no-op
}
// Tests for GetAllMergedLobFilesWithUpdatedStats
func TestLOBCompactionContext_GetAllMergedLobFilesWithUpdatedStats(t *testing.T) {
ctx := NewLOBCompactionContext()
// Segment 1: 1000 total, 200 deleted -> survival ratio = 0.8
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/seg1_file1.vortex", FieldID: 101, TotalRows: 500, ValidRows: 500},
{Path: "/lob/seg1_file2.vortex", FieldID: 101, TotalRows: 500, ValidRows: 500},
})
ctx.SetSegmentRowStats(1, 1000, 200)
// Segment 2: 800 total, 400 deleted -> survival ratio = 0.5
ctx.AddSegmentLobFiles(2, []packed.LobFileInfo{
{Path: "/lob/seg2_file1.vortex", FieldID: 101, TotalRows: 800, ValidRows: 800},
})
ctx.SetSegmentRowStats(2, 800, 400)
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 3)
// Build map for easier assertions
fileMap := make(map[string]packed.LobFileInfo)
for _, f := range files {
fileMap[f.Path] = f
}
// Segment 1 files: new_valid_rows = valid_rows * 0.8
assert.Equal(t, int64(400), fileMap["/lob/seg1_file1.vortex"].ValidRows) // 500 * 0.8
assert.Equal(t, int64(400), fileMap["/lob/seg1_file2.vortex"].ValidRows) // 500 * 0.8
// Segment 2 file: new_valid_rows = valid_rows * 0.5
assert.Equal(t, int64(400), fileMap["/lob/seg2_file1.vortex"].ValidRows) // 800 * 0.5
}
func TestLOBCompactionContext_GetAllMergedLobFilesWithUpdatedStats_NoStats(t *testing.T) {
ctx := NewLOBCompactionContext()
// Add files without setting row stats
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 900},
})
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
// Should return original values when no stats
assert.Len(t, files, 1)
assert.Equal(t, int64(900), files[0].ValidRows)
}
func TestLOBCompactionContext_GetAllMergedLobFilesWithUpdatedStats_AllDeleted(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 1000},
})
// All rows deleted -> survival ratio = 0
ctx.SetSegmentRowStats(1, 1000, 1000)
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 1)
assert.Equal(t, int64(0), files[0].ValidRows)
}
// Tests for GetReuseAllLobFilesWithUpdatedStats (mixed strategy)
func TestLOBCompactionContext_GetReuseAllLobFilesWithUpdatedStats_MixedStrategy(t *testing.T) {
ctx := NewLOBCompactionContext()
// Field 101: high valid ratio -> REUSE_ALL
// Field 102: low valid ratio -> REWRITE_ALL
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101_file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
{Path: "/lob/field102_file1.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 200},
})
ctx.SetSegmentRowStats(1, 2000, 400) // 20% deleted -> survival ratio = 0.8
ctx.ComputeStrategies([]int64{101, 102}, DefaultLOBHoleRatioThreshold)
// Verify mixed strategies
assert.True(t, ctx.IsReuseAll(101))
assert.False(t, ctx.IsReuseAll(102))
assert.True(t, ctx.ShouldRewriteAnyField())
assert.True(t, ctx.HasReuseAllFields())
// Get REUSE_ALL files only
files := ctx.GetReuseAllLobFilesWithUpdatedStats()
// Should only contain field 101 files
assert.Len(t, files, 1)
assert.Equal(t, int64(101), files[0].FieldID)
assert.Equal(t, "/lob/field101_file1.vortex", files[0].Path)
// valid_rows = old_valid_rows * survival_ratio = 950 * 0.8 = 760
assert.Equal(t, int64(760), files[0].ValidRows)
}
func TestLOBCompactionContext_GetReuseAllLobFilesWithUpdatedStats_AllRewrite(t *testing.T) {
ctx := NewLOBCompactionContext()
// All fields have low valid ratio -> all REWRITE_ALL
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101_file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 200},
{Path: "/lob/field102_file1.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 300},
})
ctx.ComputeStrategies([]int64{101, 102}, DefaultLOBHoleRatioThreshold)
assert.False(t, ctx.HasReuseAllFields())
// Should return nil when no REUSE_ALL fields
files := ctx.GetReuseAllLobFilesWithUpdatedStats()
assert.Nil(t, files)
}
func TestLOBCompactionContext_GetReuseAllLobFilesWithUpdatedStats_NilContext(t *testing.T) {
var ctx *LOBCompactionContext = nil
files := ctx.GetReuseAllLobFilesWithUpdatedStats()
assert.Nil(t, files)
}
// Tests for GetRewriteAllFieldIDs and GetReuseAllFieldIDs
func TestLOBCompactionContext_GetFieldIDsByStrategy(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950}, // REUSE_ALL
{Path: "/lob/field102.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 200}, // REWRITE_ALL
{Path: "/lob/field103.vortex", FieldID: 103, TotalRows: 1000, ValidRows: 900}, // REUSE_ALL
})
ctx.ComputeStrategies([]int64{101, 102, 103}, DefaultLOBHoleRatioThreshold)
rewriteIDs := ctx.GetRewriteAllFieldIDs()
reuseIDs := ctx.GetReuseAllFieldIDs()
assert.Len(t, rewriteIDs, 1)
assert.Contains(t, rewriteIDs, int64(102))
assert.Len(t, reuseIDs, 2)
assert.Contains(t, reuseIDs, int64(101))
assert.Contains(t, reuseIDs, int64(103))
}
func TestLOBCompactionContext_GetFieldIDsByStrategy_NilContext(t *testing.T) {
var ctx *LOBCompactionContext = nil
assert.Nil(t, ctx.GetRewriteAllFieldIDs())
assert.Nil(t, ctx.GetReuseAllFieldIDs())
}
// Tests for HasReuseAllFields
func TestLOBCompactionContext_HasReuseAllFields(t *testing.T) {
ctx := NewLOBCompactionContext()
// Before computing strategies
assert.False(t, ctx.HasReuseAllFields())
// Add REUSE_ALL field
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
})
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
assert.True(t, ctx.HasReuseAllFields())
}
// Tests for GetTextColumnConfigs
func TestLOBCompactionContext_GetTextColumnConfigs(t *testing.T) {
ctx := NewLOBCompactionContext()
// Field 101: REUSE_ALL (should not be in configs)
// Field 102: REWRITE_ALL (should be in configs)
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
{Path: "/lob/field102.vortex", FieldID: 102, TotalRows: 1000, ValidRows: 200},
})
ctx.ComputeStrategies([]int64{101, 102}, DefaultLOBHoleRatioThreshold)
configs := ctx.GetTextColumnConfigs("/lob/base/path", DefaultTextInlineThreshold, DefaultTextMaxLobFileBytes, DefaultTextFlushThresholdBytes)
// Should only contain REWRITE_ALL field
assert.Len(t, configs, 1)
assert.Equal(t, int64(102), configs[0].FieldID)
assert.Equal(t, "/lob/base/path/lobs/102", configs[0].LobBasePath)
assert.Equal(t, int64(DefaultTextInlineThreshold), configs[0].InlineThreshold)
assert.Equal(t, int64(DefaultTextMaxLobFileBytes), configs[0].MaxLobFileBytes)
assert.Equal(t, int64(DefaultTextFlushThresholdBytes), configs[0].FlushThresholdBytes)
}
func TestLOBCompactionContext_GetTextColumnConfigs_NoRewriteFields(t *testing.T) {
ctx := NewLOBCompactionContext()
// All fields are REUSE_ALL
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/field101.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 950},
})
ctx.ComputeStrategies([]int64{101}, DefaultLOBHoleRatioThreshold)
configs := ctx.GetTextColumnConfigs("/lob/base/path", DefaultTextInlineThreshold, DefaultTextMaxLobFileBytes, DefaultTextFlushThresholdBytes)
assert.Nil(t, configs)
}
func TestLOBCompactionContext_GetTextColumnConfigs_NilContext(t *testing.T) {
var ctx *LOBCompactionContext = nil
configs := ctx.GetTextColumnConfigs("/lob/base/path", DefaultTextInlineThreshold, DefaultTextMaxLobFileBytes, DefaultTextFlushThresholdBytes)
assert.Nil(t, configs)
}
// Tests for edge cases with proportional valid_rows calculation
func TestLOBCompactionContext_ProportionalValidRows_MultipleFilesPerSegment(t *testing.T) {
ctx := NewLOBCompactionContext()
// Single segment with multiple LOB files (simulating large data scenario)
// Each file has different total_rows
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 1000},
{Path: "/lob/file2.vortex", FieldID: 101, TotalRows: 500, ValidRows: 500},
{Path: "/lob/file3.vortex", FieldID: 101, TotalRows: 300, ValidRows: 300},
})
// 10% of rows deleted from segment
ctx.SetSegmentRowStats(1, 1800, 180) // survival ratio = 0.9
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 3)
// Each file's valid_rows should be updated proportionally
fileMap := make(map[string]packed.LobFileInfo)
for _, f := range files {
fileMap[f.Path] = f
}
// valid_rows = old_valid_rows * 0.9
assert.Equal(t, int64(900), fileMap["/lob/file1.vortex"].ValidRows) // 1000 * 0.9
assert.Equal(t, int64(450), fileMap["/lob/file2.vortex"].ValidRows) // 500 * 0.9
assert.Equal(t, int64(270), fileMap["/lob/file3.vortex"].ValidRows) // 300 * 0.9
}
func TestLOBCompactionContext_ProportionalValidRows_ZeroTotalRows(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 1000},
})
// Edge case: total rows is 0 (should not happen but handle gracefully)
ctx.SetSegmentRowStats(1, 0, 0)
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 1)
// With 0 total rows, survival ratio defaults to 1.0
assert.Equal(t, int64(1000), files[0].ValidRows)
}
func TestLOBCompactionContext_ProportionalValidRows_MoreDeletedThanTotal(t *testing.T) {
ctx := NewLOBCompactionContext()
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 1000},
})
// Edge case: deleted > total (data inconsistency, handle gracefully)
ctx.SetSegmentRowStats(1, 1000, 1500)
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 1)
// Survival ratio clamped to 0
assert.Equal(t, int64(0), files[0].ValidRows)
}
// Tests multi-round REUSE_ALL compaction where ValidRows < TotalRows from previous rounds.
// After a previous REUSE_ALL compaction, LOB files have ValidRows < TotalRows because
// the physical file is unchanged but some rows are logically deleted.
// The new valid_rows must be based on the previous ValidRows, not TotalRows.
func TestLOBCompactionContext_MultiRoundReuseAll(t *testing.T) {
ctx := NewLOBCompactionContext()
// Simulate state after a previous REUSE_ALL compaction:
// LOB file still has 1000 physical rows, but only 800 are valid
ctx.AddSegmentLobFiles(1, []packed.LobFileInfo{
{Path: "/lob/file1.vortex", FieldID: 101, TotalRows: 1000, ValidRows: 800},
{Path: "/lob/file2.vortex", FieldID: 101, TotalRows: 500, ValidRows: 300},
})
// In this round, binlog has 1100 rows (800+300 valid from previous round).
// 110 rows are deleted (10% survival loss).
ctx.SetSegmentRowStats(1, 1100, 110) // survival ratio = 0.9
files := ctx.GetAllMergedLobFilesWithUpdatedStats()
assert.Len(t, files, 2)
fileMap := make(map[string]packed.LobFileInfo)
for _, f := range files {
fileMap[f.Path] = f
}
// new_valid = old_valid * survival_ratio
// file1: 800 * 0.9 = 720 (NOT 1000 * 0.9 = 900)
assert.Equal(t, int64(720), fileMap["/lob/file1.vortex"].ValidRows)
// file2: 300 * 0.9 = 270 (NOT 500 * 0.9 = 450)
assert.Equal(t, int64(270), fileMap["/lob/file2.vortex"].ValidRows)
// TotalRows should remain unchanged (physical file not rewritten)
assert.Equal(t, int64(1000), fileMap["/lob/file1.vortex"].TotalRows)
assert.Equal(t, int64(500), fileMap["/lob/file2.vortex"].TotalRows)
}