[BUG] Fix log fork bugs (#4356)

## Description of changes

*Summarize the changes made by this PR.*
 - Improvements & Bug fixes
- Fixes a typo in log fork impl that led to wrong enumeration offset for
the new collection
- Replaces updates with insert when populating the offsets for the new
collection
 - New functionality
   - ...

## Test plan
*How are these changes tested?*

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js,
`cargo test` for rust

## Documentation Changes
*Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
repository](https://github.com/chroma-core/docs)?*

---------

Co-authored-by: Sicheng Pan <[email protected]>

Author: Sicheng-Pan

go/pkg/common/errors.go

@@ -21,7 +21,7 @@ var (
 	ErrCollectionNameEmpty                   = errors.New("collection name is empty")
 	ErrCollectionUniqueConstraintViolation   = errors.New("collection unique constraint violation")
 	ErrCollectionDeleteNonExistingCollection = errors.New("delete non existing collection")
-	ErrCollectionLogPositionStale            = errors.New("collection log position Stale")
+	ErrCollectionLogPositionStale            = errors.New("collection log position stale")
 	ErrCollectionVersionStale                = errors.New("collection version stale")
 	ErrCollectionVersionInvalid              = errors.New("collection version invalid")
 	ErrCollectionVersionFileNameStale        = errors.New("collection version file name stale")

go/pkg/log/repository/log.go

@@ -168,28 +168,21 @@ func (r *LogRepository) ForkRecords(ctx context.Context, sourceCollectionID stri
 		// Some logs are forked, the min offset is guaranteed to be larger than source compaction offset
 		compactionOffset = uint64(targetBounds.MinOffset - 1)
 	}
-	err = queriesWithTx.UpdateCollectionCompactionOffsetPosition(ctx, log.UpdateCollectionCompactionOffsetPositionParams{
-		ID:                             targetCollectionID,
-		RecordCompactionOffsetPosition: int64(compactionOffset),
-	})
-	if err != nil {
-		trace_log.Error("Error in updating compaction offset for target collection", zap.Error(err), zap.String("collectionId", targetCollectionID))
-		return
-	}
-
 	if targetBounds.MaxOffset == 0 {
 		// Either the source collection is empty or no log is forked
 		enumerationOffset = uint64(sourceBounds.RecordEnumerationOffsetPosition)
 	} else {
 		// Some logs are forked. The max offset is the enumeration offset
-		compactionOffset = uint64(targetBounds.MaxOffset)
+		enumerationOffset = uint64(targetBounds.MaxOffset)
 	}
-	err = queriesWithTx.UpdateCollectionEnumerationOffsetPosition(ctx, log.UpdateCollectionEnumerationOffsetPositionParams{
+
+	_, err = queriesWithTx.InsertCollection(ctx, log.InsertCollectionParams{
 		ID:                              targetCollectionID,
+		RecordCompactionOffsetPosition:  int64(compactionOffset),
 		RecordEnumerationOffsetPosition: int64(enumerationOffset),
 	})
 	if err != nil {
-		trace_log.Error("Error in updating enumeration offset for target collection", zap.Error(err), zap.String("collectionId", targetCollectionID))
+		trace_log.Error("Error in updating offset for target collection", zap.Error(err), zap.String("collectionId", targetCollectionID))
 		return
 	}
 	return