enhance: [cp3.0]disable L0 import by default to prevent commit_ts corruption (#51569)

Cherry-pick from master (PR still open)

pr: #51194
issue: #51247

## Summary

Cherry-picked from master PR #51194 (open - preemptive backport)

**Note**: Original PR is still open. This CP PR should be updated if the
original PR changes.

## Verification

- [x] File count matches original PR
- [x] Code changes verified (per-file diff matches master PR)
- [x] No conflict markers
- [ ] make static-check (skipped by cherry-pick workflow)

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

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

configs/milvus.yaml

@@ -829,6 +829,7 @@ dataCoord:
     maxImportJobNum: 1024 # Maximum number of import jobs that are executing or pending.
     waitForIndex: true # Indicates whether the import operation waits for the completion of index building.
     enableInReplicatingCluster: false # Whether to allow import in a replicating cluster. When enabled, only auto_commit=false imports are accepted.
+    enableL0Import: false # Whether to allow importing L0 (delete-only) segments during a binlog/backup import. Disabled by default because restoring L0 segments is incompatible with commit_timestamp (two-phase-commit / replication imports), where it silently breaks delete semantics. Fold the L0 deletes into per-segment deltalogs before restore instead; set to true only to re-enable the legacy L0 import behavior.
     fileNumPerSlot: 4 # The files number per slot for pre-import/import task.
     memoryLimitPerSlot: 160 # The memory limit (in MB) of buffer size per slot for pre-import/import task.
     maxSegmentsPerCopyTask: 100 # Maximum number of segments that can be grouped into a single copy task during snapshot restore.

internal/datacoord/import_services_test.go

@@ -18,6 +18,7 @@ package datacoord
 
 import (
 	"context"
+	"math"
 	"testing"
 
 	"github.com/bytedance/mockey"
@@ -40,6 +41,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
 
 // ================================
@@ -88,6 +90,56 @@ func (s *ImportServicesSuite) TestImportV2_InvalidTimeoutReturnsError() {
 	s.True(errors.Is(merr.Error(resp.GetStatus()), merr.ErrImportFailed))
 }
 
+func (s *ImportServicesSuite) TestImportV2_L0ImportDisabledReturnsError() {
+	paramtable.Init()
+	ctx := context.Background()
+	server := &Server{}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	// enableL0Import defaults to false, so an l0_import request must be rejected
+	// before reaching allocation (allocator is nil here, proving the early reject).
+	req := &internalpb.ImportRequestInternal{
+		Options: []*commonpb.KeyValuePair{
+			{Key: "timeout", Value: "300s"},
+			{Key: "l0_import", Value: "true"},
+		},
+	}
+
+	resp, err := server.ImportV2(ctx, req)
+
+	s.NoError(err)
+	s.NotNil(resp)
+	s.True(errors.Is(merr.Error(resp.GetStatus()), merr.ErrImportFailed))
+	s.Contains(resp.GetStatus().GetReason(), "l0 import is disabled")
+}
+
+func (s *ImportServicesSuite) TestImportV2_L0ImportEnabledPassesGate() {
+	paramtable.Init()
+	ctx := context.Background()
+	server := &Server{}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	params := paramtable.Get()
+	params.Save(params.DataCoordCfg.EnableL0Import.Key, "true")
+	defer params.Reset(params.DataCoordCfg.EnableL0Import.Key)
+
+	// With enableL0Import=true the same request must get past the L0 gate and
+	// fail later on the nil allocator instead — proof the gate was skipped.
+	req := &internalpb.ImportRequestInternal{
+		Options: []*commonpb.KeyValuePair{
+			{Key: "timeout", Value: "300s"},
+			{Key: "l0_import", Value: "true"},
+		},
+	}
+
+	resp, err := server.ImportV2(ctx, req)
+
+	s.NoError(err)
+	s.NotNil(resp)
+	s.True(errors.Is(merr.Error(resp.GetStatus()), merr.ErrServiceUnavailable))
+	s.Contains(resp.GetStatus().GetReason(), "allocator not initialized")
+}
+
 func (s *ImportServicesSuite) TestImportV2_AllocatorNilReturnsError() {
 	ctx := context.Background()
 	server := &Server{}
@@ -702,4 +754,136 @@ func (s *ImportServicesSuite) TestCreateImportJobFromAck_AssignsFileIDs() {
 	s.Equal(int64(1003), files[2].GetId()) // idStart + 2 + 1
 }
 
+func (s *ImportServicesSuite) TestCreateImportJobFromAck_L0ImportDisabledCreatesFailedJob() {
+	paramtable.Init()
+	ctx := context.Background()
+
+	mockHandler := NewNMockHandler(s.T())
+	mockHandler.EXPECT().GetCollection(mock.Anything, mock.Anything).Return(&collectionInfo{
+		ID:            100,
+		VChannelNames: []string{"v1"},
+	}, nil)
+
+	var savedJob *datapb.ImportJob
+	catalog := mocks.NewDataCoordCatalog(s.T())
+	catalog.EXPECT().ListImportJobs(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListPreImportTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, job *datapb.ImportJob) error {
+		savedJob = job
+		return nil
+	})
+
+	importMeta, err := NewImportMeta(context.TODO(), catalog, nil, nil)
+	s.NoError(err)
+
+	server := &Server{
+		handler:    mockHandler,
+		importMeta: importMeta,
+	}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	mockAllocator := allocator.NewMockAllocator(s.T())
+	mockAllocator.EXPECT().AllocN(mock.Anything).Return(int64(1000), int64(1002), nil)
+	server.allocator = mockAllocator
+
+	// enableL0Import defaults to false. A replicated l0_import message reaching
+	// the ack path must NOT create a runnable job, and must NOT return an error
+	// (ack callbacks retry forever); instead the job is created in Failed state
+	// so replicated CommitImport becomes a terminal no-op.
+	req := &internalpb.ImportRequestInternal{
+		CollectionID:   100,
+		CollectionName: "test_collection",
+		PartitionIDs:   []int64{1},
+		ChannelNames:   []string{"v1"},
+		Schema:         &schemapb.CollectionSchema{Name: "test_collection"},
+		Files: []*internalpb.ImportFile{
+			{Id: 1, Paths: []string{"/test/file.json"}},
+		},
+		Options: []*commonpb.KeyValuePair{
+			{Key: "timeout", Value: "300s"},
+			{Key: "l0_import", Value: "true"},
+		},
+		DataTimestamp: 123456789,
+		JobID:         2000,
+	}
+
+	resp, err := server.createImportJobFromAck(ctx, req)
+
+	s.NoError(err)
+	s.NotNil(resp)
+	s.Equal(int32(0), resp.GetStatus().GetCode())
+	s.Equal("2000", resp.GetJobID())
+
+	s.NotNil(savedJob)
+	s.Equal(internalpb.ImportJobState_Failed, savedJob.GetState())
+	s.Contains(savedJob.GetReason(), "l0 import is disabled")
+	// Failed at creation must carry a real cleanup ts so GC can reclaim the job.
+	s.NotEqual(uint64(math.MaxUint64), savedJob.GetCleanupTs())
+}
+
+func (s *ImportServicesSuite) TestCreateImportJobFromAck_L0ImportEnabledCreatesPendingJob() {
+	paramtable.Init()
+	ctx := context.Background()
+
+	params := paramtable.Get()
+	params.Save(params.DataCoordCfg.EnableL0Import.Key, "true")
+	defer params.Reset(params.DataCoordCfg.EnableL0Import.Key)
+
+	mockHandler := NewNMockHandler(s.T())
+	mockHandler.EXPECT().GetCollection(mock.Anything, mock.Anything).Return(&collectionInfo{
+		ID:            100,
+		VChannelNames: []string{"v1"},
+	}, nil)
+
+	var savedJob *datapb.ImportJob
+	catalog := mocks.NewDataCoordCatalog(s.T())
+	catalog.EXPECT().ListImportJobs(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListPreImportTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, job *datapb.ImportJob) error {
+		savedJob = job
+		return nil
+	})
+
+	importMeta, err := NewImportMeta(context.TODO(), catalog, nil, nil)
+	s.NoError(err)
+
+	server := &Server{
+		handler:    mockHandler,
+		importMeta: importMeta,
+	}
+	server.stateCode.Store(commonpb.StateCode_Healthy)
+
+	mockAllocator := allocator.NewMockAllocator(s.T())
+	mockAllocator.EXPECT().AllocN(mock.Anything).Return(int64(1000), int64(1002), nil)
+	server.allocator = mockAllocator
+
+	req := &internalpb.ImportRequestInternal{
+		CollectionID:   100,
+		CollectionName: "test_collection",
+		PartitionIDs:   []int64{1},
+		ChannelNames:   []string{"v1"},
+		Schema:         &schemapb.CollectionSchema{Name: "test_collection"},
+		Files: []*internalpb.ImportFile{
+			{Id: 1, Paths: []string{"/test/file.json"}},
+		},
+		Options: []*commonpb.KeyValuePair{
+			{Key: "timeout", Value: "300s"},
+			{Key: "l0_import", Value: "true"},
+		},
+		DataTimestamp: 123456789,
+		JobID:         2000,
+	}
+
+	resp, err := server.createImportJobFromAck(ctx, req)
+
+	s.NoError(err)
+	s.NotNil(resp)
+	s.Equal(int32(0), resp.GetStatus().GetCode())
+
+	s.NotNil(savedJob)
+	s.Equal(internalpb.ImportJobState_Pending, savedJob.GetState())
+}
+
 // Helper types are defined in import_callbacks_test.go (mockBalancerImpl, mockBroadcastAPIImpl, newMockBroadcastAPIImpl)

internal/datacoord/services.go

@@ -1877,7 +1877,7 @@ func (s *Server) ImportV2(ctx context.Context, in *internalpb.ImportRequestInter
 		Status: merr.Success(),
 	}
 
-	mlog.Info(context.TODO(), "receive import request from proxy, will broadcast",
+	mlog.Info(context.TODO(), "receive import request from proxy",
 		mlog.Int("fileNum", len(in.GetFiles())),
 		mlog.Any("files", in.GetFiles()),
 		mlog.Any("options", in.GetOptions()))
@@ -1890,6 +1890,18 @@ func (s *Server) ImportV2(ctx context.Context, in *internalpb.ImportRequestInter
 		return resp, nil
 	}
 
+	// Reject L0 import when disabled (default). Restoring L0 (delete-only) segments
+	// is incompatible with commit_timestamp (2PC / replication imports), where it
+	// silently breaks delete semantics. Backups should have their L0 deletes folded
+	// into per-segment deltalogs beforehand; set dataCoord.import.enableL0Import=true
+	// to re-enable the legacy behavior.
+	if importutilv2.IsL0Import(in.GetOptions()) && !Params.DataCoordCfg.EnableL0Import.GetAsBool() {
+		resp.Status = merr.Status(merr.WrapErrImportFailed("l0 import is disabled " +
+			"(dataCoord.import.enableL0Import=false); fold L0 deletes into data segment deltalogs " +
+			"before restore, or set the config to true to re-enable the legacy L0 import"))
+		return resp, nil
+	}
+
 	// Use the incoming JobID if provided (backward compat: old proxy allocates jobID
 	// before sending broadcast RPC, which is forwarded here with the original jobID).
 	// Otherwise allocate a new one.
@@ -1932,6 +1944,15 @@ func (s *Server) ImportV2(ctx context.Context, in *internalpb.ImportRequestInter
 
 // createImportJobFromAck creates an import job from ack callback.
 // This is called internally when broadcast ack is received.
+// Note: the pre-broadcast L0-import gate in ImportV2 covers only locally
+// originated imports. Replicated import messages (CDC) from a cluster with
+// enableL0Import=true land here directly without passing that gate, so it must
+// be re-checked. The gate here must NOT return an error: ack callbacks are
+// retried forever (callMessageAckCallbackUntilDone), and skipping job creation
+// would wedge the replicated CommitImport path (HandleCommitVchannel retries
+// on job-not-found). Instead the job is created directly in Failed state — a
+// terminal no-op for both commitImportV2AckCallback and HandleCommitVchannel —
+// and the failure stays visible via GetImportProgress.
 func (s *Server) createImportJobFromAck(ctx context.Context, in *internalpb.ImportRequestInternal) (*internalpb.ImportResponse, error) {
 	if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
 		return &internalpb.ImportResponse{
@@ -1954,9 +1975,15 @@ func (s *Server) createImportJobFromAck(ctx context.Context, in *internalpb.Impo
 		return resp, nil
 	}
 
+	// See the function comment: an L0 import reaching this callback while the
+	// gate is disabled (replicated from a cluster where it is enabled, or a
+	// config flip between broadcast and ack) is terminally failed below instead
+	// of running ungated or returning an error (which would retry forever).
+	l0ImportDisabled := importutilv2.IsL0Import(in.GetOptions()) && !Params.DataCoordCfg.EnableL0Import.GetAsBool()
+
 	files := in.GetFiles()
 	isBackup := importutilv2.IsBackup(in.GetOptions())
-	if isBackup {
+	if isBackup && !l0ImportDisabled {
 		files, err = ListBinlogImportRequestFiles(ctx, s.meta.chunkManager, files, in.GetOptions())
 		if err != nil {
 			resp.Status = merr.Status(err)
@@ -2013,6 +2040,14 @@ func (s *Server) createImportJobFromAck(ctx context.Context, in *internalpb.Impo
 		},
 		tr: timerecord.NewTimeRecorder("import job"),
 	}
+	if l0ImportDisabled {
+		mlog.Warn(ctx, "l0 import is disabled, creating the job in Failed state",
+			mlog.Int64("jobID", jobID), mlog.Int64("collectionID", in.GetCollectionID()))
+		UpdateJobState(internalpb.ImportJobState_Failed)(job)
+		UpdateJobReason("l0 import is disabled (dataCoord.import.enableL0Import=false); fold L0 deletes " +
+			"into data segment deltalogs before restore, or set the config to true on this cluster " +
+			"to re-enable the legacy L0 import")(job)
+	}
 	err = s.importMeta.AddJob(ctx, job)
 	if err != nil {
 		resp.Status = merr.Status(merr.Wrap(err, "add import job failed"))

pkg/util/paramtable/component_param.go

@@ -5335,6 +5335,7 @@ type dataCoordConfig struct {
 	MaxImportJobNum                 ParamItem `refreshable:"true"`
 	WaitForIndex                    ParamItem `refreshable:"true"`
 	ImportInReplicatingCluster      ParamItem `refreshable:"true"`
+	EnableL0Import                  ParamItem `refreshable:"true"`
 	ImportPreAllocIDExpansionFactor ParamItem `refreshable:"true"`
 	ImportFileNumPerSlot            ParamItem `refreshable:"true"`
 	ImportMemoryLimitPerSlot        ParamItem `refreshable:"true"`
@@ -6565,6 +6566,20 @@ if param targetScalarIndexVersion is not set, the default value is -1, which mea
 	}
 	p.ImportInReplicatingCluster.Init(base.mgr)
 
+	p.EnableL0Import = ParamItem{
+		Key:     "dataCoord.import.enableL0Import",
+		Version: "2.7.0",
+		Doc: "Whether to allow importing L0 (delete-only) segments during a binlog/backup import. " +
+			"Disabled by default because restoring L0 segments is incompatible with commit_timestamp " +
+			"(two-phase-commit / replication imports), where it silently breaks delete semantics. " +
+			"Fold the L0 deletes into per-segment deltalogs before restore instead; set to true only to " +
+			"re-enable the legacy L0 import behavior.",
+		DefaultValue: "false",
+		PanicIfEmpty: false,
+		Export:       true,
+	}
+	p.EnableL0Import.Init(base.mgr)
+
 	p.ImportPreAllocIDExpansionFactor = ParamItem{
 		Key:          "dataCoord.import.preAllocateIDExpansionFactor",
 		Version:      "2.5.13",

pkg/util/paramtable/component_param_test.go

@@ -636,6 +636,7 @@ func TestComponentParam(t *testing.T) {
 		assert.Equal(t, 1024, Params.MaxImportJobNum.GetAsInt())
 		assert.Equal(t, true, Params.WaitForIndex.GetAsBool())
 		assert.Equal(t, false, Params.ImportInReplicatingCluster.GetAsBool())
+		assert.Equal(t, false, Params.EnableL0Import.GetAsBool())
 		assert.Equal(t, 4, Params.ImportFileNumPerSlot.GetAsInt())
 		assert.Equal(t, 160*1024*1024, Params.ImportMemoryLimitPerSlot.GetAsInt())
 

tests/integration/import/import_test.go

@@ -57,6 +57,10 @@ type BulkInsertSuite struct {
 
 func (s *BulkInsertSuite) SetupSuite() {
 	s.WithMilvusConfig(paramtable.Get().RootCoordCfg.DmlChannelNum.Key, "4")
+	// The binlog-import cases restore legacy L0 (delete-only) segments, which is
+	// disabled by default (dataCoord.import.enableL0Import=false); re-enable it
+	// so they keep covering the legacy path.
+	s.WithMilvusConfig(paramtable.Get().DataCoordCfg.EnableL0Import.Key, "true")
 	s.MiniClusterSuite.SetupSuite()
 }
 

tests/restful_client_v2/testcases/test_import_2pc_infra_dependent.py

@@ -422,6 +422,12 @@ class Import2PCInfraBase(TestBase):
         if partition_name is not None:
             payload["partitionName"] = partition_name
         rsp = self.import_job_client.create_import_jobs(payload)
+        if (
+            payload["options"].get("l0_import") == "true"
+            and rsp.get("code") != 0
+            and "l0 import is disabled" in str(rsp.get("message", ""))
+        ):
+            pytest.skip("dataCoord.import.enableL0Import is disabled on this cluster")
         assert rsp["code"] == 0, rsp
         return rsp["data"]["jobId"]