/brz/remove-bazaar

# Copyright (C) 2006-2010 Canonical Ltd

#

# Authors:

#   Johan Rydberg <jrydberg@gnu.org>

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Versioned text file storage api."""

from copy import copy

from cStringIO import StringIO

import os

import struct

from zlib import adler32

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

import urllib

from bzrlib import (

    annotate,

    errors,

    graph as _mod_graph,

    groupcompress,

    index,

    knit,

    osutils,

    multiparent,

    tsort,

    revision,

    ui,

    )

from bzrlib.graph import DictParentsProvider, Graph, StackedParentsProvider

from bzrlib.transport.memory import MemoryTransport

""")

from bzrlib.registry import Registry

from bzrlib.symbol_versioning import *

from bzrlib.textmerge import TextMerge

from bzrlib import bencode

adapter_registry = Registry()

adapter_registry.register_lazy(('knit-delta-gz', 'fulltext'), 'bzrlib.knit',

    'DeltaPlainToFullText')

adapter_registry.register_lazy(('knit-ft-gz', 'fulltext'), 'bzrlib.knit',

    'FTPlainToFullText')

adapter_registry.register_lazy(('knit-annotated-delta-gz', 'knit-delta-gz'),

    'bzrlib.knit', 'DeltaAnnotatedToUnannotated')

adapter_registry.register_lazy(('knit-annotated-delta-gz', 'fulltext'),

    'bzrlib.knit', 'DeltaAnnotatedToFullText')

adapter_registry.register_lazy(('knit-annotated-ft-gz', 'knit-ft-gz'),

    'bzrlib.knit', 'FTAnnotatedToUnannotated')

adapter_registry.register_lazy(('knit-annotated-ft-gz', 'fulltext'),

    'bzrlib.knit', 'FTAnnotatedToFullText')

# adapter_registry.register_lazy(('knit-annotated-ft-gz', 'chunked'),

#     'bzrlib.knit', 'FTAnnotatedToChunked')

class ContentFactory(object):

    """Abstract interface for insertion and retrieval from a VersionedFile.

    :ivar sha1: None, or the sha1 of the content fulltext.

    :ivar storage_kind: The native storage kind of this factory. One of

        'mpdiff', 'knit-annotated-ft', 'knit-annotated-delta', 'knit-ft',

        'knit-delta', 'fulltext', 'knit-annotated-ft-gz',

        'knit-annotated-delta-gz', 'knit-ft-gz', 'knit-delta-gz'.

    :ivar key: The key of this content. Each key is a tuple with a single

        string in it.

    :ivar parents: A tuple of parent keys for self.key. If the object has

        no parent information, None (as opposed to () for an empty list of

        parents).

    """

    def __init__(self):

        """Create a ContentFactory."""

        self.sha1 = None

        self.storage_kind = None

        self.key = None

        self.parents = None

class ChunkedContentFactory(ContentFactory):

    """Static data content factory.

    This takes a 'chunked' list of strings. The only requirement on 'chunked' is

    that ''.join(lines) becomes a valid fulltext. A tuple of a single string

    satisfies this, as does a list of lines.

    :ivar sha1: None, or the sha1 of the content fulltext.

    :ivar storage_kind: The native storage kind of this factory. Always

        'chunked'

    :ivar key: The key of this content. Each key is a tuple with a single

        string in it.

    :ivar parents: A tuple of parent keys for self.key. If the object has

        no parent information, None (as opposed to () for an empty list of

        parents).

     """

    def __init__(self, key, parents, sha1, chunks):

        """Create a ContentFactory."""

        self.sha1 = sha1

        self.storage_kind = 'chunked'

        self.key = key

        self.parents = parents

        self._chunks = chunks

    def get_bytes_as(self, storage_kind):

        if storage_kind == 'chunked':

            return self._chunks

        elif storage_kind == 'fulltext':

            return ''.join(self._chunks)

        raise errors.UnavailableRepresentation(self.key, storage_kind,

            self.storage_kind)

