You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: ingestion.go
    path of file: ./repos/aresdb/memstore
    
    the code of the file until where you have to start completion: //  Copyright (c) 2017-2018 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the Lic
    
func (shard *TableShard) ApplyUpsertBatch(upsertBatch *common.UpsertBatch, redoLogFile int64, offset uint32, skipBackfillRows bool) (bool, error) {
	shard.Schema.RLock()
	valueTypeByColumn := shard.Schema.ValueTypeByColumn
	columnDeletions := shard.Schema.GetColumnDeletions()
	allowMissingEventTime := shard.Schema.Schema.Config.AllowMissingEventTime
	shard.Schema.RUnlock()
	primaryKeyColumns := shard.Schema.GetPrimaryKeyColumns()
	// IsFactTable should be immutable.
	isFactTable := shard.Schema.Schema.IsFactTable

	// This is the upsertbatch column index that points to the first logic column (which is
	// event time for fact table).
	eventTimeColumnIndex := -1

	// Validate columns in upsert batch are valid.
	for i := 0; i < upsertBatch.NumColumns; i++ {
		columnID, _ := upsertBatch.GetColumnID(i)
		if columnID >= len(valueTypeByColumn) {
			return false, utils.StackError(nil, "Unrecognized column id %d in upsert batch", columnID)
		}

		columnType, _ := upsertBatch.GetColumnType(i)
		if valueTypeByColumn[columnID] != columnType {
			return false, utils.StackError(
				nil,
				"Mismatched data type (upsert batch: %d, schema %d) for table %s shard %d column %d", columnType, valueTypeByColumn[columnID], shard.Schema.Schema.Name, shard.ShardID, columnID)
		}

		if columnID == 0 && isFactTable {
			eventTimeColumnIndex = i
		}
	}

	// For fact table ingestion, we will need to get the event time from the first column. we don't
	// have to validate the column type in the upsertbatch because the loop above already handled it.
	if isFactTable && eventTimeColumnIndex < 0 && !allowMissingEventTime {
		utils.GetReporter(shard.Schema.Schema.Name, shard.ShardID).GetCounter(utils.TimeColumnMissing).Inc(1)
		return false, utils.StackError(nil, "Fact table's event time column (first column) is missing")
	}

	start := utils.Now()
	updateRecords, insertRecords, backfillUpsertBatch, err := shard.insertPrimaryKeys(primaryKeyColumns, eventTimeColumnIndex,
		redoLogFile, upsertBatch, skipBackfillRows)
	utils.GetReporter(shard.Schema.Schema.Name, shard.ShardID).GetTimer(utils.IngestionPrimaryKeyLookupTime).Record(utils.Now().Sub(start))

	if err != nil {
		return false, err
	}

	// We write insert records first so records with the same primary key in a upsert batch
	// will be updated in order.
	for batchID, records := range insertRecords {
		if err := shard.writeBatchRecords(columnDeletions, upsertBatch, batchID, records, false); err != nil {
			return false, err
		}
	}
	for batchID, records := range updateRecords {
		if err := shard.writeBatchRecords(columnDeletions, upsertBatch, batchID, records, true); err != nil {
			return false, err
		}
	}

	shard.LiveStore.AdvanceLastReadRecord()
	numMutations := len(insertRecords) + len(updateRecords)
	return shard.postUpsertBatchApplication(upsertBatch, backfillUpsertBatch, redoLogFile, offset, numMutations), nil
}