code.grnet.gr Git - pithos/blob - pithos/backends/lib/sqlite/node.py

   1 # Copyright 2011 GRNET S.A. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or
   4 # without modification, are permitted provided that the following
   5 # conditions are met:
   6 #
   7 #   1. Redistributions of source code must retain the above
   8 #      copyright notice, this list of conditions and the following
   9 #      disclaimer.
  10 #
  11 #   2. Redistributions in binary form must reproduce the above
  12 #      copyright notice, this list of conditions and the following
  13 #      disclaimer in the documentation and/or other materials
  14 #      provided with the distribution.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  17 # OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  18 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  19 # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  20 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  23 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  24 # AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  25 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  26 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  27 # POSSIBILITY OF SUCH DAMAGE.
  28 #
  29 # The views and conclusions contained in the software and
  30 # documentation are those of the authors and should not be
  31 # interpreted as representing official policies, either expressed
  32 # or implied, of GRNET S.A.
  33
  34 from time import time
  35
  36 from dbworker import DBWorker
  37
  38 from pithos.lib.filter import parse_filters, OPERATORS
  39
  40
  41 ROOTNODE  = 0
  42
  43 ( SERIAL, NODE, HASH, SIZE, SOURCE, MTIME, MUSER, UUID, CLUSTER ) = range(9)
  44
  45 inf = float('inf')
  46
  47
  48 def strnextling(prefix):
  49     """Return the first unicode string
  50        greater than but not starting with given prefix.
  51        strnextling('hello') -> 'hellp'
  52     """
  53     if not prefix:
  54         ## all strings start with the null string,
  55         ## therefore we have to approximate strnextling('')
  56         ## with the last unicode character supported by python
  57         ## 0x10ffff for wide (32-bit unicode) python builds
  58         ## 0x00ffff for narrow (16-bit unicode) python builds
  59         ## We will not autodetect. 0xffff is safe enough.
  60         return unichr(0xffff)
  61     s = prefix[:-1]
  62     c = ord(prefix[-1])
  63     if c >= 0xffff:
  64         raise RuntimeError
  65     s += unichr(c+1)
  66     return s
  67
  68 def strprevling(prefix):
  69     """Return an approximation of the last unicode string
  70        less than but not starting with given prefix.
  71        strprevling(u'hello') -> u'helln\\xffff'
  72     """
  73     if not prefix:
  74         ## There is no prevling for the null string
  75         return prefix
  76     s = prefix[:-1]
  77     c = ord(prefix[-1])
  78     if c > 0:
  79         s += unichr(c-1) + unichr(0xffff)
  80     return s
  81
  82
  83 _propnames = {
  84     'serial'    : 0,
  85     'node'      : 1,
  86     'hash'      : 2,
  87     'size'      : 3,
  88     'source'    : 4,
  89     'mtime'     : 5,
  90     'muser'     : 6,
  91     'uuid'      : 7,
  92     'cluster'   : 8
  93 }
  94
  95
  96 class Node(DBWorker):
  97     """Nodes store path organization and have multiple versions.
  98        Versions store object history and have multiple attributes.
  99        Attributes store metadata.
 100     """
 101
 102     # TODO: Provide an interface for included and excluded clusters.
 103
 104     def __init__(self, **params):
 105         DBWorker.__init__(self, **params)
 106         execute = self.execute
 107
 108         execute(""" pragma foreign_keys = on """)
 109
 110         execute(""" create table if not exists nodes
 111                           ( node       integer primary key,
 112                             parent     integer default 0,
 113                             path       text    not null default '',
 114                             foreign key (parent)
 115                             references nodes(node)
 116                             on update cascade
 117                             on delete cascade ) """)
 118         execute(""" create unique index if not exists idx_nodes_path
 119                     on nodes(path) """)
 120
 121         execute(""" create table if not exists policy
 122                           ( node   integer,
 123                             key    text,
 124                             value  text,
 125                             primary key (node, key)
 126                             foreign key (node)
 127                             references nodes(node)
 128                             on update cascade
 129                             on delete cascade ) """)
 130
 131         execute(""" create table if not exists statistics
 132                           ( node       integer,
 133                             population integer not null default 0,
 134                             size       integer not null default 0,
 135                             mtime      integer,
 136                             cluster    integer not null default 0,
 137                             primary key (node, cluster)
 138                             foreign key (node)
 139                             references nodes(node)
 140                             on update cascade
 141                             on delete cascade ) """)
 142
 143         execute(""" create table if not exists versions
 144                           ( serial     integer primary key,
 145                             node       integer,
 146                             hash       text,
 147                             size       integer not null default 0,
 148                             source     integer,
 149                             mtime      integer,
 150                             muser      text    not null default '',
 151                             uuid       text    not null default '',
 152                             cluster    integer not null default 0,
 153                             foreign key (node)
 154                             references nodes(node)
 155                             on update cascade
 156                             on delete cascade ) """)
 157         execute(""" create index if not exists idx_versions_node_mtime
 158                     on versions(node, mtime) """)
 159         execute(""" create index if not exists idx_versions_node_uuid
 160                     on versions(uuid) """)
 161
 162         execute(""" create table if not exists attributes
 163                           ( serial integer,
 164                             domain text,
 165                             key    text,
 166                             value  text,
 167                             primary key (serial, domain, key)
 168                             foreign key (serial)
 169                             references versions(serial)
 170                             on update cascade
 171                             on delete cascade ) """)
 172
 173         q = "insert or ignore into nodes(node, parent) values (?, ?)"
 174         execute(q, (ROOTNODE, ROOTNODE))
 175
 176     def node_create(self, parent, path):
 177         """Create a new node from the given properties.
 178            Return the node identifier of the new node.
 179         """
 180
 181         q = ("insert into nodes (parent, path) "
 182              "values (?, ?)")
 183         props = (parent, path)
 184         return self.execute(q, props).lastrowid
 185
 186     def node_lookup(self, path):
 187         """Lookup the current node of the given path.
 188            Return None if the path is not found.
 189         """
 190
 191         q = "select node from nodes where path = ?"
 192         self.execute(q, (path,))
 193         r = self.fetchone()
 194         if r is not None:
 195             return r[0]
 196         return None
 197
 198     def node_get_properties(self, node):
 199         """Return the node's (parent, path).
 200            Return None if the node is not found.
 201         """
 202
 203         q = "select parent, path from nodes where node = ?"
 204         self.execute(q, (node,))
 205         return self.fetchone()
 206
 207     def node_get_versions(self, node, keys=(), propnames=_propnames):
 208         """Return the properties of all versions at node.
 209            If keys is empty, return all properties in the order
 210            (serial, node, hash, size, source, mtime, muser, uuid, cluster).
 211         """
 212
 213         q = ("select serial, node, hash, size, source, mtime, muser, uuid, cluster "
 214              "from versions "
 215              "where node = ?")
 216         self.execute(q, (node,))
 217         r = self.fetchall()
 218         if r is None:
 219             return r
 220
 221         if not keys:
 222             return r
 223         return [[p[propnames[k]] for k in keys if k in propnames] for p in r]
 224
 225     def node_count_children(self, node):
 226         """Return node's child count."""
 227
 228         q = "select count(node) from nodes where parent = ? and node != 0"
 229         self.execute(q, (node,))
 230         r = self.fetchone()
 231         if r is None:
 232             return 0
 233         return r[0]
 234
 235     def node_purge_children(self, parent, before=inf, cluster=0):
 236         """Delete all versions with the specified
 237            parent and cluster, and return
 238            the hashes of versions deleted.
 239            Clears out nodes with no remaining versions.
 240         """
 241
 242         execute = self.execute
 243         q = ("select count(serial), sum(size) from versions "
 244              "where node in (select node "
 245                             "from nodes "
 246                             "where parent = ?) "
 247              "and cluster = ? "
 248              "and mtime <= ?")
 249         args = (parent, cluster, before)
 250         execute(q, args)
 251         nr, size = self.fetchone()
 252         if not nr:
 253             return ()
 254         mtime = time()
 255         self.statistics_update(parent, -nr, -size, mtime, cluster)
 256         self.statistics_update_ancestors(parent, -nr, -size, mtime, cluster)
 257
 258         q = ("select hash from versions "
 259              "where node in (select node "
 260                             "from nodes "
 261                             "where parent = ?) "
 262              "and cluster = ? "
 263              "and mtime <= ?")
 264         execute(q, args)
 265         hashes = [r[0] for r in self.fetchall()]
 266         q = ("delete from versions "
 267              "where node in (select node "
 268                             "from nodes "
 269                             "where parent = ?) "
 270              "and cluster = ? "
 271              "and mtime <= ?")
 272         execute(q, args)
 273         q = ("delete from nodes "
 274              "where node in (select node from nodes n "
 275                             "where (select count(serial) "
 276                                    "from versions "
 277                                    "where node = n.node) = 0 "
 278                             "and parent = ?)")
 279         execute(q, (parent,))
 280         return hashes
 281
 282     def node_purge(self, node, before=inf, cluster=0):
 283         """Delete all versions with the specified
 284            node and cluster, and return
 285            the hashes of versions deleted.
 286            Clears out the node if it has no remaining versions.
 287         """
 288
 289         execute = self.execute
 290         q = ("select count(serial), sum(size) from versions "
 291              "where node = ? "
 292              "and cluster = ? "
 293              "and mtime <= ?")
 294         args = (node, cluster, before)
 295         execute(q, args)
 296         nr, size = self.fetchone()
 297         if not nr:
 298             return ()
 299         mtime = time()
 300         self.statistics_update_ancestors(node, -nr, -size, mtime, cluster)
 301
 302         q = ("select hash from versions "
 303              "where node = ? "
 304              "and cluster = ? "
 305              "and mtime <= ?")
 306         execute(q, args)
 307         hashes = [r[0] for r in self.fetchall()]
 308         q = ("delete from versions "
 309              "where node = ? "
 310              "and cluster = ? "
 311              "and mtime <= ?")
 312         execute(q, args)
 313         q = ("delete from nodes "
 314              "where node in (select node from nodes n "
 315                             "where (select count(serial) "
 316                                    "from versions "
 317                                    "where node = n.node) = 0 "
 318                             "and node = ?)")
 319         execute(q, (node,))
 320         return hashes
 321
 322     def node_remove(self, node):
 323         """Remove the node specified.
 324            Return false if the node has children or is not found.
 325         """
 326
 327         if self.node_count_children(node):
 328             return False
 329
 330         mtime = time()
 331         q = ("select count(serial), sum(size), cluster "
 332              "from versions "
 333              "where node = ? "
 334              "group by cluster")
 335         self.execute(q, (node,))
 336         for population, size, cluster in self.fetchall():
 337             self.statistics_update_ancestors(node, -population, -size, mtime, cluster)
 338
 339         q = "delete from nodes where node = ?"
 340         self.execute(q, (node,))
 341         return True
 342
 343     def policy_get(self, node):
 344         q = "select key, value from policy where node = ?"
 345         self.execute(q, (node,))
 346         return dict(self.fetchall())
 347
 348     def policy_set(self, node, policy):
 349         q = "insert or replace into policy (node, key, value) values (?, ?, ?)"
 350         self.executemany(q, ((node, k, v) for k, v in policy.iteritems()))
 351
 352     def statistics_get(self, node, cluster=0):
 353         """Return population, total size and last mtime
 354            for all versions under node that belong to the cluster.
 355         """
 356
 357         q = ("select population, size, mtime from statistics "
 358              "where node = ? and cluster = ?")
 359         self.execute(q, (node, cluster))
 360         return self.fetchone()
 361
 362     def statistics_update(self, node, population, size, mtime, cluster=0):
 363         """Update the statistics of the given node.
 364            Statistics keep track the population, total
 365            size of objects and mtime in the node's namespace.
 366            May be zero or positive or negative numbers.
 367         """
 368
 369         qs = ("select population, size from statistics "
 370               "where node = ? and cluster = ?")
 371         qu = ("insert or replace into statistics (node, population, size, mtime, cluster) "
 372               "values (?, ?, ?, ?, ?)")
 373         self.execute(qs, (node, cluster))
 374         r = self.fetchone()
 375         if r is None:
 376             prepopulation, presize = (0, 0)
 377         else:
 378             prepopulation, presize = r
 379         population += prepopulation
 380         size += presize
 381         self.execute(qu, (node, population, size, mtime, cluster))
 382
 383     def statistics_update_ancestors(self, node, population, size, mtime, cluster=0):
 384         """Update the statistics of the given node's parent.
 385            Then recursively update all parents up to the root.
 386            Population is not recursive.
 387         """
 388
 389         while True:
 390             if node == 0:
 391                 break
 392             props = self.node_get_properties(node)
 393             if props is None:
 394                 break
 395             parent, path = props
 396             self.statistics_update(parent, population, size, mtime, cluster)
 397             node = parent
 398             population = 0 # Population isn't recursive
 399
 400     def statistics_latest(self, node, before=inf, except_cluster=0):
 401         """Return population, total size and last mtime
 402            for all latest versions under node that
 403            do not belong to the cluster.
 404         """
 405
 406         execute = self.execute
 407         fetchone = self.fetchone
 408
 409         # The node.
 410         props = self.node_get_properties(node)
 411         if props is None:
 412             return None
 413         parent, path = props
 414
 415         # The latest version.
 416         q = ("select serial, node, hash, size, source, mtime, muser, uuid, cluster "
 417              "from versions "
 418              "where serial = (select max(serial) "
 419                              "from versions "
 420                              "where node = ? and mtime < ?) "
 421              "and cluster != ?")
 422         execute(q, (node, before, except_cluster))
 423         props = fetchone()
 424         if props is None:
 425             return None
 426         mtime = props[MTIME]
 427
 428         # First level, just under node (get population).
 429         q = ("select count(serial), sum(size), max(mtime) "
 430              "from versions v "
 431              "where serial = (select max(serial) "
 432                              "from versions "
 433                              "where node = v.node and mtime < ?) "
 434              "and cluster != ? "
 435              "and node in (select node "
 436                           "from nodes "
 437                           "where parent = ?)")
 438         execute(q, (before, except_cluster, node))
 439         r = fetchone()
 440         if r is None:
 441             return None
 442         count = r[0]
 443         mtime = max(mtime, r[2])
 444         if count == 0:
 445             return (0, 0, mtime)
 446
 447         # All children (get size and mtime).
 448         # XXX: This is why the full path is stored.
 449         q = ("select count(serial), sum(size), max(mtime) "
 450              "from versions v "
 451              "where serial = (select max(serial) "
 452                              "from versions "
 453                              "where node = v.node and mtime < ?) "
 454              "and cluster != ? "
 455              "and node in (select node "
 456                           "from nodes "
 457                           "where path like ? escape '\\')")
 458         execute(q, (before, except_cluster, self.escape_like(path) + '%'))
 459         r = fetchone()
 460         if r is None:
 461             return None
 462         size = r[1] - props[SIZE]
 463         mtime = max(mtime, r[2])
 464         return (count, size, mtime)
 465
 466     def version_create(self, node, hash, size, source, muser, uuid, cluster=0):
 467         """Create a new version from the given properties.
 468            Return the (serial, mtime) of the new version.
 469         """
 470
 471         q = ("insert into versions (node, hash, size, source, mtime, muser, uuid, cluster) "
 472              "values (?, ?, ?, ?, ?, ?, ?, ?)")
 473         mtime = time()
 474         props = (node, hash, size, source, mtime, muser, uuid, cluster)
 475         serial = self.execute(q, props).lastrowid
 476         self.statistics_update_ancestors(node, 1, size, mtime, cluster)
 477         return serial, mtime
 478
 479     def version_lookup(self, node, before=inf, cluster=0):
 480         """Lookup the current version of the given node.
 481            Return a list with its properties:
 482            (serial, node, hash, size, source, mtime, muser, uuid, cluster)
 483            or None if the current version is not found in the given cluster.
 484         """
 485
 486         q = ("select serial, node, hash, size, source, mtime, muser, uuid, cluster "
 487              "from versions "
 488              "where serial = (select max(serial) "
 489                              "from versions "
 490                              "where node = ? and mtime < ?) "
 491              "and cluster = ?")
 492         self.execute(q, (node, before, cluster))
 493         props = self.fetchone()
 494         if props is not None:
 495             return props
 496         return None
 497
 498     def version_get_properties(self, serial, keys=(), propnames=_propnames):
 499         """Return a sequence of values for the properties of
 500            the version specified by serial and the keys, in the order given.
 501            If keys is empty, return all properties in the order
 502            (serial, node, hash, size, source, mtime, muser, uuid, cluster).
 503         """
 504
 505         q = ("select serial, node, hash, size, source, mtime, muser, uuid, cluster "
 506              "from versions "
 507              "where serial = ?")
 508         self.execute(q, (serial,))
 509         r = self.fetchone()
 510         if r is None:
 511             return r
 512
 513         if not keys:
 514             return r
 515         return [r[propnames[k]] for k in keys if k in propnames]
 516
 517     def version_recluster(self, serial, cluster):
 518         """Move the version into another cluster."""
 519
 520         props = self.version_get_properties(serial)
 521         if not props:
 522             return
 523         node = props[NODE]
 524         size = props[SIZE]
 525         oldcluster = props[CLUSTER]
 526         if cluster == oldcluster:
 527             return
 528
 529         mtime = time()
 530         self.statistics_update_ancestors(node, -1, -size, mtime, oldcluster)
 531         self.statistics_update_ancestors(node, 1, size, mtime, cluster)
 532
 533         q = "update versions set cluster = ? where serial = ?"
 534         self.execute(q, (cluster, serial))
 535
 536     def version_remove(self, serial):
 537         """Remove the serial specified."""
 538
 539         props = self.version_get_properties(serial)
 540         if not props:
 541             return
 542         node = props[NODE]
 543         hash = props[HASH]
 544         size = props[SIZE]
 545         cluster = props[CLUSTER]
 546
 547         mtime = time()
 548         self.statistics_update_ancestors(node, -1, -size, mtime, cluster)
 549
 550         q = "delete from versions where serial = ?"
 551         self.execute(q, (serial,))
 552         return hash
 553
 554     def attribute_get(self, serial, domain, keys=()):
 555         """Return a list of (key, value) pairs of the version specified by serial.
 556            If keys is empty, return all attributes.
 557            Othwerise, return only those specified.
 558         """
 559
 560         execute = self.execute
 561         if keys:
 562             marks = ','.join('?' for k in keys)
 563             q = ("select key, value from attributes "
 564                  "where key in (%s) and serial = ? and domain = ?" % (marks,))
 565             execute(q, keys + (serial, domain))
 566         else:
 567             q = "select key, value from attributes where serial = ? and domain = ?"
 568             execute(q, (serial, domain))
 569         return self.fetchall()
 570
 571     def attribute_set(self, serial, domain, items):
 572         """Set the attributes of the version specified by serial.
 573            Receive attributes as an iterable of (key, value) pairs.
 574         """
 575
 576         q = ("insert or replace into attributes (serial, domain, key, value) "
 577              "values (?, ?, ?, ?)")
 578         self.executemany(q, ((serial, domain, k, v) for k, v in items))
 579
 580     def attribute_del(self, serial, domain, keys=()):
 581         """Delete attributes of the version specified by serial.
 582            If keys is empty, delete all attributes.
 583            Otherwise delete those specified.
 584         """
 585
 586         if keys:
 587             q = "delete from attributes where serial = ? and domain = ? and key = ?"
 588             self.executemany(q, ((serial, domain, key) for key in keys))
 589         else:
 590             q = "delete from attributes where serial = ? and domain = ?"
 591             self.execute(q, (serial, domain))
 592
 593     def attribute_copy(self, source, dest):
 594         q = ("insert or replace into attributes "
 595              "select ?, domain, key, value from attributes "
 596              "where serial = ?")
 597         self.execute(q, (dest, source))
 598
 599     def _construct_paths(self, pathq):
 600         if not pathq:
 601             return None, None
 602
 603         subq = " and ("
 604         subq += ' or '.join(("n.path like ? escape '\\'" for x in pathq))
 605         subq += ")"
 606         args = tuple([self.escape_like(x) + '%' for x in pathq])
 607
 608         return subq, args
 609
 610     def latest_attribute_keys(self, parent, domain, before=inf, except_cluster=0, pathq=[]):
 611         """Return a list with all keys pairs defined
 612            for all latest versions under parent that
 613            do not belong to the cluster.
 614         """
 615
 616         # TODO: Use another table to store before=inf results.
 617         q = ("select distinct a.key "
 618              "from attributes a, versions v, nodes n "
 619              "where v.serial = (select max(serial) "
 620                                "from versions "
 621                                "where node = v.node and mtime < ?) "
 622              "and v.cluster != ? "
 623              "and v.node in (select node "
 624                             "from nodes "
 625                             "where parent = ?) "
 626              "and a.serial = v.serial "
 627              "and a.domain = ? "
 628              "and n.node = v.node")
 629         args = (before, except_cluster, parent, domain)
 630         subq, subargs = self._construct_paths(pathq)
 631         if subq is not None:
 632             q += subq
 633             args += subargs
 634         self.execute(q, args)
 635         return [r[0] for r in self.fetchall()]
 636
 637     def latest_version_list(self, parent, prefix='', delimiter=None,
 638                             start='', limit=10000, before=inf,
 639                             except_cluster=0, pathq=[], domain=None, filterq=[]):
 640         """Return a (list of (path, serial) tuples, list of common prefixes)
 641            for the current versions of the paths with the given parent,
 642            matching the following criteria.
 643
 644            The property tuple for a version is returned if all
 645            of these conditions are true:
 646
 647                 a. parent matches
 648
 649                 b. path > start
 650
 651                 c. path starts with prefix (and paths in pathq)
 652
 653                 d. version is the max up to before
 654
 655                 e. version is not in cluster
 656
 657                 f. the path does not have the delimiter occuring
 658                    after the prefix, or ends with the delimiter
 659
 660                 g. serial matches the attribute filter query.
 661
 662                    A filter query is a comma-separated list of
 663                    terms in one of these three forms:
 664
 665                    key
 666                        an attribute with this key must exist
 667
 668                    !key
 669                        an attribute with this key must not exist
 670
 671                    key ?op value
 672                        the attribute with this key satisfies the value
 673                        where ?op is one of =, != <=, >=, <, >.
 674
 675            The list of common prefixes includes the prefixes
 676            matching up to the first delimiter after prefix,
 677            and are reported only once, as "virtual directories".
 678            The delimiter is included in the prefixes.
 679
 680            If arguments are None, then the corresponding matching rule
 681            will always match.
 682
 683            Limit applies to the first list of tuples returned.
 684         """
 685         execute = self.execute
 686
 687         if not start or start < prefix:
 688             start = strprevling(prefix)
 689         nextling = strnextling(prefix)
 690
 691         q = ("select distinct n.path, v.serial "
 692              "from versions v, nodes n "
 693              "where v.serial = (select max(serial) "
 694                                "from versions "
 695                                "where node = v.node and mtime < ?) "
 696              "and v.cluster != ? "
 697              "and v.node in (select node "
 698                             "from nodes "
 699                             "where parent = ?) "
 700              "and n.node = v.node "
 701              "and n.path > ? and n.path < ?")
 702         args = [before, except_cluster, parent, start, nextling]
 703
 704         subq, subargs = self._construct_paths(pathq)
 705         if subq is not None:
 706             q += subq
 707             args += subargs
 708         q += " order by n.path"
 709
 710         def filterout(r):
 711             if not filterq:
 712                 return False
 713             path, serial = r
 714             included, excluded, opers = parse_filters(filterq)
 715
 716             #retrieve metadata
 717             meta = dict(self.attribute_get(serial, domain))
 718             keyset= set([k.encode('utf8') for k in meta.keys()])
 719
 720             if included:
 721                 if not set(included) & keyset:
 722                     return True
 723             if excluded:
 724                 if set(excluded) & keyset:
 725                     return True
 726             for k, op, v in opers:
 727                 k = k.decode('utf8')
 728                 v = v.decode('utf8')
 729                 if k not in meta:
 730                     return True
 731                 operation = OPERATORS[op]
 732                 if not operation(meta[k], v):
 733                     return True
 734             return False
 735
 736         if not delimiter:
 737             q += " limit ?"
 738             args.append(limit)
 739             execute(q, args)
 740             filter_ = lambda r : [t for t in r if not filterout(t)]
 741             r = filter_(self.fetchall())
 742             return r, ()
 743
 744         pfz = len(prefix)
 745         dz = len(delimiter)
 746         count = 0
 747         prefixes = []
 748         pappend = prefixes.append
 749         matches = []
 750         mappend = matches.append
 751
 752         execute(q, args)
 753         r = self.fetchall()
 754         i = 0
 755         while i < len(r):
 756             props = r[i]
 757             if props is None:
 758                 break
 759             if filterout(props):
 760                 i+=1
 761                 continue
 762             path, serial = props
 763             idx = path.find(delimiter, pfz)
 764
 765             if idx < 0:
 766                 mappend(props)
 767                 count += 1
 768                 if count >= limit:
 769                     break
 770                 i+=1
 771                 continue
 772
 773             if idx + dz == len(path):
 774                 mappend(props)
 775                 count += 1
 776                 i+=1
 777                 continue # Get one more, in case there is a path.
 778             pf = path[:idx + dz]
 779             pappend(pf)
 780             if count >= limit:
 781                 break
 782
 783             args[3] = strnextling(pf) # New start.
 784             execute(q, args)
 785             r = self.fetchall()
 786             i = 0
 787
 788         return matches, prefixes
 789
 790     def latest_uuid(self, uuid):
 791         """Return a (path, serial) tuple, for the latest version of the given uuid."""
 792
 793         q = ("select n.path, v.serial "
 794              "from versions v, nodes n "
 795              "where v.serial = (select max(serial) "
 796                                "from versions "
 797                                "where uuid = ?) "
 798              "and n.node = v.node")
 799         self.execute(q, (uuid,))
 800         return self.fetchone()