class FulltextContentFactory(ContentFactory):

    """Static data content factory.

    This takes a fulltext when created and just returns that during

    get_bytes_as('fulltext').

    :ivar sha1: None, or the sha1 of the content fulltext.

    :ivar storage_kind: The native storage kind of this factory. Always

        'fulltext'.

    :ivar key: The key of this content. Each key is a tuple with a single

        string in it.

    :ivar parents: A tuple of parent keys for self.key. If the object has

        no parent information, None (as opposed to () for an empty list of

        parents).

     """

    def __init__(self, key, parents, sha1, text):

        """Create a ContentFactory."""

        self.sha1 = sha1

        self.storage_kind = 'fulltext'

        self.key = key

        self.parents = parents

        self._text = text

    def get_bytes_as(self, storage_kind):

        if storage_kind == self.storage_kind:

            return self._text

        elif storage_kind == 'chunked':

            return [self._text]

        raise errors.UnavailableRepresentation(self.key, storage_kind,

            self.storage_kind)

class AbsentContentFactory(ContentFactory):

    """A placeholder content factory for unavailable texts.

    :ivar sha1: None.

    :ivar storage_kind: 'absent'.

    :ivar key: The key of this content. Each key is a tuple with a single

        string in it.

    :ivar parents: None.

    """

    def __init__(self, key):

        """Create a ContentFactory."""

        self.sha1 = None

        self.storage_kind = 'absent'

        self.key = key

        self.parents = None

    def get_bytes_as(self, storage_kind):

        raise ValueError('A request was made for key: %s, but that'

                         ' content is not available, and the calling'

                         ' code does not handle if it is missing.'

                         % (self.key,))

class AdapterFactory(ContentFactory):

    """A content factory to adapt between key prefix's."""

    def __init__(self, key, parents, adapted):

        """Create an adapter factory instance."""

        self.key = key

        self.parents = parents

        self._adapted = adapted

    def __getattr__(self, attr):

        """Return a member from the adapted object."""

        if attr in ('key', 'parents'):

            return self.__dict__[attr]

        else:

            return getattr(self._adapted, attr)

def filter_absent(record_stream):

    """Adapt a record stream to remove absent records."""

    for record in record_stream:

        if record.storage_kind != 'absent':

            yield record

