enhance: [cherry-pick] Move compactor into sub package

This PR consists of the following commits:

- enhance: Tidy compactor and remove dup codes (milvus-io#32198)
- fix: Fix l0 compactor may cause DN from OOM (milvus-io#33554)
- enhance: Add deltaRowCount in l0 compaction (milvus-io#33997)
- enhance: enable stream writer in compactions (milvus-io#32612)

See also: milvus-io#32451, milvus-io#33547, milvus-io#33998, milvus-io#31679
pr: milvus-io#32198, milvus-io#33554, milvus-io#33997, milvus-io#32612

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

Makefile

@@ -475,6 +475,7 @@ generate-mockery-datanode: getdeps
 	$(INSTALL_PATH)/mockery --name=BinlogIO --dir=$(PWD)/internal/datanode/io --output=$(PWD)/internal/datanode/io --filename=mock_binlogio.go --with-expecter --structname=MockBinlogIO --outpkg=io --inpackage
 	$(INSTALL_PATH)/mockery --name=FlowgraphManager --dir=$(PWD)/internal/datanode --output=$(PWD)/internal/datanode --filename=mock_fgmanager.go --with-expecter --structname=MockFlowgraphManager --outpkg=datanode --inpackage
 	$(INSTALL_PATH)/mockery --name=ChannelManager --dir=$(PWD)/internal/datanode --output=$(PWD)/internal/datanode --filename=mock_channelmanager.go --with-expecter --structname=MockChannelManager --outpkg=datanode --inpackage
+	$(INSTALL_PATH)/mockery --name=Compactor --dir=$(PWD)/internal/datanode/compaction --output=$(PWD)/internal/datanode/compaction --filename=mock_compactor.go --with-expecter --structname=MockCompactor --outpkg=compaction --inpackage
 
 generate-mockery-metastore: getdeps
 	$(INSTALL_PATH)/mockery --name=RootCoordCatalog --dir=$(PWD)/internal/metastore --output=$(PWD)/internal/metastore/mocks --filename=mock_rootcoord_catalog.go --with-expecter --structname=RootCoordCatalog --outpkg=mocks
@@ -517,4 +518,4 @@ mmap-migration:
 	@source $(PWD)/scripts/setenv.sh && \
     		mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && \
     		GO111MODULE=on $(GO) build -pgo=$(PGO_PATH)/default.pgo -ldflags="-r $${RPATH} -X '$(OBJPREFIX).BuildTags=$(BUILD_TAGS)' -X '$(OBJPREFIX).BuildTime=$(BUILD_TIME)' -X '$(OBJPREFIX).GitCommit=$(GIT_COMMIT)' -X '$(OBJPREFIX).GoVersion=$(GO_VERSION)'" \
-    		-tags dynamic -o $(INSTALL_PATH)/mmap-migration $(MMAP_MIGRATION_PATH)/main.go 1>/dev/null
+    		-tags dynamic -o $(INSTALL_PATH)/mmap-migration $(MMAP_MIGRATION_PATH)/main.go 1>/dev/null

internal/datacoord/compaction_view.go

@@ -148,7 +148,8 @@ func (v *SegmentView) Equal(other *SegmentView) bool {
 		v.DeltaSize == other.DeltaSize &&
 		v.BinlogCount == other.BinlogCount &&
 		v.StatslogCount == other.StatslogCount &&
-		v.DeltalogCount == other.DeltalogCount
+		v.DeltalogCount == other.DeltalogCount &&
+		v.DeltaRowCount == other.DeltaRowCount
 }
 
 func (v *SegmentView) String() string {