// 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 datacoord

import (
	"context"
	"testing"

	"github.com/cockroachdb/errors"
	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"

	"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/internal/datacoord/allocator"
	"github.com/milvus-io/milvus/internal/datacoord/session"
	"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
	"github.com/milvus-io/milvus/pkg/v3/util/merr"
	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
)

func TestL0CompactionTaskSuite(t *testing.T) {
	suite.Run(t, new(L0CompactionTaskSuite))
}

type L0CompactionTaskSuite struct {
	suite.Suite

	mockAlloc *allocator.MockAllocator
	mockMeta  *MockCompactionMeta
}

func (s *L0CompactionTaskSuite) SetupTest() {
	s.mockMeta = NewMockCompactionMeta(s.T())
	s.mockAlloc = allocator.NewMockAllocator(s.T())
	// s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything).Return(nil)
}

func (s *L0CompactionTaskSuite) SetupSubTest() {
	s.SetupTest()
}

func (s *L0CompactionTaskSuite) TestSaveSegmentMetaUsesAtomicDeltalogOperator() {
	actualDeltaPath := "/tmp/milvus/insert_log/1/10/200/_delta/not-log-id-suffix"

	task := s.generateTestL0Task(datapb.CompactionTaskState_executing)
	output := []*datapb.CompactionSegment{{
		SegmentID: 200,
		Deltalogs: []*datapb.FieldBinlog{{
			Binlogs: []*datapb.Binlog{{LogID: 9001, LogPath: actualDeltaPath, EntriesNum: 3}},
		}},
	}}

	s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
		func(ctx context.Context, operators ...UpdateOperator) error {
			s.Len(operators, 5)
			s.Equal(actualDeltaPath, output[0].GetDeltalogs()[0].GetBinlogs()[0].GetLogPath())
			s.EqualValues(9001, output[0].GetDeltalogs()[0].GetBinlogs()[0].GetLogID())
			return nil
		},
	).Once()

	s.NoError(task.saveSegmentMeta(output))
}

func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_NormalL0() {
	channel := "Ch-1"
	deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}

	s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
		[]*SegmentInfo{
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            200,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            201,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            202,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
		},
	)

	s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
		return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
			ID:            segID,
			Level:         datapb.SegmentLevel_L0,
			InsertChannel: channel,
			State:         commonpb.SegmentState_Flushed,
			Deltalogs:     deltaLogs,
		}}
	}).Times(2)
	task := newL0CompactionTask(&datapb.CompactionTask{
		PlanID:        1,
		TriggerID:     19530,
		CollectionID:  1,
		PartitionID:   10,
		Type:          datapb.CompactionType_Level0DeleteCompaction,
		NodeID:        1,
		State:         datapb.CompactionTaskState_executing,
		InputSegments: []int64{100, 101},
	}, nil, s.mockMeta)
	alloc := allocator.NewMockAllocator(s.T())
	alloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
	task.allocator = alloc
	plan, err := task.BuildCompactionRequest()
	s.Require().NoError(err)

	s.Equal(5, len(plan.GetSegmentBinlogs()))
	segIDs := lo.Map(plan.GetSegmentBinlogs(), func(b *datapb.CompactionSegmentBinlogs, _ int) int64 {
		return b.GetSegmentID()
	})

	s.ElementsMatch([]int64{200, 201, 202, 100, 101}, segIDs)
}

func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_SegmentNotFoundL0() {
	channel := "Ch-1"
	s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
		return nil
	}).Once()
	task := newL0CompactionTask(&datapb.CompactionTask{
		InputSegments: []int64{102},
		PlanID:        1,
		TriggerID:     19530,
		CollectionID:  1,
		PartitionID:   10,
		Channel:       channel,
		Type:          datapb.CompactionType_Level0DeleteCompaction,
		NodeID:        1,
		State:         datapb.CompactionTaskState_executing,
	}, nil, s.mockMeta)

	_, err := task.BuildCompactionRequest()
	s.Error(err)
	s.ErrorIs(err, merr.ErrSegmentNotFound)
}

