path: root/active_document/index_proxy.py

# Copyright (C) 2012 Aleksey Lim
#
# 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 3 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, see <http://www.gnu.org/licenses/>.

import bisect
import logging
from os.path import exists

import xapian

from active_toolkit import util
from active_document import index_queue, env
from active_document.storage import Storage
from active_document.index import IndexReader, IndexWriter
from active_document.metadata import StoredProperty


_logger = logging.getLogger('active_document.index_proxy')


class IndexProxy(IndexReader):

    def __init__(self, root, metadata):
        self._commit_seqno = 0
        self._cache_seqno = 1
        self._term_props = {}
        self._pages = {}
        self._dirty = False

        for prop in metadata.values():
            if isinstance(prop, StoredProperty) and \
                    prop.permissions & env.ACCESS_WRITE:
                self._term_props[prop.name] = prop

        IndexReader.__init__(self, root, metadata)

    def commit(self):
        index_queue.commit_and_wait(self.metadata.name)

    def close(self):
        pass

    def get_cached(self, guid):
        self._drop_pages()
        result = {}
        for page in self._sorted_pages:
            cached = page.get(guid)
            if cached is not None:
                result.update(cached.properties)
        return result

    def store(self, guid, properties, new, pre_cb=None, post_cb=None, *args):
        if properties and new is not None:
            if new:
                orig = None
            else:
                orig = self.get_cached(guid)
                try:
                    # XXX Avoid creating Storage every time
                    record = Storage(self._path, self.metadata).get(guid)
                    for prop in self._term_props.values():
                        if prop.name not in orig:
                            orig[prop.name] = record.get(prop.name)['value']
                except env.NotFound:
                    pass
            # Needs to be called before `index_queue.put()`
            # to give it a chance to read original properties from the storage
            page = self._pages.get(self._cache_seqno)
            if page is None:
                page = self._pages[self._cache_seqno] = \
                        _CachedPage(self._term_props)
                _logger.debug('New cache page for %r with seqno %s to '
                        'insert %r', self.metadata.name, self._cache_seqno,
                        guid)
            page.update(guid, properties, orig)

        self._put(IndexWriter.store, guid, properties, new, pre_cb, post_cb,
                *args)

    def delete(self, guid, post_cb=None, *args):
        self._put(IndexWriter.delete, guid, post_cb, *args)

    def find(self, query):
        self._do_open(False)

        if query.no_cache:
            pages = []
        else:
            pages = self._sorted_pages

        def next_page_find(query):
            if pages:
                return pages.pop().find(query, next_page_find)
            elif self._db is None:
                return [], 0
            else:
                return IndexReader.find(self, query)

        return next_page_find(query)

    @property
    def _sorted_pages(self):
        return [self._pages[i] for i in sorted(self._pages.keys())]

    def _do_open(self, reset):
        if reset:
            self._dirty = True

        if self._db is None:
            if exists(self._path):
                self._drop_pages()
            else:
                return
        elif not self._drop_pages() and not self._dirty:
            return

        try:
            if self._db is None:
                self._db = xapian.Database(self._path)
                _logger.debug('Opened %r RO index', self.metadata.name)
            else:
                self._db.reopen()
                _logger.debug('Re-opened %r RO index', self.metadata.name)
            self._dirty = False
        except Exception:
            util.exception(_logger,
                    'Cannot open %r RO index', self.metadata.name)
            self._db = None

    def _drop_pages(self):
        commit_seqno = index_queue.commit_seqno(self.metadata.name)
        if commit_seqno == self._commit_seqno:
            return False
        for seqno in self._pages.keys():
            if seqno <= commit_seqno:
                del self._pages[seqno]
                _logger.debug('Drop cache page for %r with seqno %s',
                        self.metadata.name, seqno)
        self._commit_seqno = commit_seqno
        self._dirty = True
        return True

    def _put(self, op, *args):
        _logger.debug('Push %r(%r) to %rs queue',
                op, args, self.metadata.name)
        new_cache_seqno = index_queue.put(self.metadata.name, op, *args)
        if new_cache_seqno != self._cache_seqno:
            self._cache_seqno = new_cache_seqno
            self._pages[new_cache_seqno] = _CachedPage(self._term_props)
            _logger.debug('New cache page for %r with seqno %s',
                    self.metadata.name, new_cache_seqno)


