backup.go
func (db *DB) Load(r io.Reader) error {
	br := bufio.NewReaderSize(r, 16<<10)
	unmarshalBuf := make([]byte, 1<<10)
	var entries []*Entry
	var wg sync.WaitGroup
	errChan := make(chan error, 1)

	// func to check for pending error before sending off a batch for writing
	batchSetAsyncIfNoErr := func(entries []*Entry) error {
		select {
		case err := <-errChan:
			return err
		default:
			wg.Add(1)
			return db.batchSetAsync(entries, func(err error) {
				defer wg.Done()
				if err != nil {
					select {
					case errChan <- err:
					default:
					}
				}
			})
		}
	}

	for {
		var sz uint64
		err := binary.Read(br, binary.LittleEndian, &sz)
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}

		if cap(unmarshalBuf) < int(sz) {
			unmarshalBuf = make([]byte, sz)
		}

		e := &pb.KV{}
		if _, err = io.ReadFull(br, unmarshalBuf[:sz]); err != nil {
			return err
		}
		if err = e.Unmarshal(unmarshalBuf[:sz]); err != nil {
			return err
		}
		var userMeta byte
		if len(e.UserMeta) > 0 {
			userMeta = e.UserMeta[0]
		}
		entries = append(entries, &Entry{
			Key:       y.KeyWithTs(e.Key, e.Version),
			Value:     e.Value,
			UserMeta:  userMeta,
			ExpiresAt: e.ExpiresAt,
			meta:      e.Meta[0],
		})
		// Update nextTxnTs, memtable stores this timestamp in badger head
		// when flushed.
		if e.Version >= db.orc.nextTxnTs {
			db.orc.nextTxnTs = e.Version + 1
		}

		if len(entries) == 1000 {
			if err := batchSetAsyncIfNoErr(entries); err != nil {
				return err
			}
			entries = make([]*Entry, 0, 1000)
		}
	}

	if len(entries) > 0 {
		if err := batchSetAsyncIfNoErr(entries); err != nil {
			return err
		}
	}
	wg.Wait()

	select {
	case err := <-errChan:
		return err
	default:
		// Mark all versions done up until nextTxnTs.
		db.orc.txnMark.Done(db.orc.nextTxnTs - 1)
		return nil
	}
}
func (db *DB) Load(r io.Reader) error {
	br := bufio.NewReaderSize(r, 16<<10)
	unmarshalBuf := make([]byte, 1<<10)
	var entries []*Entry
	var wg sync.WaitGroup
	errChan := make(chan error, 1)

	// func to check for pending error before sending off a batch for writing
	batchSetAsyncIfNoErr := func(entries []*Entry) error {
		select {
		case err := <-errChan:
			return err
		default:
			wg.Add(1)
			return db.batchSetAsync(entries, func(err error) {
				defer wg.Done()
				if err != nil {
					select {
					case errChan <- err:
					default:
					}
				}
			})
		}
	}

	for {
		var sz uint64
		err := binary.Read(br, binary.LittleEndian, &sz)
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}

		if cap(unmarshalBuf) < int(sz) {
			unmarshalBuf = make([]byte, sz)
		}

		e := &pb.KV{}
		if _, err = io.ReadFull(br, unmarshalBuf[:sz]); err != nil {
			return err
		}
		if err = e.Unmarshal(unmarshalBuf[:sz]); err != nil {
			return err
		}
		var userMeta byte
		if len(e.UserMeta) > 0 {
			userMeta = e.UserMeta[0]
		}
		entries = append(entries, &Entry{
			Key:       y.KeyWithTs(e.Key, e.Version),
			Value:     e.Value,
			UserMeta:  userMeta,
			ExpiresAt: e.ExpiresAt,
			meta:      e.Meta[0],
		})
		// Update nextTxnTs, memtable stores this timestamp in badger head
		// when flushed.
		if e.Version >= db.orc.nextTxnTs {
			db.orc.nextTxnTs = e.Version + 1
		}

		if len(entries) == 1000 {
			if err := batchSetAsyncIfNoErr(entries); err != nil {
				return err
			}
			entries = make([]*Entry, 0, 1000)
		}
	}

	if len(entries) > 0 {
		if err := batchSetAsyncIfNoErr(entries); err != nil {
			return err
		}
	}
	wg.Wait()

	select {
	case err := <-errChan:
		return err
	default:
		// Mark all versions done up until nextTxnTs.
		db.orc.txnMark.Done(db.orc.nextTxnTs - 1)
		return nil
	}
}
func ( db * DB ) Load ( r io . Reader ) error { br := bufio . NewReaderSize ( r , 16 << 10 ) unmarshalBuf := make ( [ ] byte , 1 << 10 ) var entries [ ] * Entry var wg sync . WaitGroup errChan := make ( chan error , 1 ) // func to check for pending error before sending off a batch for writing batchSetAsyncIfNoErr := func ( entries [ ] * Entry ) error { select { case err := <- errChan : return err default : wg . Add ( 1 ) return db . batchSetAsync ( entries , func ( err error ) { defer wg . Done ( ) if err != nil { select { case errChan <- err : default : } } } ) } } for { var sz uint64 err := binary . Read ( br , binary . LittleEndian , & sz ) if err == io . EOF { break } else if err != nil { return err } if cap ( unmarshalBuf ) < int ( sz ) { unmarshalBuf = make ( [ ] byte , sz ) } e := & pb . KV { } if _ , err = io . ReadFull ( br , unmarshalBuf [ : sz ] ) ; err != nil { return err } if err = e . Unmarshal ( unmarshalBuf [ : sz ] ) ; err != nil { return err } var userMeta byte if len ( e . UserMeta ) > 0 { userMeta = e . UserMeta [ 0 ] } entries = append ( entries , & Entry { Key : y . KeyWithTs ( e . Key , e . Version ) , Value : e . Value , UserMeta : userMeta , ExpiresAt : e . ExpiresAt , meta : e . Meta [ 0 ] , } ) // Update nextTxnTs, memtable stores this timestamp in badger head // when flushed. if e . Version >= db . orc . nextTxnTs { db . orc . nextTxnTs = e . Version + 1 } if len ( entries ) == 1000 { if err := batchSetAsyncIfNoErr ( entries ) ; err != nil { return err } entries = make ( [ ] * Entry , 0 , 1000 ) } } if len ( entries ) > 0 { if err := batchSetAsyncIfNoErr ( entries ) ; err != nil { return err } } wg . Wait ( ) select { case err := <- errChan : return err default : // Mark all versions done up until nextTxnTs. db . orc . txnMark . Done ( db . orc . nextTxnTs - 1 ) return nil } }
// Load reads a protobuf-encoded list of all entries from a reader and writes
// them to the database. This can be used to restore the database from a backup
// made by calling DB.Backup().
//
// DB.Load() should be called on a database that is not running any other
// concurrent transactions while it is running.
Load reads a protobuf - encoded list of all entries from a reader and writes them to the database . This can be used to restore the database from a backup made by calling DB . Backup () . DB . Load () should be called on a database that is not running any other concurrent transactions while it is running .