fix: 修复ddc_blob references字段过大导致慢查询问题 #2181

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/model/TDdcBlob.kt

@@ -85,5 +85,10 @@ data class TDdcBlob(
      * ref类型引用 ref/{bucket}/{key}
      * blob类型引用 blob/{blobId}
      */
-    var references: Set<String> = emptySet()
+    @Deprecated("性能原因不在使用，关联关系存在TDdcReferenceBlob中")
+    var references: Set<String> = emptySet(),
+    /**
+     * 被引用的次数，计数为0时将被清理
+     */
+    var refCount: Long = 0L,
 )

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/model/TDdcBlobRef.kt

@@ -0,0 +1,37 @@
+package com.tencent.bkrepo.ddc.model
+
+import org.springframework.data.mongodb.core.index.CompoundIndex
+import org.springframework.data.mongodb.core.index.CompoundIndexes
+import org.springframework.data.mongodb.core.mapping.Document
+
+
+@Document("ddc_blob_ref")
+@CompoundIndexes(
+    CompoundIndex(
+        name = "projectId_repoName_blobId_ref_idx",
+        def = "{'projectId': 1, 'repoName': 1, 'blobId': 1, 'ref': 1}",
+        unique = true,
+        background = true
+    ),
+    CompoundIndex(
+        name = "projectId_repoName_ref_blobId_idx",
+        def = "{'projectId': 1, 'repoName': 1, 'ref': 1, 'blobId': 1}",
+        unique = true,
+        background = true
+    ),
+)
+data class TDdcBlobRef(
+    var id: String? = null,
+    var projectId: String,
+    var repoName: String,
+    /**
+     * blob blake3 hash
+     */
+    var blobId: String,
+    /**
+     * 引用了该blob的ref或blob，ref的inline blob中直接或间接引用的所有blob都会关联到ref
+     * ref类型引用 ref/{bucket}/{key}
+     * blob类型引用 blob/{blobId}
+     */
+    var ref: String,
+)

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/pojo/Blob.kt

@@ -39,7 +39,6 @@ data class Blob(
     val size: Long,
     val blobId: ContentHash,
     val contentId: ContentHash,
-    val references: Set<String> = emptySet(),
     val sha1: String? = null,
 ) {
     companion object {
@@ -52,7 +51,6 @@ data class Blob(
                 size = size,
                 blobId = ContentHash.fromHex(blobId),
                 contentId = ContentHash.fromHex(contentId),
-                references = references,
                 sha1 = sha1
             )
         }

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/repository/BlobRefRepository.kt

@@ -0,0 +1,39 @@
+package com.tencent.bkrepo.ddc.repository
+
+import com.tencent.bkrepo.common.mongo.dao.simple.SimpleMongoDao
+import com.tencent.bkrepo.ddc.model.TDdcBlobRef
+import org.springframework.dao.DuplicateKeyException
+import org.springframework.data.mongodb.core.query.Criteria
+import org.springframework.data.mongodb.core.query.Query
+import org.springframework.data.mongodb.core.query.isEqualTo
+import org.springframework.stereotype.Repository
+
+@Repository
+class BlobRefRepository : SimpleMongoDao<TDdcBlobRef>() {
+    fun addRefToBlob(projectId: String, repoName: String, bucket: String, refKey: String, blobIds: Set<String>) {
+        blobIds.forEach {
+            try {
+                insert(
+                    TDdcBlobRef(
+                        id = null,
+                        projectId = projectId,
+                        repoName = repoName,
+                        blobId = it,
+                        ref = buildRef(bucket, refKey)
+                    )
+                )
+            } catch (ignore: DuplicateKeyException) {
+            }
+        }
+    }
+
+    fun removeRefFromBlob(projectId: String, repoName: String, bucket: String, refKey: String): List<TDdcBlobRef> {
+        val criteria = Criteria
+            .where(TDdcBlobRef::projectId.name).isEqualTo(projectId)
+            .and(TDdcBlobRef::repoName.name).isEqualTo(repoName)
+            .and(TDdcBlobRef::ref.name).isEqualTo(buildRef(bucket, refKey))
+        return determineMongoTemplate().findAllAndRemove(Query(criteria), TDdcBlobRef::class.java)
+    }
+
+    private fun buildRef(bucket: String, key: String): String = "ref/$bucket/$key"
+}

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/repository/BlobRepository.kt

@@ -76,11 +76,11 @@ class BlobRepository : SimpleMongoDao<TDdcBlob>() {
         }
     }
 