class _MPDiffGenerator(object):

    """Pull out the functionality for generating mp_diffs."""

    def __init__(self, vf, keys):

        self.vf = vf

        # This is the order the keys were requested in

        self.ordered_keys = tuple(keys)

        # keys + their parents, what we need to compute the diffs

        self.needed_keys = ()

        # Map from key: mp_diff

        self.diffs = {}

        # Map from key: parents_needed (may have ghosts)

        self.parent_map = {}

        # Parents that aren't present

        self.ghost_parents = ()

        # Map from parent_key => number of children for this text

        self.refcounts = {}

        # Content chunks that are cached while we still need them

        self.chunks = {}

    def _find_needed_keys(self):

        """Find the set of keys we need to request.

        This includes all the original keys passed in, and the non-ghost

        parents of those keys.

        :return: (needed_keys, refcounts)

            needed_keys is the set of all texts we need to extract

            refcounts is a dict of {key: num_children} letting us know when we

                no longer need to cache a given parent text

        """

        # All the keys and their parents

        needed_keys = set(self.ordered_keys)

        parent_map = self.vf.get_parent_map(needed_keys)

        self.parent_map = parent_map

        # TODO: Should we be using a different construct here? I think this

        #       uses difference_update internally, and we expect the result to

        #       be tiny

        missing_keys = needed_keys.difference(parent_map)

        if missing_keys:

            raise errors.RevisionNotPresent(list(missing_keys)[0], self.vf)

        # Parents that might be missing. They are allowed to be ghosts, but we

        # should check for them

        refcounts = {}

        setdefault = refcounts.setdefault

        maybe_ghosts = set()

        for child_key, parent_keys in parent_map.iteritems():

            if not parent_keys:

                # Ghost? We should never get here

                continue

            maybe_ghosts.update(p for p in parent_keys if p not in needed_keys)

            needed_keys.update(parent_keys)

            for p in parent_keys:

                refcounts[p] = setdefault(p, 0) + 1

        # Remove any parents that are actually ghosts

        self.ghost_parents = maybe_ghosts.difference(

                                self.vf.get_parent_map(maybe_ghosts))

        needed_keys.difference_update(self.ghost_parents)

        self.needed_keys = needed_keys

        self.refcounts = refcounts

        return needed_keys, refcounts

    def _compute_diff(self, key, parent_lines, lines):

        """Compute a single mp_diff, and store it in self._diffs"""

        if len(parent_lines) > 0:

            # XXX: _extract_blocks is not usefully defined anywhere...

            #      It was meant to extract the left-parent diff without

            #      having to recompute it for Knit content (pack-0.92,

            #      etc). That seems to have regressed somewhere

            left_parent_blocks = self.vf._extract_blocks(key,

                parent_lines[0], lines)

        else:

            left_parent_blocks = None

        diff = multiparent.MultiParent.from_lines(lines,

                    parent_lines, left_parent_blocks)

        self.diffs[key] = diff

    def _process_one_record(self, record):

        this_chunks = record.get_bytes_as('chunked')

        if record.key in self.parent_map:

            # This record should be ready to diff, since we requested

            # content in 'topological' order

            parent_keys = self.parent_map.pop(record.key)

            parent_lines = []

            for p in parent_keys:

                # Alternatively we could check p not in self.needed_keys, but

                # ghost_parents should be tiny versus huge

                if p in self.ghost_parents:

                    continue

                refcount = self.refcounts[p]

                if refcount == 1: # Last child reference

                    self.refcounts.pop(p)

                    parent_chunks = self.chunks.pop(p)

                else:

                    self.refcounts[p] = refcount - 1

                    parent_chunks = self.chunks[p]

                # TODO: Should we cache the line form? We did the

                #       computation to get it, but storing it this way will

                #       be less memory efficient...

                parent_lines.append(osutils.chunks_to_lines(parent_chunks))

            lines = osutils.chunks_to_lines(this_chunks)

            # TODO: Should we be caching lines instead of chunks?

            #       Higher-memory, but avoids double extracting.

            #       If we have good topological sorting, we shouldn't have

            #       much pending stuff cached...

            ## this_chunks = lines

            self._compute_diff(record.key, parent_lines, lines)

        # Is this content required for any more children?

        if record.key in self.refcounts:

            self.chunks[record.key] = this_chunks

    def _extract_diffs(self):

        needed_keys, refcounts = self._find_needed_keys()

        for record in self.vf.get_record_stream(needed_keys,

                                                'topological', True):

            if record.storage_kind == 'absent':

                raise errors.RevisionNotPresent(record.key, self.vf)

            self._process_one_record(record)

        # At this point, we should have *no* remaining content

        assert not self.parent_map

        assert set(self.refcounts) == self.ghost_parents

    def compute_diffs(self):

        self._extract_diffs()

        dpop = self.diffs.pop

        return [dpop(k) for k in self.ordered_keys]

