Method parseDataFiles

db.go:532–650

(dataFileIds []int64)

}

func (db *DB) parseDataFiles(dataFileIds []int64) (err error) {
	var (
		off      int64
		f        *fileRecovery
		fID      int64
		dataInTx core.DataInTx
	)

	parseDataInTx := func() error {
		for _, entry := range dataInTx.Es {
			// if this bucket is not existed in bucket manager right now
			// its because it already deleted in the feature WAL utils.GetLogger().
			// so we can just ignore here.
			bucketId := entry.Meta.BucketId
			if _, err := db.bucketMgr.GetBucketById(bucketId); errors.Is(err, ErrBucketNotExist) {
				continue
			}

			record := db.createRecordByModeWithFidAndOff(entry.Fid, uint64(entry.Off), &entry.Entry)

			if err = db.buildIdxes(record, &entry.Entry); err != nil {
				return err
			}

			db.KeyCount++

		}
		return nil
	}

	readEntriesFromFile := func() error {
		for {
			entry, err := f.readEntry(off)
			if err != nil {
				// whatever which logic branch it will choose, we will release the fd.
				_ = f.release()
				if errors.Is(err, io.EOF) || errors.Is(err, ErrIndexOutOfBound) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, ErrEntryZero) || errors.Is(err, core.ErrHeaderSizeOutOfBounds) {
					break
				}
				if off >= db.opt.SegmentSize {
					break
				}

				return err
			}

			if entry == nil {
				break
			}

			entryWhenRecovery := &core.EntryWhenRecovery{
				Entry: *entry,
				Fid:   fID,
				Off:   off,
			}
			if dataInTx.TxId == 0 {
				dataInTx.AppendEntry(entryWhenRecovery)
				dataInTx.TxId = entry.Meta.TxID