~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd

#

# 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

"""Knit versionedfile implementation.

A knit is a versioned file implementation that supports efficient append only

updates.

Knit file layout:

lifeless: the data file is made up of "delta records".  each delta record has a delta header

that contains; (1) a version id, (2) the size of the delta (in lines), and (3)  the digest of

the -expanded data- (ie, the delta applied to the parent).  the delta also ends with a

end-marker; simply "end VERSION"

delta can be line or full contents.a

... the 8's there are the index number of the annotation.

version robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad 7 c7d23b2a5bd6ca00e8e266cec0ec228158ee9f9e

59,59,3

8

8         if ie.executable:

8             e.set('executable', 'yes')

130,130,2

8         if elt.get('executable') == 'yes':

8             ie.executable = True

end robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad

whats in an index:

09:33 < jrydberg> lifeless: each index is made up of a tuple of; version id, options, position, size, parents

09:33 < jrydberg> lifeless: the parents are currently dictionary compressed

09:33 < jrydberg> lifeless: (meaning it currently does not support ghosts)

09:33 < lifeless> right

09:33 < jrydberg> lifeless: the position and size is the range in the data file

so the index sequence is the dictionary compressed sequence number used

in the deltas to provide line annotation

"""

from cStringIO import StringIO

from itertools import izip

import operator

import os

import sys

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

from bzrlib import (

    annotate,

    debug,

    diff,

    graph as _mod_graph,

    index as _mod_index,

    lru_cache,

    pack,

    progress,

    trace,

    tsort,

    tuned_gzip,

    )