class VersionedFile(object):

    """Versioned text file storage.

    A versioned file manages versions of line-based text files,

    keeping track of the originating version for each line.

    To clients the "lines" of the file are represented as a list of

    strings. These strings will typically have terminal newline

    characters, but this is not required.  In particular files commonly

    do not have a newline at the end of the file.

    Texts are identified by a version-id string.

    """

    @staticmethod

    def check_not_reserved_id(version_id):

        revision.check_not_reserved_id(version_id)

    def copy_to(self, name, transport):

        """Copy this versioned file to name on transport."""

        raise NotImplementedError(self.copy_to)

    def get_record_stream(self, versions, ordering, include_delta_closure):

        """Get a stream of records for versions.

        :param versions: The versions to include. Each version is a tuple

            (version,).

        :param ordering: Either 'unordered' or 'topological'. A topologically

            sorted stream has compression parents strictly before their

            children.

        :param include_delta_closure: If True then the closure across any

            compression parents will be included (in the data content of the

            stream, not in the emitted records). This guarantees that

            'fulltext' can be used successfully on every record.

        :return: An iterator of ContentFactory objects, each of which is only

            valid until the iterator is advanced.

        """

        raise NotImplementedError(self.get_record_stream)

    def has_version(self, version_id):

        """Returns whether version is present."""

        raise NotImplementedError(self.has_version)

    def insert_record_stream(self, stream):

        """Insert a record stream into this versioned file.

        :param stream: A stream of records to insert.

        :return: None

        :seealso VersionedFile.get_record_stream:

        """

        raise NotImplementedError

    def add_lines(self, version_id, parents, lines, parent_texts=None,

        left_matching_blocks=None, nostore_sha=None, random_id=False,

        check_content=True):

        """Add a single text on top of the versioned file.

        Must raise RevisionAlreadyPresent if the new version is

        already present in file history.

        Must raise RevisionNotPresent if any of the given parents are

        not present in file history.

        :param lines: A list of lines. Each line must be a bytestring. And all

            of them except the last must be terminated with \n and contain no

            other \n's. The last line may either contain no \n's or a single

            terminated \n. If the lines list does meet this constraint the add

            routine may error or may succeed - but you will be unable to read

            the data back accurately. (Checking the lines have been split

            correctly is expensive and extremely unlikely to catch bugs so it

            is not done at runtime unless check_content is True.)

        :param parent_texts: An optional dictionary containing the opaque

            representations of some or all of the parents of version_id to

            allow delta optimisations.  VERY IMPORTANT: the texts must be those

            returned by add_lines or data corruption can be caused.

        :param left_matching_blocks: a hint about which areas are common

            between the text and its left-hand-parent.  The format is

            the SequenceMatcher.get_matching_blocks format.

        :param nostore_sha: Raise ExistingContent and do not add the lines to

            the versioned file if the digest of the lines matches this.

        :param random_id: If True a random id has been selected rather than

            an id determined by some deterministic process such as a converter

            from a foreign VCS. When True the backend may choose not to check

            for uniqueness of the resulting key within the versioned file, so

            this should only be done when the result is expected to be unique

            anyway.

        :param check_content: If True, the lines supplied are verified to be

            bytestrings that are correctly formed lines.

        :return: The text sha1, the number of bytes in the text, and an opaque

                 representation of the inserted version which can be provided

                 back to future add_lines calls in the parent_texts dictionary.

        """

        self._check_write_ok()

        return self._add_lines(version_id, parents, lines, parent_texts,

            left_matching_blocks, nostore_sha, random_id, check_content)

    def _add_lines(self, version_id, parents, lines, parent_texts,

        left_matching_blocks, nostore_sha, random_id, check_content):

        """Helper to do the class specific add_lines."""

        raise NotImplementedError(self.add_lines)

    def add_lines_with_ghosts(self, version_id, parents, lines,

        parent_texts=None, nostore_sha=None, random_id=False,

        check_content=True, left_matching_blocks=None):

        """Add lines to the versioned file, allowing ghosts to be present.

        This takes the same parameters as add_lines and returns the same.

        """

        self._check_write_ok()

        return self._add_lines_with_ghosts(version_id, parents, lines,

            parent_texts, nostore_sha, random_id, check_content, left_matching_blocks)

    def _add_lines_with_ghosts(self, version_id, parents, lines, parent_texts,

        nostore_sha, random_id, check_content, left_matching_blocks):

        """Helper to do class specific add_lines_with_ghosts."""

        raise NotImplementedError(self.add_lines_with_ghosts)

    def check(self, progress_bar=None):

        """Check the versioned file for integrity."""

        raise NotImplementedError(self.check)

    def _check_lines_not_unicode(self, lines):

        """Check that lines being added to a versioned file are not unicode."""

        for line in lines:

            if line.__class__ is not str:

                raise errors.BzrBadParameterUnicode("lines")

    def _check_lines_are_lines(self, lines):

        """Check that the lines really are full lines without inline EOL."""

        for line in lines:

            if '\n' in line[:-1]:

                raise errors.BzrBadParameterContainsNewline("lines")

    def get_format_signature(self):

        """Get a text description of the data encoding in this file.

        :since: 0.90

        """

        raise NotImplementedError(self.get_format_signature)

    def make_mpdiffs(self, version_ids):

        """Create multiparent diffs for specified versions."""

        knit_versions = set()

        knit_versions.update(version_ids)

        parent_map = self.get_parent_map(version_ids)

        for version_id in version_ids:

            try:

                knit_versions.update(parent_map[version_id])

            except KeyError:

                raise errors.RevisionNotPresent(version_id, self)

        # We need to filter out ghosts, because we can't diff against them.

        knit_versions = set(self.get_parent_map(knit_versions).keys())

        lines = dict(zip(knit_versions,

            self._get_lf_split_line_list(knit_versions)))

        diffs = []

        for version_id in version_ids:

            target = lines[version_id]

            try:

                parents = [lines[p] for p in parent_map[version_id] if p in

                    knit_versions]

            except KeyError:

                # I don't know how this could ever trigger.

                # parent_map[version_id] was already triggered in the previous

                # for loop, and lines[p] has the 'if p in knit_versions' check,

                # so we again won't have a KeyError.

                raise errors.RevisionNotPresent(version_id, self)

            if len(parents) > 0:

                left_parent_blocks = self._extract_blocks(version_id,

                                                          parents[0], target)

            else:

                left_parent_blocks = None

            diffs.append(multiparent.MultiParent.from_lines(target, parents,

                         left_parent_blocks))

        return diffs

    def _extract_blocks(self, version_id, source, target):

        return None

    def add_mpdiffs(self, records):

        """Add mpdiffs to this VersionedFile.

        Records should be iterables of version, parents, expected_sha1,

        mpdiff. mpdiff should be a MultiParent instance.

        """

        # Does this need to call self._check_write_ok()? (IanC 20070919)

        vf_parents = {}

        mpvf = multiparent.MultiMemoryVersionedFile()

        versions = []

        for version, parent_ids, expected_sha1, mpdiff in records:

            versions.append(version)

            mpvf.add_diff(mpdiff, version, parent_ids)

        needed_parents = set()

        for version, parent_ids, expected_sha1, mpdiff in records:

            needed_parents.update(p for p in parent_ids

                                  if not mpvf.has_version(p))

        present_parents = set(self.get_parent_map(needed_parents).keys())

        for parent_id, lines in zip(present_parents,

                                 self._get_lf_split_line_list(present_parents)):

            mpvf.add_version(lines, parent_id, [])

        for (version, parent_ids, expected_sha1, mpdiff), lines in\

            zip(records, mpvf.get_line_list(versions)):

            if len(parent_ids) == 1:

                left_matching_blocks = list(mpdiff.get_matching_blocks(0,

                    mpvf.get_diff(parent_ids[0]).num_lines()))

            else:

                left_matching_blocks = None

            try:

                _, _, version_text = self.add_lines_with_ghosts(version,

                    parent_ids, lines, vf_parents,

                    left_matching_blocks=left_matching_blocks)

            except NotImplementedError:

                # The vf can't handle ghosts, so add lines normally, which will

                # (reasonably) fail if there are ghosts in the data.

                _, _, version_text = self.add_lines(version,

                    parent_ids, lines, vf_parents,

                    left_matching_blocks=left_matching_blocks)

            vf_parents[version] = version_text

        sha1s = self.get_sha1s(versions)

        for version, parent_ids, expected_sha1, mpdiff in records:

            if expected_sha1 != sha1s[version]:

                raise errors.VersionedFileInvalidChecksum(version)

    def get_text(self, version_id):

        """Return version contents as a text string.

        Raises RevisionNotPresent if version is not present in

        file history.

        """

        return ''.join(self.get_lines(version_id))

    get_string = get_text

    def get_texts(self, version_ids):

        """Return the texts of listed versions as a list of strings.

        Raises RevisionNotPresent if version is not present in

        file history.

        """

        return [''.join(self.get_lines(v)) for v in version_ids]

    def get_lines(self, version_id):

        """Return version contents as a sequence of lines.

        Raises RevisionNotPresent if version is not present in

        file history.

        """

        raise NotImplementedError(self.get_lines)

    def _get_lf_split_line_list(self, version_ids):

        return [StringIO(t).readlines() for t in self.get_texts(version_ids)]

    def get_ancestry(self, version_ids, topo_sorted=True):

        """Return a list of all ancestors of given version(s). This

        will not include the null revision.

        This list will not be topologically sorted if topo_sorted=False is

        passed.

        Must raise RevisionNotPresent if any of the given versions are

        not present in file history."""

        if isinstance(version_ids, basestring):

            version_ids = [version_ids]

        raise NotImplementedError(self.get_ancestry)

    def get_ancestry_with_ghosts(self, version_ids):

        """Return a list of all ancestors of given version(s). This

        will not include the null revision.

        Must raise RevisionNotPresent if any of the given versions are

        not present in file history.

        Ghosts that are known about will be included in ancestry list,

        but are not explicitly marked.

        """

        raise NotImplementedError(self.get_ancestry_with_ghosts)

    def get_parent_map(self, version_ids):

        """Get a map of the parents of version_ids.

        :param version_ids: The version ids to look up parents for.

        :return: A mapping from version id to parents.

        """

        raise NotImplementedError(self.get_parent_map)

    def get_parents_with_ghosts(self, version_id):

        """Return version names for parents of version_id.

        Will raise RevisionNotPresent if version_id is not present

        in the history.

        Ghosts that are known about will be included in the parent list,

        but are not explicitly marked.

        """

        try:

            return list(self.get_parent_map([version_id])[version_id])

        except KeyError:

            raise errors.RevisionNotPresent(version_id, self)

    def annotate(self, version_id):

        """Return a list of (version-id, line) tuples for version_id.

        :raise RevisionNotPresent: If the given version is

        not present in file history.

        """

        raise NotImplementedError(self.annotate)

    def iter_lines_added_or_present_in_versions(self, version_ids=None,

                                                pb=None):

        """Iterate over the lines in the versioned file from version_ids.

        This may return lines from other versions. Each item the returned

        iterator yields is a tuple of a line and a text version that that line

        is present in (not introduced in).

        Ordering of results is in whatever order is most suitable for the

        underlying storage format.

        If a progress bar is supplied, it may be used to indicate progress.

        The caller is responsible for cleaning up progress bars (because this

        is an iterator).

        NOTES: Lines are normalised: they will all have \n terminators.

               Lines are returned in arbitrary order.

        :return: An iterator over (line, version_id).

        """

        raise NotImplementedError(self.iter_lines_added_or_present_in_versions)

    def plan_merge(self, ver_a, ver_b):

        """Return pseudo-annotation indicating how the two versions merge.

        This is computed between versions a and b and their common

        base.

        Weave lines present in none of them are skipped entirely.

        Legend:

        killed-base Dead in base revision

        killed-both Killed in each revision

        killed-a    Killed in a

        killed-b    Killed in b

        unchanged   Alive in both a and b (possibly created in both)

        new-a       Created in a

        new-b       Created in b

        ghost-a     Killed in a, unborn in b

        ghost-b     Killed in b, unborn in a

        irrelevant  Not in either revision

        """

        raise NotImplementedError(VersionedFile.plan_merge)

    def weave_merge(self, plan, a_marker=TextMerge.A_MARKER,

                    b_marker=TextMerge.B_MARKER):

        return PlanWeaveMerge(plan, a_marker, b_marker).merge_lines()[0]

