Method processReqCh

worker/restore_map.go:291–647

(ctx context.Context)

}

func (m *mapper) processReqCh(ctx context.Context) error {
	buf := z.NewBuffer(20<<20, "processKVList")
	defer func() {
		if err := buf.Release(); err != nil {
			glog.Warningf("error in releasing buffer: %v", err)
		}
	}()

	maxNs := uint64(0)
	maxUid := uint64(0)

	toBuffer := func(kv *bpb.KV, version uint64) error {
		key := y.KeyWithTs(kv.Key, version)
		sz := proto.Size(kv)
		b := buf.SliceAllocate(2 + len(key) + sz)

		binary.BigEndian.PutUint16(b[0:2], uint16(len(key)))
		x.AssertTrue(copy(b[2:], key) == len(key))
		_, err := x.MarshalToSizedBuffer(b[2+len(key):], kv)
		return err
	}

	processKV := func(in *loadBackupInput, kv *bpb.KV) error {
		if len(kv.GetUserMeta()) != 1 {
			return errors.Errorf(
				"Unexpected meta: %v for key: %s", kv.UserMeta, hex.Dump(kv.Key))
		}

		restoreKey, ns, err := fromBackupKey(kv.Key)
		if err != nil {
			return errors.Wrap(err, "fromBackupKey")
		}

		// we will not process further if restore is namespace aware restore and namespace
		// of current key is other than fromNamespace.
		if in.isNamespaceAwareRestore && ns != in.fromNamespace {
			return nil
		}

		// Filter keys using the preds set. Do not do this filtering for type keys
		// as they are meant to be in every group and their Attr value does not
		// match a predicate name.
		parsedKey, err := x.Parse(restoreKey)
		if err != nil {
			return errors.Wrapf(err, "could not parse key %s", hex.Dump(restoreKey))
		}

		// Update the local max uid and max namespace values.
		maxUid = x.Max(maxUid, parsedKey.Uid)
		if in.isNamespaceAwareRestore {
			maxNs = x.Max(maxNs, 0)
		} else {
			maxNs = x.Max(maxNs, ns)
		}

		if !in.keepSchema && (parsedKey.IsSchema() || parsedKey.IsType()) {
			return nil
		}