-    fun addRefToBlob(projectId: String, repoName: String, bucket: String, refKey: String, blobIds: Set<String>) {
+    fun incRefCount(projectId: String, repoName: String, blobIds: Set<String>, inc: Long = 1L) {
         val criteria = TDdcBlob::projectId.isEqualTo(projectId)
             .and(TDdcBlob::repoName.name).isEqualTo(repoName)
             .and(TDdcBlob::blobId.name).inValues(blobIds)
-        val update = Update().addToSet(TDdcBlob::references.name, "ref/$bucket/$refKey")
+        val update = Update().inc(TDdcBlob::refCount.name, inc)
         updateMulti(Query(criteria), update)
     }
 

src/backend/ddc/biz-ddc/src/main/kotlin/com/tencent/bkrepo/ddc/service/BlobService.kt

@@ -38,13 +38,15 @@ import com.tencent.bkrepo.ddc.exception.BlobNotFoundException
 import com.tencent.bkrepo.ddc.model.TDdcBlob
 import com.tencent.bkrepo.ddc.pojo.Blob
 import com.tencent.bkrepo.ddc.pojo.Reference
+import com.tencent.bkrepo.ddc.repository.BlobRefRepository
 import com.tencent.bkrepo.ddc.repository.BlobRepository
 import org.springframework.stereotype.Service
 import java.time.LocalDateTime
 
 @Service
 class BlobService(
     private val blobRepository: BlobRepository,
+    private val blobRefRepository: BlobRefRepository,
     private val nodeService: NodeService,
     private val storageManager: StorageManager
 ) {
@@ -107,10 +109,15 @@ class BlobService(
     }
 
     fun addRefToBlobs(ref: Reference, blobIds: Set<String>) {
-        blobRepository.addRefToBlob(ref.projectId, ref.repoName, ref.bucket, ref.key.toString(), blobIds)
+        blobRefRepository.addRefToBlob(ref.projectId, ref.repoName, ref.bucket, ref.key.toString(), blobIds)
+        blobRepository.incRefCount(ref.projectId, ref.repoName, blobIds)
     }
 
     fun removeRefFromBlobs(projectId: String, repoName: String, bucket: String, key: String) {
+        val blobIds = HashSet<String>()
+        blobRefRepository.removeRefFromBlob(projectId, repoName, bucket, key).mapTo(blobIds) { it.blobId }
+        blobRepository.incRefCount(projectId, repoName, blobIds, -1L)
+        // 兼容旧逻辑，所有blob的references字段为空后可以移除该逻辑
         blobRepository.removeRefFromBlob(projectId, repoName, bucket, key)
     }
 }

src/backend/job/biz-job/src/main/kotlin/com/tencent/bkrepo/job/batch/task/ddc/DdcBlobCleanupJob.kt

@@ -38,6 +38,7 @@ import org.springframework.data.mongodb.core.query.Criteria
 import org.springframework.data.mongodb.core.query.Query
 import org.springframework.data.mongodb.core.query.exists
 import org.springframework.data.mongodb.core.query.isEqualTo
+import org.springframework.data.mongodb.core.query.lte
 import org.springframework.data.mongodb.core.query.size
 import org.springframework.stereotype.Component
 import java.time.LocalDateTime
@@ -53,11 +54,16 @@ class DdcBlobCleanupJob(
     override fun buildQuery(): Query {
         val referencesCriteria = Criteria().orOperator(
             Blob::references.exists(false),
-            Blob::references.size(0)
+            Blob::references.size(0),
+        )
+        val refCountCriteria = Criteria().orOperator(
+            Blob::refCount.exists(false),
+            Blob::refCount.lte(0L),
         )
         // 最近1小时上传的blob不清理，避免将未finalized的ref所引用的blob清理掉
         val criteria = Criteria().andOperator(
             referencesCriteria,
+            refCountCriteria,
             Criteria.where("lastModifiedDate").lt(LocalDateTime.now().minusHours(1L))
         )
         val query = Query(criteria)
@@ -93,10 +99,20 @@ class DdcBlobCleanupJob(
         val projectId: String,
         val repoName: String,
         val blobId: String,
-        val references: Set<String> = emptySet()
+        val references: Set<String> = emptySet(),
+        val refCount: Long = 0L,
+    )
+
+    data class BlobRef(
+        val id: String,
+        val projectId: String,
+        val repoName: String,
+        val blobId: String,
+        val ref: String,
     )
 
     companion object {
         const val COLLECTION_NAME = "ddc_blob"
+        const val BLOB_REF_COLLECTION_NAME = "ddc_blob_ref"
     }
 }

src/backend/job/biz-job/src/main/kotlin/com/tencent/bkrepo/job/batch/task/ddc/ExpiredDdcRefCleanupJob.kt

@@ -85,13 +85,37 @@ class ExpiredDdcRefCleanupJob(
             )
         }
 
-        // 从blob ref列表中移除ref
+        removeBlobRef(row)
+    }
+
+    private fun removeBlobRef(row: Ref) {
+        // 移除blob与ref关联关系
         val refKey = "ref/${row.bucket}/${row.key}"
-        val criteria = Criteria
+        var criteria = Criteria
+            .where(DdcBlobCleanupJob.BlobRef::projectId.name).isEqualTo(row.projectId)
+            .and(DdcBlobCleanupJob.BlobRef::repoName.name).isEqualTo(row.repoName)
+            .and(DdcBlobCleanupJob.BlobRef::ref.name).isEqualTo(refKey)
+        val blobIds = HashSet<String>()
+        mongoTemplate.findAllAndRemove(
+            Query(criteria),
+            DdcBlobCleanupJob.BlobRef::class.java,
+            DdcBlobCleanupJob.BLOB_REF_COLLECTION_NAME
+        ).mapTo(blobIds) { it.blobId }
+
+        // 减少blob引用计数
+        criteria = Criteria
+            .where(DdcBlobCleanupJob.Blob::projectId.name).isEqualTo(row.projectId)
+            .and(DdcBlobCleanupJob.Blob::repoName.name).isEqualTo(row.repoName)
+            .and(DdcBlobCleanupJob.Blob::blobId.name).inValues(blobIds)
+        var update = Update().inc(DdcBlobCleanupJob.Blob::refCount.name, -1L)
+        mongoTemplate.updateMulti(Query(criteria), update, DdcBlobCleanupJob.COLLECTION_NAME)
+
+        // 兼容旧逻辑，从blob ref列表中移除ref，所有blob的reference字段都清空后可移除该代码
+        criteria = Criteria
             .where(DdcBlobCleanupJob.Blob::projectId.name).isEqualTo(row.projectId)
             .and(DdcBlobCleanupJob.Blob::repoName.name).isEqualTo(row.repoName)
             .and(DdcBlobCleanupJob.Blob::references.name).inValues(refKey)
-        val update = Update().pull(DdcBlobCleanupJob.Blob::references.name, refKey)
+        update = Update().pull(DdcBlobCleanupJob.Blob::references.name, refKey)
         mongoTemplate.updateMulti(Query(criteria), update, DdcBlobCleanupJob.COLLECTION_NAME)
     }