Method createNewArchiveStoreVersionForBackfill

memstore/backfill.go:108–266

(
	backfillPatches map[int32]*backfillPatch, reporter BackfillJobDetailReporter, jobKey string)

}

func (shard *TableShard) createNewArchiveStoreVersionForBackfill(
	backfillPatches map[int32]*backfillPatch, reporter BackfillJobDetailReporter, jobKey string) (err error) {
	// Block column deletion
	shard.columnDeletion.Lock()
	defer shard.columnDeletion.Unlock()

	// Snapshot schema
	shard.Schema.RLock()
	columnDeletions := shard.Schema.GetColumnDeletions()
	sortColumns := shard.Schema.Schema.ArchivingSortColumns
	primaryKeyColumns := shard.Schema.Schema.PrimaryKeyColumns
	dataTypes := shard.Schema.ValueTypeByColumn
	defaultValues := shard.Schema.DefaultValues
	numColumns := len(shard.Schema.ValueTypeByColumn)
	shard.Schema.RUnlock()

	var numAffectedDays int
	dayIdx := 1
	reporter(jobKey, func(status *BackfillJobDetail) {
		status.Stage = BackfillApplyPatch
		status.Current = 0
		status.Total = len(backfillPatches)
	})

	lockTimer := utils.GetReporter(shard.Schema.Schema.Name, shard.ShardID).
		GetTimer(utils.BackfillLockTiming)
	var totalLockDuration time.Duration
	defer func() {
		lockTimer.Record(totalLockDuration)
		reporter(jobKey, func(status *BackfillJobDetail) {
			status.LockDuration = totalLockDuration
		})
	}()

	// Only those batches that are affected and changed need to be cleaned.
	for day, patch := range backfillPatches {
		baseBatch := shard.ArchiveStore.CurrentVersion.RequestBatch(day)

		var requestedVPs []memCom.ArchiveVectorParty
		for columnID := 0; columnID < numColumns; columnID++ {
			requestedVP := baseBatch.RequestVectorParty(columnID)
			requestedVP.WaitForDiskLoad()
			requestedVPs = append(requestedVPs, requestedVP)
		}

		backfillCtx := newBackfillContext(baseBatch, patch, shard.Schema, columnDeletions, sortColumns,
			primaryKeyColumns, dataTypes, defaultValues, shard.HostMemoryManager)

		// Real backfill implementation.
		if err = backfillCtx.backfill(reporter, jobKey); err != nil {
			UnpinVectorParties(requestedVPs)
			backfillCtx.release()
			return
		}

		if backfillCtx.okForEarlyUnpin {
			UnpinVectorParties(requestedVPs)
		}