func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_SelectZeroSegmentsL0() {
	channel := "Ch-1"
	deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
	s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
		return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
			ID:            segID,
			Level:         datapb.SegmentLevel_L0,
			InsertChannel: channel,
			State:         commonpb.SegmentState_Flushed,
			Deltalogs:     deltaLogs,
		}}
	}).Times(2)
	s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()

	task := newL0CompactionTask(&datapb.CompactionTask{
		PlanID:        1,
		TriggerID:     19530,
		CollectionID:  1,
		PartitionID:   10,
		Type:          datapb.CompactionType_Level0DeleteCompaction,
		NodeID:        1,
		State:         datapb.CompactionTaskState_executing,
		InputSegments: []int64{100, 101},
	}, nil, s.mockMeta)
	plan, err := task.BuildCompactionRequest()
	// Fast finish: should return a plan with only L0 segments (no error)
	s.NoError(err)
	s.Require().NotNil(plan)
	// Verify plan only contains L0 input segments (2 segments)
	s.Equal(2, len(plan.GetSegmentBinlogs()))
	segIDs := lo.Map(plan.GetSegmentBinlogs(), func(b *datapb.CompactionSegmentBinlogs, _ int) int64 {
		return b.GetSegmentID()
	})
	s.ElementsMatch([]int64{100, 101}, segIDs)
	// Verify no binlog IDs were allocated for fast finish
	s.Nil(plan.GetPreAllocatedLogIDs())
}

func (s *L0CompactionTaskSuite) TestBuildCompactionRequestFailed_AllocFailed() {
	channel := "Ch-1"
	deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}

	s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
		[]*SegmentInfo{
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            200,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            201,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
			{SegmentInfo: &datapb.SegmentInfo{
				ID:            202,
				Level:         datapb.SegmentLevel_L1,
				InsertChannel: channel,
			}},
		},
	)

	s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
		return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
			ID:            segID,
			Level:         datapb.SegmentLevel_L0,
			InsertChannel: channel,
			State:         commonpb.SegmentState_Flushed,
			Deltalogs:     deltaLogs,
		}}
	}).Times(2)
	task := newL0CompactionTask(&datapb.CompactionTask{
		PlanID:        1,
		TriggerID:     19530,
		CollectionID:  1,
		PartitionID:   10,
		Type:          datapb.CompactionType_Level0DeleteCompaction,
		NodeID:        1,
		State:         datapb.CompactionTaskState_executing,
		InputSegments: []int64{100, 101},
	}, s.mockAlloc, s.mockMeta)

	s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(0, 0, errors.New("mock alloc err"))

	_, err := task.BuildCompactionRequest()
	s.T().Logf("err=%v", err)
	s.Error(err)
}

func (s *L0CompactionTaskSuite) generateTestL0Task(state datapb.CompactionTaskState) *l0CompactionTask {
	return newL0CompactionTask(&datapb.CompactionTask{
		PlanID:        1,
		TriggerID:     19530,
		CollectionID:  1,
		PartitionID:   10,
		Type:          datapb.CompactionType_Level0DeleteCompaction,
		NodeID:        NullNodeID,
		State:         state,
		Channel:       "ch-1",
		InputSegments: []int64{100, 101},
	}, s.mockAlloc, s.mockMeta)
}

