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
237 lines
9.4 KiB
Go
237 lines
9.4 KiB
Go
package datacoord
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/msgpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/tsoutil"
|
|
)
|
|
|
|
// ── helpers ──────────────────────────────────────────────────────────────
|
|
|
|
func TestSegmentEffectiveTs_NormalSegment(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
}
|
|
assert.Equal(t, uint64(1000), segmentEffectiveTs(seg))
|
|
}
|
|
|
|
func TestSegmentEffectiveTs_ImportSegment(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
CommitTimestamp: 5000,
|
|
}
|
|
assert.Equal(t, uint64(5000), segmentEffectiveTs(seg))
|
|
}
|
|
|
|
func TestSegmentEffectiveTs_NilStartPosition(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{CommitTimestamp: 5000}
|
|
assert.Equal(t, uint64(5000), segmentEffectiveTs(seg))
|
|
}
|
|
|
|
func TestSegmentEffectiveTs_BothZero(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{}
|
|
assert.Equal(t, uint64(0), segmentEffectiveTs(seg))
|
|
}
|
|
|
|
func TestSegmentEffectiveDmlTs_NormalSegment(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
DmlPosition: &msgpb.MsgPosition{Timestamp: 2000},
|
|
}
|
|
assert.Equal(t, uint64(2000), segmentEffectiveDmlTs(seg))
|
|
}
|
|
|
|
func TestSegmentEffectiveDmlTs_ImportSegment(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
DmlPosition: &msgpb.MsgPosition{Timestamp: 2000},
|
|
CommitTimestamp: 5000,
|
|
}
|
|
assert.Equal(t, uint64(5000), segmentEffectiveDmlTs(seg))
|
|
}
|
|
|
|
// ── UpdateCommitTimestamp operator ──────────────────────────────────────
|
|
|
|
func TestUpdateCommitTimestamp_SetsField(t *testing.T) {
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
err = meta.AddSegment(context.Background(), &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{ID: 1, State: commonpb.SegmentState_Flushed}})
|
|
assert.NoError(t, err)
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 9999))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(9999), meta.GetSegment(context.Background(), 1).GetCommitTimestamp())
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_SegmentNotFound(t *testing.T) {
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(999, 9999))
|
|
assert.NoError(t, err)
|
|
}
|
|
|
|
// ── UpdateCommitTimestamp input validation ──────────────────────────────
|
|
// Invariant: commit_ts (when non-zero) must be >= max(binlog.TimestampTo).
|
|
// Reject violations at the entry point so C++ segcore never sees an invalid
|
|
// commit_ts that would silently lower row timestamps during load-time overwrite.
|
|
|
|
// segWithBinlogs builds a Flushed SegmentInfo carrying a single binlog whose
|
|
// TimestampTo == maxTsTo, used to exercise the max(binlog.TimestampTo) check.
|
|
func segWithBinlogs(id int64, maxTsTo uint64) *SegmentInfo {
|
|
return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: id,
|
|
State: commonpb.SegmentState_Flushed,
|
|
Binlogs: []*datapb.FieldBinlog{{
|
|
FieldID: 100,
|
|
Binlogs: []*datapb.Binlog{{LogID: 1, TimestampFrom: 100, TimestampTo: maxTsTo}},
|
|
}},
|
|
}}
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_AcceptAboveMaxTimestampTo(t *testing.T) {
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
assert.NoError(t, meta.AddSegment(context.Background(), segWithBinlogs(1, 5000)))
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 9999))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(9999), meta.GetSegment(context.Background(), 1).GetCommitTimestamp())
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_AcceptEqualToMaxTimestampTo(t *testing.T) {
|
|
// Boundary: commit_ts == max(binlog.TimestampTo) is valid.
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
assert.NoError(t, meta.AddSegment(context.Background(), segWithBinlogs(1, 5000)))
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 5000))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(5000), meta.GetSegment(context.Background(), 1).GetCommitTimestamp())
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_RejectBelowMaxTimestampTo(t *testing.T) {
|
|
// Violation: commit_ts < max(binlog.TimestampTo). The operator must leave
|
|
// commit_timestamp unchanged so downstream code never observes a
|
|
// commit_ts that would lower row timestamps.
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
assert.NoError(t, meta.AddSegment(context.Background(), segWithBinlogs(1, 5000)))
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 3000))
|
|
// UpdateSegmentsInfo returns nil when the operator returns false (no-op);
|
|
// the assertion is on the persisted state, not the error.
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(0), meta.GetSegment(context.Background(), 1).GetCommitTimestamp(),
|
|
"rejected update must leave commit_timestamp unchanged")
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_AcceptZeroReset(t *testing.T) {
|
|
// The ts == 0 reset path is used by compaction completion and must always
|
|
// pass, regardless of the binlog TimestampTo values on the segment.
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
seg := segWithBinlogs(1, 5000)
|
|
seg.CommitTimestamp = 9999
|
|
assert.NoError(t, meta.AddSegment(context.Background(), seg))
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 0))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(0), meta.GetSegment(context.Background(), 1).GetCommitTimestamp())
|
|
}
|
|
|
|
func TestUpdateCommitTimestamp_AcceptWhenNoBinlogs(t *testing.T) {
|
|
// Pre-binlog state (e.g. just before first flush): max(binlog.TimestampTo)
|
|
// is 0, so any commit_ts passes the >= check.
|
|
meta, err := newMemoryMeta(t)
|
|
assert.NoError(t, err)
|
|
assert.NoError(t, meta.AddSegment(context.Background(), &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 1, State: commonpb.SegmentState_Flushed,
|
|
}}))
|
|
|
|
err = meta.UpdateSegmentsInfo(context.Background(), UpdateCommitTimestamp(1, 7777))
|
|
assert.NoError(t, err)
|
|
assert.Equal(t, uint64(7777), meta.GetSegment(context.Background(), 1).GetCommitTimestamp())
|
|
}
|
|
|
|
// ── EffectiveTimestamp / helper boundary values ─────────────────────────
|
|
// Documents the expected behavior at the commit_ts == raw_ts boundary.
|
|
|
|
func TestEffectiveTimestamp_CommitEqualsRawTs(t *testing.T) {
|
|
// At the boundary, either value is correct; the helper returns rawTs
|
|
// because the `commitTs > rawTs` branch does not trigger.
|
|
assert.Equal(t, uint64(5000), tsoutil.EffectiveTimestamp(5000, 5000))
|
|
}
|
|
|
|
func TestSegmentEffectiveTs_CommitEqualsStartPosition(t *testing.T) {
|
|
// Boundary: commit_ts == start_position.Timestamp. Helper returns
|
|
// commit_ts (non-zero branch wins) — behavior-wise identical to rawTs.
|
|
seg := &datapb.SegmentInfo{
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 5000},
|
|
CommitTimestamp: 5000,
|
|
}
|
|
assert.Equal(t, uint64(5000), segmentEffectiveTs(seg))
|
|
}
|
|
|
|
// ── GenSnapshot (handler.go) ─────────────────────────────────────────────
|
|
|
|
func TestGenSnapshot_ImportSegment_ExcludedBeforeCommitTs(t *testing.T) {
|
|
seg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 1,
|
|
State: commonpb.SegmentState_Flushed,
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
CommitTimestamp: 5000,
|
|
Binlogs: []*datapb.FieldBinlog{{FieldID: 0, Binlogs: []*datapb.Binlog{{LogID: 1}}}},
|
|
IsImporting: false,
|
|
}}
|
|
assert.False(t, segmentEffectiveTs(seg.SegmentInfo) < 3000,
|
|
"import segment with commit_ts=5000 must NOT pass snapshot filter at snapshotTs=3000")
|
|
}
|
|
|
|
func TestGenSnapshot_ImportSegment_IncludedAfterCommitTs(t *testing.T) {
|
|
seg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 1,
|
|
State: commonpb.SegmentState_Flushed,
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
CommitTimestamp: 5000,
|
|
Binlogs: []*datapb.FieldBinlog{{FieldID: 0, Binlogs: []*datapb.Binlog{{LogID: 1}}}},
|
|
}}
|
|
assert.True(t, segmentEffectiveTs(seg.SegmentInfo) < 6000,
|
|
"import segment with commit_ts=5000 MUST pass snapshot filter at snapshotTs=6000")
|
|
}
|
|
|
|
func TestGenSnapshot_NormalSegment_UnchangedBehavior(t *testing.T) {
|
|
seg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 2,
|
|
State: commonpb.SegmentState_Flushed,
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 3000},
|
|
}}
|
|
assert.True(t, segmentEffectiveTs(seg.SegmentInfo) < 4000)
|
|
assert.False(t, segmentEffectiveTs(seg.SegmentInfo) < 2000)
|
|
}
|
|
|
|
// ── GC eligibility (garbage_collector.go) ───────────────────────────────
|
|
|
|
func TestSegmentEffectiveDmlTs_GCProtection(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
DmlPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
CommitTimestamp: 5000,
|
|
}
|
|
cpTs := uint64(3000)
|
|
assert.True(t, segmentEffectiveDmlTs(seg) > cpTs,
|
|
"import segment with commit_ts=5000 must NOT be GC'd at checkpoint=3000")
|
|
}
|
|
|
|
func TestSegmentEffectiveDmlTs_GCAllowed_AfterCommitTs(t *testing.T) {
|
|
seg := &datapb.SegmentInfo{
|
|
DmlPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
CommitTimestamp: 5000,
|
|
}
|
|
cpTs := uint64(6000)
|
|
assert.False(t, segmentEffectiveDmlTs(seg) > cpTs,
|
|
"import segment with commit_ts=5000 CAN be GC'd at checkpoint=6000")
|
|
}
|