class RecordingVersionedFilesDecorator(object):

    """A minimal versioned files that records calls made on it.

    Only enough methods have been added to support tests using it to date.

    :ivar calls: A list of the calls made; can be reset at any time by

        assigning [] to it.

    """

    def __init__(self, backing_vf):

        """Create a RecordingVersionedFilesDecorator decorating backing_vf.

        :param backing_vf: The versioned file to answer all methods.

        """

        self._backing_vf = backing_vf

        self.calls = []

    def add_lines(self, key, parents, lines, parent_texts=None,

        left_matching_blocks=None, nostore_sha=None, random_id=False,

        check_content=True):

        self.calls.append(("add_lines", key, parents, lines, parent_texts,

            left_matching_blocks, nostore_sha, random_id, check_content))

        return self._backing_vf.add_lines(key, parents, lines, parent_texts,

            left_matching_blocks, nostore_sha, random_id, check_content)

    def check(self):

        self._backing_vf.check()

    def get_parent_map(self, keys):

        self.calls.append(("get_parent_map", copy(keys)))

        return self._backing_vf.get_parent_map(keys)

    def get_record_stream(self, keys, sort_order, include_delta_closure):

        self.calls.append(("get_record_stream", list(keys), sort_order,

            include_delta_closure))

        return self._backing_vf.get_record_stream(keys, sort_order,