func (s *L0CompactionTaskSuite) TestPorcessStateTrans() {
	s.Run("test pipelining Compaction failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_pipelining)
		t.updateAndSaveTaskMeta(setNodeID(100))
		channel := "ch-1"
		deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}

		s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
			[]*SegmentInfo{
				{SegmentInfo: &datapb.SegmentInfo{
					ID:            200,
					Level:         datapb.SegmentLevel_L1,
					InsertChannel: channel,
				}},
			},
		)

		s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
			return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
				ID:            segID,
				Level:         datapb.SegmentLevel_L0,
				InsertChannel: channel,
				State:         commonpb.SegmentState_Flushed,
				Deltalogs:     deltaLogs,
			}}
		}).Twice()
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().CreateCompaction(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(nodeID int64, plan *datapb.CompactionPlan, collectionID int64) error {
			s.Require().EqualValues(t.GetTaskProto().NodeID, nodeID)
			s.Require().EqualValues(t.GetTaskProto().GetCollectionID(), collectionID)
			return errors.New("mock error")
		})

		t.CreateTaskOnWorker(100, cluster)
		s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().State)
		s.EqualValues(NullNodeID, t.GetTaskProto().NodeID)
	})

	s.Run("test pipelining success", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_pipelining)
		t.updateAndSaveTaskMeta(setNodeID(100))
		channel := "ch-1"
		deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}

		s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
			[]*SegmentInfo{
				{SegmentInfo: &datapb.SegmentInfo{
					ID:            200,
					Level:         datapb.SegmentLevel_L1,
					InsertChannel: channel,
				}},
			},
		)

		s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
			return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
				ID:            segID,
				Level:         datapb.SegmentLevel_L0,
				InsertChannel: channel,
				State:         commonpb.SegmentState_Flushed,
				Deltalogs:     deltaLogs,
			}}
		}).Twice()

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().CreateCompaction(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(nodeID int64, plan *datapb.CompactionPlan, collectionID int64) error {
			s.Require().EqualValues(t.GetTaskProto().NodeID, nodeID)
			s.Require().EqualValues(t.GetTaskProto().GetCollectionID(), collectionID)
			return nil
		})

		t.CreateTaskOnWorker(100, cluster)
		s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
	})

	// stay in executing state when GetCompactionPlanResults error except ErrNodeNotFound
	s.Run("test executing GetCompactionPlanResult fail NodeNotFound", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).Return(nil, merr.WrapErrNodeNotFound(t.GetTaskProto().NodeID)).Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().GetState())
		s.EqualValues(NullNodeID, t.GetTaskProto().GetNodeID())
	})

	// stay in executing state when GetCompactionPlanResults error except ErrNodeNotFound
	s.Run("test executing GetCompactionPlanResult fail mock error", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).Return(nil, errors.New("mock error"))
		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().GetState())
		s.EqualValues(-1, t.GetTaskProto().GetNodeID())
	})

	s.Run("test executing with result executing", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)
		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_executing,
			}, nil).Once()

		t.QueryTaskOnWorker(cluster)
	})

	s.Run("test executing with result completed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_completed,
			}, nil).Once()

		s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
		s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Times(2)
		s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).Return().Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
	})
	s.Run("test executing with result completed save segment meta failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_completed,
			}, nil).Once()

		s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
		s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
			Return(errors.New("mock error")).Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
	})
	s.Run("test executing with result completed save compaction meta failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_completed,
			}, nil).Once()

		s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
		s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
	})

	s.Run("test executing with result failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_failed,
			}, nil).Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
	})
	s.Run("test executing with result failed save compaction meta failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
		t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		cluster := session.NewMockCluster(s.T())
		cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
			Return(&datapb.CompactionPlanResult{
				PlanID: t.GetTaskProto().GetPlanID(),
				State:  datapb.CompactionTaskState_failed,
			}, nil).Once()
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()

		t.QueryTaskOnWorker(cluster)
		s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
	})

	s.Run("test metaSaved success", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_meta_saved)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
			s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
		}).Once()

		got := t.Process()
		s.True(got)
		s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
	})

	s.Run("test metaSaved failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
		t := s.generateTestL0Task(datapb.CompactionTaskState_meta_saved)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()

		got := t.Process()
		s.False(got)
		s.Equal(datapb.CompactionTaskState_meta_saved, t.GetTaskProto().GetState())
	})

	s.Run("test complete drop failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_completed)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)
		s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
			s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
		}).Once()

		got := t.Process()
		s.True(got)
		s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
	})

	s.Run("test complete success", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_completed)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)
		s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
			s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
		}).Once()

		got := t.Process()
		s.True(got)
		s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
	})

	s.Run("test process failed success", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_failed)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		got := t.Process()
		s.True(got)
		s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
	})

	s.Run("test process failed failed", func() {
		s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
		t := s.generateTestL0Task(datapb.CompactionTaskState_failed)
		t.updateAndSaveTaskMeta(setNodeID(100))
		s.Require().True(t.GetTaskProto().GetNodeID() > 0)

		got := t.Process()
		s.True(got)
		s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
	})

	s.Run("test process states", func() {
		testCases := []struct {
			state         datapb.CompactionTaskState
			processResult bool
		}{
			{state: datapb.CompactionTaskState_unknown, processResult: false},
			{state: datapb.CompactionTaskState_pipelining, processResult: false},
			{state: datapb.CompactionTaskState_executing, processResult: false},
			{state: datapb.CompactionTaskState_failed, processResult: true},
			{state: datapb.CompactionTaskState_timeout, processResult: true},
		}

		for _, tc := range testCases {
			t := s.generateTestL0Task(tc.state)
			res := t.Process()
			s.Equal(tc.processResult, res)
		}
	})
}