class _CachedPage(dict):

    def __init__(self, term_props):
        self._term_props = term_props

    def update(self, guid, props, orig):
        existing = self.get(guid)
        if existing is None:
            self[guid] = _CachedDocument(self._term_props, guid, props, orig)
        else:
            existing.update(props)

    def find(self, query, direct_find):
        if 'guid' in query.request:
            documents, total = direct_find(query)
            cache = self.get(query.request['guid'])
            if cache is None:
                return documents, total

            def patched_guid_find():
                processed = False
                for guid, props in documents:
                    processed = True
                    props.update(cache.properties)
                    yield guid, props
                if not processed:
                    yield cache.guid, cache.properties

            return patched_guid_find(), total

        if not self:
            return direct_find(query)

        adds, deletes, updates = self._patch_find(query.request)
        if not adds and not deletes and not updates:
            return direct_find(query)

        orig_limit = query.limit
        query.limit += len(deletes)
        documents, total = direct_find(query)
        total.value += len(adds)

        def patched_find(orig_limit):
            for guid, props in documents:
                if orig_limit < 1:
                    break
                if guid in deletes:
                    total.value -= 1
                    continue
                cache = updates.get(guid)
                if cache is not None:
                    props.update(cache.properties)
                yield guid, props
                orig_limit -= 1

            for doc in adds:
                if orig_limit < 1:
                    break
                yield doc.guid, doc.properties
                orig_limit -= 1

        return patched_find(orig_limit), total

    def _patch_find(self, request):
        adds = []
        deletes = set()
        updates = {}

        terms = set()
        for prop_name, value in request.items():
            prop = self._term_props.get(prop_name)
            if prop is None:
                continue
            try:
                value = prop.decode(value)
            except ValueError, error:
                _logger.debug('Wrong request property value %r for %r '
                        'property, thus the whole request is empty: %s',
                        value, prop_name, error)
                return None, None, None
            terms.add(_TermValue(prop, value))

        for cache in self.values():
            if cache.new:
                if terms.issubset(cache.terms):
                    bisect.insort(adds, cache)
            else:
                if terms:
                    if terms.issubset(cache.terms):
                        if not terms.issubset(cache.orig_terms):
                            bisect.insort(adds, cache)
                            continue
                    else:
                        if terms.issubset(cache.orig_terms):
                            deletes.add(cache.guid)
                        continue
                updates[cache.guid] = cache

        return adds, deletes, updates


class _CachedDocument(object):

    def __init__(self, term_props, guid, properties, orig):
        self.guid = guid
        self.properties = properties.copy()
        self.new = orig is None
        self.terms = set()
        self.orig_terms = set()
        self._term_props = term_props

        for prop in term_props.values():
            if orig is not None:
                self.orig_terms.add(_TermValue(prop, orig.get(prop.name)))

        self._update_terms()

    def __sort__(self, other):
        return cmp(self.guid, other.guid)

    def update(self, properties):
        self.properties.update(properties)
        self._update_terms()

    def _update_terms(self):
        self.terms.clear()
        orig_terms = {}
        for i in self.orig_terms:
            orig_terms[i.prop] = i.value
        for prop in self._term_props.values():
            term = self.properties.get(prop.name, orig_terms.get(prop))
            self.terms.add(_TermValue(prop, term))


class _TermValue:

    def __init__(self, prop, value):
        self.prop = prop
        self.value = value

    def __repr__(self):
        return '%s=%s' % (self.prop.name, self.value)

    def __cmp__(self, other):
        result = cmp(self.prop.name, other.prop.name)
        if result:
            return result
        if not self.prop.composite:
            return cmp(self.value, other.value)
        self_value = set(self.value)
        other_value = set(other.value)
        if self_value and self_value.issubset(other_value) or \
                other_value and other_value.issubset(self_value):
            return 0
        else:
            return cmp(self.value, other.value)

    def __hash__(self):
        return hash(self.prop.name)