Method Commit

engine/batch.go:88–152

Commit The transaction commits, writes the transient data to the data file, and updates the in-memory index

()

// Commit The transaction commits, writes the transient data to the data file, and updates the in-memory index
func (wb *WriteBatch) Commit() error {
	wb.lock.Lock()
	defer wb.lock.Unlock()

	if len(wb.temporaryDataWrites) == 0 {
		return nil
	}
	if uint(len(wb.temporaryDataWrites)) > wb.options.MaxBatchNum {
		return _const.ErrExceedMaxBatchNum
	}

	// Gets the current, most recent transaction sequence number
	transSeq := atomic.AddUint64(&wb.db.transSeqNo, 1)

	// Start writing data to the data file
	// The index is not updated immediately after a single piece of data is written.
	// It needs to be stored temporarily
	positions := make(map[string]*data.LogRecordPst)
	for _, record := range wb.temporaryDataWrites {
		logRecordPst, err := wb.db.appendLogRecord(&data.LogRecord{
			Key:   encodeLogRecordKeyWithSeq(record.Key, transSeq),
			Value: record.Value,
			Type:  record.Type,
		})
		if err != nil {
			return err
		}
		positions[string(record.Key)] = logRecordPst
	}

	// Write a piece of data that identifies the completion of the transaction
	finishedRecord := &data.LogRecord{
		Key:  encodeLogRecordKeyWithSeq(lgrTransFinaKey, transSeq),
		Type: data.LogRecordTransFinished,
	}
	if _, err := wb.db.appendLogRecord(finishedRecord); err != nil {
		return err
	}

	// Decide whether to persist based on the configuration
	if wb.options.SyncWrites && wb.db.activeFile != nil {
		if err := wb.db.activeFile.Sync(); err != nil {
			return err
		}
	}

	// Update memory index
	for _, record := range wb.temporaryDataWrites {
		pst := positions[string(record.Key)]
		if record.Type == data.LogRecordNormal {
			// Put the record in the index if it is of type LogRecordNormal
			wb.db.index.Put(record.Key, pst)
		}
		if record.Type == data.LogRecordDeleted {
			// Delete the record from the index if it is of type LogRecordDeleted
			wb.db.index.Delete(record.Key)
		}
	}