// TestSelectFlushedSegment_ForceSelectAllFlag exercises the flag end-to-end:
// build an L0 view, call Trigger() with the flag off / on, feed the resulting
// latestDeletePos into an l0CompactionTask (the same wiring
// compaction_trigger_v2.go uses), and verify selectFlushedSegment's output
// actually changes based on the flag. A high-StartPosition segment (the
// shape silently dropped by the import-position bug) is excluded with the
// flag off and included with the flag on.
func (s *L0CompactionTaskSuite) TestSelectFlushedSegment_ForceSelectAllFlag() {
	paramtable.Init()
	const flagKey = "dataCoord.compaction.levelzero.forceSelectAllSegments"

	channel := "ch-1"
	collectionID := int64(1)
	partitionID := int64(10)
	label := &CompactionGroupLabel{
		CollectionID: collectionID,
		PartitionID:  partitionID,
		Channel:      channel,
	}

	// Flushed L1 segments visible to selectFlushedSegment. The one with
	// StartPosition > realL0DmlTs is the case we care about — it would
	// normally be filtered out by `startPos < taskPos`.
	const realL0DmlTs = uint64(5000)
	lowPosSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
		ID:            200,
		CollectionID:  collectionID,
		PartitionID:   partitionID,
		InsertChannel: channel,
		Level:         datapb.SegmentLevel_L1,
		State:         commonpb.SegmentState_Flushed,
		StartPosition: &msgpb.MsgPosition{ChannelName: channel, Timestamp: 3000},
	}}
	highPosSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
		ID:            201,
		CollectionID:  collectionID,
		PartitionID:   partitionID,
		InsertChannel: channel,
		Level:         datapb.SegmentLevel_L1,
		State:         commonpb.SegmentState_Flushed,
		StartPosition: &msgpb.MsgPosition{ChannelName: channel, Timestamp: 20000},
	}}

	// Mockery's SelectSegments returns whatever we tell it without applying
	// filters. We need the real filter logic to run so the `startPos <
	// taskPos` predicate is actually exercised — install RunAndReturn that
	// evaluates each SegmentFilter.Match against our fixed segment set.
	installFilteringMock := func() {
		s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).
			RunAndReturn(func(ctx context.Context, filters ...SegmentFilter) []*SegmentInfo {
				all := []*SegmentInfo{lowPosSeg, highPosSeg}
				result := make([]*SegmentInfo, 0, len(all))
				for _, seg := range all {
					matched := true
					for _, f := range filters {
						if !f.Match(seg) {
							matched = false
							break
						}
					}
					if matched {
						result = append(result, seg)
					}
				}
				return result
			})
	}

	// Build an L0 view with a couple of L0 segments whose dmlPos is
	// `realL0DmlTs`. Trigger() runs resolveLatestDeletePos under the current
	// flag value, so the returned view's latestDeletePos reflects the flag.
	buildView := func() *LevelZeroCompactionView {
		l0Segs := []*SegmentView{
			{
				ID:            100,
				label:         label,
				dmlPos:        &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
				Level:         datapb.SegmentLevel_L0,
				State:         commonpb.SegmentState_Flushed,
				DeltalogCount: 100,
				DeltaSize:     1,
				DeltaRowCount: 1,
			},
			{
				ID:            101,
				label:         label,
				dmlPos:        &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
				Level:         datapb.SegmentLevel_L0,
				State:         commonpb.SegmentState_Flushed,
				DeltalogCount: 100,
				DeltaSize:     1,
				DeltaRowCount: 1,
			},
		}
		return &LevelZeroCompactionView{
			label:           label,
			l0Segments:      l0Segs,
			latestDeletePos: &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
			triggerID:       19530,
		}
	}

	// Mirrors compaction_trigger_v2.go:406 — feed the triggered view's
	// latestDeletePos into task.Pos and run selectFlushedSegment.
	runSelectWithTriggeredPos := func() []int64 {
		installFilteringMock()
		srcView := buildView()
		triggered, reason := srcView.Trigger()
		s.Require().NotNil(triggered, "Trigger returned nil: %s", reason)
		triggeredView := triggered.(*LevelZeroCompactionView)

		task := newL0CompactionTask(&datapb.CompactionTask{
			PlanID:        1,
			TriggerID:     19530,
			CollectionID:  collectionID,
			PartitionID:   partitionID,
			Channel:       channel,
			Type:          datapb.CompactionType_Level0DeleteCompaction,
			NodeID:        1,
			State:         datapb.CompactionTaskState_executing,
			InputSegments: []int64{100, 101},
			Pos:           triggeredView.latestDeletePos,
		}, nil, s.mockMeta)

		flushed, _, err := task.selectFlushedSegment()
		s.Require().NoError(err)
		return lo.Map(flushed, func(seg *SegmentInfo, _ int) int64 { return seg.GetID() })
	}

	s.Run("flag_off_excludes_high_start_position_segment", func() {
		paramtable.Get().Save(flagKey, "false")
		defer paramtable.Get().Reset(flagKey)

		gotIDs := runSelectWithTriggeredPos()
		s.ElementsMatch([]int64{200}, gotIDs,
			"with flag off, segment 201 (StartPosition=20000 > taskPos=%d) must be excluded", realL0DmlTs)
	})

	s.Run("flag_on_includes_high_start_position_segment", func() {
		paramtable.Get().Save(flagKey, "true")
		defer paramtable.Get().Reset(flagKey)

		gotIDs := runSelectWithTriggeredPos()
		s.ElementsMatch([]int64{200, 201}, gotIDs,
			"with flag on, resolveLatestDeletePos must lift taskPos so segment 201 is included")
	})
}

