diff --git a/configs/milvus.yaml b/configs/milvus.yaml index 4158a0c6c5..ba1fd552c4 100644 --- a/configs/milvus.yaml +++ b/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. diff --git a/internal/datacoord/import_services_test.go b/internal/datacoord/import_services_test.go index 56a9997d2f..ccb9b65c90 100644 --- a/internal/datacoord/import_services_test.go +++ b/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) diff --git a/internal/datacoord/services.go b/internal/datacoord/services.go index 53aa4341b8..1db3a95348 100644 --- a/internal/datacoord/services.go +++ b/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")) diff --git a/pkg/util/paramtable/component_param.go b/pkg/util/paramtable/component_param.go index 233acce98f..f150450531 100644 --- a/pkg/util/paramtable/component_param.go +++ b/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", diff --git a/pkg/util/paramtable/component_param_test.go b/pkg/util/paramtable/component_param_test.go index 033d7e5f38..08aa1fca48 100644 --- a/pkg/util/paramtable/component_param_test.go +++ b/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()) diff --git a/tests/integration/import/import_test.go b/tests/integration/import/import_test.go index c6c26bb13c..50db86381b 100644 --- a/tests/integration/import/import_test.go +++ b/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() } diff --git a/tests/restful_client_v2/testcases/test_import_2pc_infra_dependent.py b/tests/restful_client_v2/testcases/test_import_2pc_infra_dependent.py index 8b081a406f..ef0877f05f 100644 --- a/tests/restful_client_v2/testcases/test_import_2pc_infra_dependent.py +++ b/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"]