""")

from bzrlib import (

    errors,

    osutils,

    patiencediff,

    )

from bzrlib.errors import (

    FileExists,

    NoSuchFile,

    KnitError,

    InvalidRevisionId,

    KnitCorrupt,

    KnitHeaderError,

    RevisionNotPresent,

    RevisionAlreadyPresent,

    SHA1KnitCorrupt,

    )

from bzrlib.osutils import (

    contains_whitespace,

    contains_linebreaks,

    sha_string,

    sha_strings,

    split_lines,

    )

from bzrlib.versionedfile import (

    AbsentContentFactory,

    adapter_registry,

    ConstantMapper,

    ContentFactory,

    ChunkedContentFactory,

    sort_groupcompress,

    VersionedFile,

    VersionedFiles,

    )

# TODO: Split out code specific to this format into an associated object.

# TODO: Can we put in some kind of value to check that the index and data

# files belong together?

# TODO: accommodate binaries, perhaps by storing a byte count

# TODO: function to check whole file

# TODO: atomically append data, then measure backwards from the cursor

# position after writing to work out where it was located.  we may need to

# bypass python file buffering.

DATA_SUFFIX = '.knit'

INDEX_SUFFIX = '.kndx'

_STREAM_MIN_BUFFER_SIZE = 5*1024*1024

class KnitAdapter(object):

    """Base class for knit record adaption."""

    def __init__(self, basis_vf):

        """Create an adapter which accesses full texts from basis_vf.

        :param basis_vf: A versioned file to access basis texts of deltas from.

            May be None for adapters that do not need to access basis texts.

        """

        self._data = KnitVersionedFiles(None, None)

        self._annotate_factory = KnitAnnotateFactory()

        self._plain_factory = KnitPlainFactory()

        self._basis_vf = basis_vf

class FTAnnotatedToUnannotated(KnitAdapter):

    """An adapter from FT annotated knits to unannotated ones."""

    def get_bytes(self, factory):

        annotated_compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(annotated_compressed_bytes)

        content = self._annotate_factory.parse_fulltext(contents, rec[1])

        size, bytes = self._data._record_to_data((rec[1],), rec[3], content.text())

        return bytes

class DeltaAnnotatedToUnannotated(KnitAdapter):

    """An adapter for deltas from annotated to unannotated."""

    def get_bytes(self, factory):

        annotated_compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(annotated_compressed_bytes)

        delta = self._annotate_factory.parse_line_delta(contents, rec[1],

            plain=True)

        contents = self._plain_factory.lower_line_delta(delta)

        size, bytes = self._data._record_to_data((rec[1],), rec[3], contents)

        return bytes

class FTAnnotatedToFullText(KnitAdapter):

    """An adapter from FT annotated knits to unannotated ones."""

    def get_bytes(self, factory):

        annotated_compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(annotated_compressed_bytes)

        content, delta = self._annotate_factory.parse_record(factory.key[-1],

            contents, factory._build_details, None)

        return ''.join(content.text())

class DeltaAnnotatedToFullText(KnitAdapter):

    """An adapter for deltas from annotated to unannotated."""

    def get_bytes(self, factory):

        annotated_compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(annotated_compressed_bytes)

        delta = self._annotate_factory.parse_line_delta(contents, rec[1],

            plain=True)

        compression_parent = factory.parents[0]

        basis_entry = self._basis_vf.get_record_stream(

            [compression_parent], 'unordered', True).next()

        if basis_entry.storage_kind == 'absent':

            raise errors.RevisionNotPresent(compression_parent, self._basis_vf)

        basis_chunks = basis_entry.get_bytes_as('chunked')

        basis_lines = osutils.chunks_to_lines(basis_chunks)

        # Manually apply the delta because we have one annotated content and

        # one plain.

        basis_content = PlainKnitContent(basis_lines, compression_parent)

        basis_content.apply_delta(delta, rec[1])

        basis_content._should_strip_eol = factory._build_details[1]

        return ''.join(basis_content.text())

class FTPlainToFullText(KnitAdapter):

    """An adapter from FT plain knits to unannotated ones."""

    def get_bytes(self, factory):

        compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(compressed_bytes)

        content, delta = self._plain_factory.parse_record(factory.key[-1],

            contents, factory._build_details, None)

        return ''.join(content.text())

class DeltaPlainToFullText(KnitAdapter):

    """An adapter for deltas from annotated to unannotated."""

    def get_bytes(self, factory):

        compressed_bytes = factory._raw_record

        rec, contents = \

            self._data._parse_record_unchecked(compressed_bytes)

        delta = self._plain_factory.parse_line_delta(contents, rec[1])

        compression_parent = factory.parents[0]

        # XXX: string splitting overhead.

        basis_entry = self._basis_vf.get_record_stream(

            [compression_parent], 'unordered', True).next()

        if basis_entry.storage_kind == 'absent':

            raise errors.RevisionNotPresent(compression_parent, self._basis_vf)

        basis_chunks = basis_entry.get_bytes_as('chunked')

        basis_lines = osutils.chunks_to_lines(basis_chunks)

        basis_content = PlainKnitContent(basis_lines, compression_parent)

        # Manually apply the delta because we have one annotated content and

        # one plain.

        content, _ = self._plain_factory.parse_record(rec[1], contents,

            factory._build_details, basis_content)

        return ''.join(content.text())

class KnitContentFactory(ContentFactory):

    """Content factory for streaming from knits.

    :seealso ContentFactory:

    """

    def __init__(self, key, parents, build_details, sha1, raw_record,

        annotated, knit=None, network_bytes=None):

        """Create a KnitContentFactory for key.

        :param key: The key.

        :param parents: The parents.

        :param build_details: The build details as returned from

            get_build_details.

        :param sha1: The sha1 expected from the full text of this object.

        :param raw_record: The bytes of the knit data from disk.

        :param annotated: True if the raw data is annotated.

        :param network_bytes: None to calculate the network bytes on demand,

            not-none if they are already known.

        """

        ContentFactory.__init__(self)

        self.sha1 = sha1

        self.key = key

        self.parents = parents

        if build_details[0] == 'line-delta':

            kind = 'delta'

        else:

            kind = 'ft'

        if annotated:

            annotated_kind = 'annotated-'

        else:

            annotated_kind = ''

        self.storage_kind = 'knit-%s%s-gz' % (annotated_kind, kind)

        self._raw_record = raw_record

        self._network_bytes = network_bytes

        self._build_details = build_details

        self._knit = knit

    def _create_network_bytes(self):

        """Create a fully serialised network version for transmission."""

        # storage_kind, key, parents, Noeol, raw_record

        key_bytes = '\x00'.join(self.key)

        if self.parents is None:

            parent_bytes = 'None:'

        else:

            parent_bytes = '\t'.join('\x00'.join(key) for key in self.parents)

        if self._build_details[1]:

            noeol = 'N'

        else:

            noeol = ' '

        network_bytes = "%s\n%s\n%s\n%s%s" % (self.storage_kind, key_bytes,

            parent_bytes, noeol, self._raw_record)

        self._network_bytes = network_bytes

    def get_bytes_as(self, storage_kind):

        if storage_kind == self.storage_kind:

            if self._network_bytes is None:

                self._create_network_bytes()

            return self._network_bytes

        if ('-ft-' in self.storage_kind and

            storage_kind in ('chunked', 'fulltext')):

            adapter_key = (self.storage_kind, 'fulltext')

            adapter_factory = adapter_registry.get(adapter_key)

            adapter = adapter_factory(None)

            bytes = adapter.get_bytes(self)

            if storage_kind == 'chunked':

                return [bytes]

            else:

                return bytes

        if self._knit is not None:

            # Not redundant with direct conversion above - that only handles

            # fulltext cases.

            if storage_kind == 'chunked':

                return self._knit.get_lines(self.key[0])

            elif storage_kind == 'fulltext':

                return self._knit.get_text(self.key[0])

        raise errors.UnavailableRepresentation(self.key, storage_kind,

            self.storage_kind)

class LazyKnitContentFactory(ContentFactory):

    """A ContentFactory which can either generate full text or a wire form.

    :seealso ContentFactory:

    """

    def __init__(self, key, parents, generator, first):

        """Create a LazyKnitContentFactory.

        :param key: The key of the record.

        :param parents: The parents of the record.

        :param generator: A _ContentMapGenerator containing the record for this

            key.

        :param first: Is this the first content object returned from generator?

            if it is, its storage kind is knit-delta-closure, otherwise it is

            knit-delta-closure-ref

        """

        self.key = key

        self.parents = parents

        self.sha1 = None

        self._generator = generator

        self.storage_kind = "knit-delta-closure"

        if not first:

            self.storage_kind = self.storage_kind + "-ref"

        self._first = first

    def get_bytes_as(self, storage_kind):

        if storage_kind == self.storage_kind:

            if self._first:

                return self._generator._wire_bytes()

            else:

                # all the keys etc are contained in the bytes returned in the

                # first record.

                return ''

        if storage_kind in ('chunked', 'fulltext'):

            chunks = self._generator._get_one_work(self.key).text()

            if storage_kind == 'chunked':

                return chunks

            else:

                return ''.join(chunks)

        raise errors.UnavailableRepresentation(self.key, storage_kind,

            self.storage_kind)

def knit_delta_closure_to_records(storage_kind, bytes, line_end):

    """Convert a network record to a iterator over stream records.

    :param storage_kind: The storage kind of the record.

        Must be 'knit-delta-closure'.

    :param bytes: The bytes of the record on the network.

    """

    generator = _NetworkContentMapGenerator(bytes, line_end)

    return generator.get_record_stream()

def knit_network_to_record(storage_kind, bytes, line_end):

    """Convert a network record to a record object.

    :param storage_kind: The storage kind of the record.

    :param bytes: The bytes of the record on the network.

    """

    start = line_end

    line_end = bytes.find('\n', start)

    key = tuple(bytes[start:line_end].split('\x00'))

    start = line_end + 1

    line_end = bytes.find('\n', start)

    parent_line = bytes[start:line_end]

    if parent_line == 'None:':

        parents = None

    else:

        parents = tuple(

            [tuple(segment.split('\x00')) for segment in parent_line.split('\t')

             if segment])

    start = line_end + 1

    noeol = bytes[start] == 'N'

    if 'ft' in storage_kind:

        method = 'fulltext'

    else:

        method = 'line-delta'

    build_details = (method, noeol)

    start = start + 1

    raw_record = bytes[start:]

    annotated = 'annotated' in storage_kind

    return [KnitContentFactory(key, parents, build_details, None, raw_record,

        annotated, network_bytes=bytes)]

class KnitContent(object):

    """Content of a knit version to which deltas can be applied.

    This is always stored in memory as a list of lines with \n at the end,

    plus a flag saying if the final ending is really there or not, because that

    corresponds to the on-disk knit representation.

    """

    def __init__(self):

        self._should_strip_eol = False

    def apply_delta(self, delta, new_version_id):

        """Apply delta to this object to become new_version_id."""

        raise NotImplementedError(self.apply_delta)

    def line_delta_iter(self, new_lines):

        """Generate line-based delta from this content to new_lines."""

        new_texts = new_lines.text()

        old_texts = self.text()

        s = patiencediff.PatienceSequenceMatcher(None, old_texts, new_texts)

        for tag, i1, i2, j1, j2 in s.get_opcodes():

            if tag == 'equal':

                continue

            # ofrom, oto, length, data

            yield i1, i2, j2 - j1, new_lines._lines[j1:j2]

    def line_delta(self, new_lines):

        return list(self.line_delta_iter(new_lines))

    @staticmethod

    def get_line_delta_blocks(knit_delta, source, target):

        """Extract SequenceMatcher.get_matching_blocks() from a knit delta"""

        target_len = len(target)

        s_pos = 0

        t_pos = 0

        for s_begin, s_end, t_len, new_text in knit_delta:

            true_n = s_begin - s_pos

            n = true_n

            if n > 0:

                # knit deltas do not provide reliable info about whether the

                # last line of a file matches, due to eol handling.

                if source[s_pos + n -1] != target[t_pos + n -1]:

                    n-=1

                if n > 0:

                    yield s_pos, t_pos, n

            t_pos += t_len + true_n

            s_pos = s_end

        n = target_len - t_pos

        if n > 0:

            if source[s_pos + n -1] != target[t_pos + n -1]:

                n-=1

            if n > 0:

                yield s_pos, t_pos, n

        yield s_pos + (target_len - t_pos), target_len, 0

class AnnotatedKnitContent(KnitContent):

    """Annotated content."""

    def __init__(self, lines):

        KnitContent.__init__(self)

        self._lines = lines

    def annotate(self):

        """Return a list of (origin, text) for each content line."""

        lines = self._lines[:]

        if self._should_strip_eol:

            origin, last_line = lines[-1]

            lines[-1] = (origin, last_line.rstrip('\n'))

        return lines

    def apply_delta(self, delta, new_version_id):

        """Apply delta to this object to become new_version_id."""

        offset = 0

        lines = self._lines

        for start, end, count, delta_lines in delta:

            lines[offset+start:offset+end] = delta_lines

            offset = offset + (start - end) + count

    def text(self):

        try:

            lines = [text for origin, text in self._lines]

        except ValueError, e:

            # most commonly (only?) caused by the internal form of the knit

            # missing annotation information because of a bug - see thread

            # around 20071015

            raise KnitCorrupt(self,

                "line in annotated knit missing annotation information: %s"

                % (e,))

        if self._should_strip_eol:

            lines[-1] = lines[-1].rstrip('\n')

        return lines

    def copy(self):

        return AnnotatedKnitContent(self._lines[:])

class PlainKnitContent(KnitContent):

    """Unannotated content.

    When annotate[_iter] is called on this content, the same version is reported

    for all lines. Generally, annotate[_iter] is not useful on PlainKnitContent

    objects.

    """

    def __init__(self, lines, version_id):

        KnitContent.__init__(self)

        self._lines = lines

        self._version_id = version_id

    def annotate(self):

        """Return a list of (origin, text) for each content line."""

        return [(self._version_id, line) for line in self._lines]

    def apply_delta(self, delta, new_version_id):

        """Apply delta to this object to become new_version_id."""

        offset = 0

        lines = self._lines

        for start, end, count, delta_lines in delta:

            lines[offset+start:offset+end] = delta_lines

            offset = offset + (start - end) + count

        self._version_id = new_version_id

    def copy(self):

        return PlainKnitContent(self._lines[:], self._version_id)

    def text(self):

        lines = self._lines

        if self._should_strip_eol:

            lines = lines[:]

            lines[-1] = lines[-1].rstrip('\n')

        return lines

class _KnitFactory(object):

    """Base class for common Factory functions."""

    def parse_record(self, version_id, record, record_details,

                     base_content, copy_base_content=True):

        """Parse a record into a full content object.

        :param version_id: The official version id for this content

        :param record: The data returned by read_records_iter()

        :param record_details: Details about the record returned by

            get_build_details

        :param base_content: If get_build_details returns a compression_parent,

            you must return a base_content here, else use None

        :param copy_base_content: When building from the base_content, decide

            you can either copy it and return a new object, or modify it in

            place.

        :return: (content, delta) A Content object and possibly a line-delta,

            delta may be None

        """

        method, noeol = record_details

        if method == 'line-delta':

            if copy_base_content:

                content = base_content.copy()

            else:

                content = base_content

            delta = self.parse_line_delta(record, version_id)

            content.apply_delta(delta, version_id)

        else:

            content = self.parse_fulltext(record, version_id)

            delta = None

        content._should_strip_eol = noeol

        return (content, delta)

class KnitAnnotateFactory(_KnitFactory):

    """Factory for creating annotated Content objects."""

    annotated = True

    def make(self, lines, version_id):

        num_lines = len(lines)

        return AnnotatedKnitContent(zip([version_id] * num_lines, lines))

    def parse_fulltext(self, content, version_id):

        """Convert fulltext to internal representation

        fulltext content is of the format

        revid(utf8) plaintext\n

        internal representation is of the format:

        (revid, plaintext)

        """

        # TODO: jam 20070209 The tests expect this to be returned as tuples,

        #       but the code itself doesn't really depend on that.

        #       Figure out a way to not require the overhead of turning the

        #       list back into tuples.

        lines = [tuple(line.split(' ', 1)) for line in content]

        return AnnotatedKnitContent(lines)

    def parse_line_delta_iter(self, lines):

        return iter(self.parse_line_delta(lines))

    def parse_line_delta(self, lines, version_id, plain=False):

        """Convert a line based delta into internal representation.

        line delta is in the form of:

        intstart intend intcount

        1..count lines:

        revid(utf8) newline\n

        internal representation is

        (start, end, count, [1..count tuples (revid, newline)])

        :param plain: If True, the lines are returned as a plain

            list without annotations, not as a list of (origin, content) tuples, i.e.

            (start, end, count, [1..count newline])

        """

        result = []

        lines = iter(lines)

        next = lines.next

        cache = {}

        def cache_and_return(line):

            origin, text = line.split(' ', 1)

            return cache.setdefault(origin, origin), text

        # walk through the lines parsing.

        # Note that the plain test is explicitly pulled out of the

        # loop to minimise any performance impact

        if plain:

            for header in lines:

                start, end, count = [int(n) for n in header.split(',')]

                contents = [next().split(' ', 1)[1] for i in xrange(count)]

                result.append((start, end, count, contents))

        else:

            for header in lines:

                start, end, count = [int(n) for n in header.split(',')]

                contents = [tuple(next().split(' ', 1)) for i in xrange(count)]

                result.append((start, end, count, contents))

        return result

    def get_fulltext_content(self, lines):

        """Extract just the content lines from a fulltext."""

        return (line.split(' ', 1)[1] for line in lines)

    def get_linedelta_content(self, lines):

        """Extract just the content from a line delta.

        This doesn't return all of the extra information stored in a delta.

        Only the actual content lines.

        """

        lines = iter(lines)

        next = lines.next

        for header in lines:

            header = header.split(',')

            count = int(header[2])

            for i in xrange(count):

                origin, text = next().split(' ', 1)

                yield text

    def lower_fulltext(self, content):

        """convert a fulltext content record into a serializable form.

        see parse_fulltext which this inverts.

        """

        # TODO: jam 20070209 We only do the caching thing to make sure that

        #       the origin is a valid utf-8 line, eventually we could remove it

        return ['%s %s' % (o, t) for o, t in content._lines]

    def lower_line_delta(self, delta):

        """convert a delta into a serializable form.

        See parse_line_delta which this inverts.

        """

        # TODO: jam 20070209 We only do the caching thing to make sure that

        #       the origin is a valid utf-8 line, eventually we could remove it

        out = []

        for start, end, c, lines in delta:

            out.append('%d,%d,%d\n' % (start, end, c))

            out.extend(origin + ' ' + text

                       for origin, text in lines)