// TestSelectFlushedSegment_RespectsCommitTimestamp verifies that import segments
// with a commit_timestamp are excluded from L0 compaction when the trigger
// position is before the commit_timestamp.
func TestSelectFlushedSegment_RespectsCommitTimestamp(t *testing.T) {
	channel := "ch-1"

	// Import segment: start_position.ts=1000, commit_ts=5000.
	// Its effective timestamp is 5000 (controlled by segmentEffectiveTs).
	importSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
		ID:              777,
		CollectionID:    1,
		PartitionID:     10,
		InsertChannel:   channel,
		State:           commonpb.SegmentState_Flushed,
		Level:           datapb.SegmentLevel_L1,
		CommitTimestamp: 5000,
		StartPosition:   &msgpb.MsgPosition{Timestamp: 1000},
	}}

	// applyFilters applies a SegmentFilter slice to a candidate list,
	// mirroring what meta.SelectSegments does internally.
	applyFilters := func(candidates []*SegmentInfo, filters ...SegmentFilter) []*SegmentInfo {
		var result []*SegmentInfo
		for _, seg := range candidates {
			pass := true
			for _, f := range filters {
				if !f.Match(seg) {
					pass = false
					break
				}
			}
			if pass {
				result = append(result, seg)
			}
		}
		return result
	}

	makeTask := func(triggerTs uint64) *l0CompactionTask {
		mockAlloc := allocator.NewMockAllocator(t)
		mockMeta := NewMockCompactionMeta(t)
		mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).
			RunAndReturn(func(ctx context.Context, filters ...SegmentFilter) []*SegmentInfo {
				return applyFilters([]*SegmentInfo{importSeg}, filters...)
			})
		return newL0CompactionTask(&datapb.CompactionTask{
			PlanID:       1,
			TriggerID:    19530,
			CollectionID: 1,
			PartitionID:  10,
			Type:         datapb.CompactionType_Level0DeleteCompaction,
			Channel:      channel,
			Pos:          &msgpb.MsgPosition{Timestamp: triggerTs},
		}, mockAlloc, mockMeta)
	}

	t.Run("import segment not selected when trigger pos < commit_timestamp", func(t *testing.T) {
		// triggerTs=3000 < commit_ts=5000 → segment must NOT be selected
		task := makeTask(3000)
		selected, _, err := task.selectFlushedSegment()
		assert.NoError(t, err)
		assert.Empty(t, selected, "import segment with commit_ts=5000 must not be selected at triggerTs=3000")
	})

	t.Run("import segment selected when trigger pos > commit_timestamp", func(t *testing.T) {
		// triggerTs=6000 > commit_ts=5000 → segment must be selected
		task := makeTask(6000)
		selected, _, err := task.selectFlushedSegment()
		assert.NoError(t, err)
		assert.Len(t, selected, 1, "import segment with commit_ts=5000 must be selected at triggerTs=6000")
		assert.Equal(t, int64(777), selected[0].GetID())
	})
}
