@@ -1789,6 +1789,76 @@ func (tc *Catalog) FlushCollectionCompactionAndAttachedFunction(
17891789 return flushCollectionInfo , nil
17901790}
17911791
1792+ // FlushCollectionCompactionAndAttachedFunctionExtended atomically updates multiple collection compaction data
1793+ // and attached function completion offset in a single transaction.
1794+ // NOTE: This does NOT advance next_nonce - that is done separately by AdvanceAttachedFunction.
1795+ // This only updates the completion_offset to record how far we've processed.
1796+ // This is only supported for versioned collections (the modern/default path).
1797+ func (tc * Catalog ) FlushCollectionCompactionAndAttachedFunctionExtended (
1798+ ctx context.Context ,
1799+ collectionCompactions []* model.FlushCollectionCompaction ,
1800+ attachedFunctionID uuid.UUID ,
1801+ runNonce uuid.UUID ,
1802+ completionOffset int64 ,
1803+ ) (* model.ExtendedFlushCollectionInfo , error ) {
1804+ if ! tc .versionFileEnabled {
1805+ // Attached-function-based compactions are only supported with versioned collections
1806+ log .Error ("FlushCollectionCompactionAndAttachedFunctionExtended is only supported for versioned collections" )
1807+ return nil , errors .New ("attached-function-based compaction requires versioned collections" )
1808+ }
1809+
1810+ if len (collectionCompactions ) == 0 {
1811+ return nil , errors .New ("at least one collection compaction is required" )
1812+ }
1813+
1814+ flushInfos := make ([]* model.FlushCollectionInfo , 0 , len (collectionCompactions ))
1815+
1816+ err := tc .txImpl .Transaction (ctx , func (txCtx context.Context ) error {
1817+ var err error
1818+ // Get the transaction from context to pass to FlushCollectionCompactionForVersionedCollection
1819+ tx := dbcore .GetDB (txCtx )
1820+
1821+ // Handle all collection compactions
1822+ for _ , collectionCompaction := range collectionCompactions {
1823+ log .Info ("FlushCollectionCompactionAndAttachedFunctionExtended" , zap .String ("collection_id" , collectionCompaction .ID .String ()))
1824+ flushInfo , err := tc .FlushCollectionCompactionForVersionedCollection (txCtx , collectionCompaction , tx )
1825+ if err != nil {
1826+ return err
1827+ }
1828+ flushInfos = append (flushInfos , flushInfo )
1829+ }
1830+
1831+ // Update ONLY completion_offset - next_nonce was already advanced by AdvanceAttachedFunction
1832+ // We still validate runNonce to ensure we're updating the correct nonce
1833+ err = tc .metaDomain .AttachedFunctionDb (txCtx ).UpdateCompletionOffset (attachedFunctionID , runNonce , completionOffset )
1834+ if err != nil {
1835+ return err
1836+ }
1837+
1838+ return nil
1839+ })
1840+
1841+ if err != nil {
1842+ return nil , err
1843+ }
1844+
1845+ // Populate attached function fields with authoritative values from database
1846+ for _ , flushInfo := range flushInfos {
1847+ flushInfo .AttachedFunctionCompletionOffset = & completionOffset
1848+ }
1849+
1850+ // Log with first collection ID (typically the output collection)
1851+ log .Info ("FlushCollectionCompactionAndAttachedFunctionExtended" ,
1852+ zap .String ("first_collection_id" , collectionCompactions [0 ].ID .String ()),
1853+ zap .Int ("collection_count" , len (collectionCompactions )),
1854+ zap .String ("attached_function_id" , attachedFunctionID .String ()),
1855+ zap .Int64 ("completion_offset" , completionOffset ))
1856+
1857+ return & model.ExtendedFlushCollectionInfo {
1858+ Collections : flushInfos ,
1859+ }, nil
1860+ }
1861+
17921862func (tc * Catalog ) validateVersionFile (versionFile * coordinatorpb.CollectionVersionFile , collectionID string , version int64 ) error {
17931863 if versionFile .GetCollectionInfoImmutable ().GetCollectionId () != collectionID {
17941864 log .Error ("collection id mismatch" , zap .String ("collection_id" , collectionID ), zap .String ("version_file_collection_id" , versionFile .GetCollectionInfoImmutable ().GetCollectionId ()))
0 commit comments