Hi all, I got problems similar to those described by Erik Enge a few weeks ago with indexing a somewhat larger amount of text (~500 MB in ~194000 objects): Zope tends to "eat up" all available memory (640MB in my case) and swap space. The main problem seem to be that the classes Zcatalog, Catalog, UnTextIndex and [Globbing}Lexicon try to do all partial indexing tasks at once: The Catalog instance collects meta data for each indexed object; the UnTextIndex instance adds new words to the Lexicon instance, builds an entry for each indexed object in _unindex and - most important - updates _index "just in time". The lexicon contains more than 1 million entries for "my" data, and the _index entries for common words may contain a considerable number of references to indexed object. I am relatively new to Python and Zope, so I don't have any idea, how much memory is required for a reference to an indexed object in UnTextIndex._index. But assuming 10 bytes for each reference, and assuming that on average each indexed word/Lexicon entry occurs in 20 objects, the _index object will require around 200 MB, if it is kept completely in RAM. This can be easily avoided, if the process of updating UnTextIndex._index is separated from the other indexing stuff. With the attached patch, UnTextIndex.index_object writes the data for new _index entries as lines containing WordId, DocumentId, score, into a pipe which is connected to Good Old sort(1). The output from sort is read by the new UnTextIndex method updateWordBTree. This way, each entry of UnTextIndex._index needs to be updated only once, while the original implementation of UnTextIndex required an update for each occurence of a word in every indexed object. There are two obvious points, where the patch can be optimized. (1) the "perfectly sorted" list returned by sort(1) is obviously not very efficient for building the IIBTree indexRow in updateWordBTree, and (2) the type check for indexRow in updateWordBTree can of course be avoided, after the type has been set to IIBTree. (updateWordBTree is simply modified version of insertForwardIndexEntry -- not very clever but most easy to implement :) Abel PS: Watching Zope indexing the above mentioned data with subtransaction enabled, I wondered, where all the data from the finished subtransactions is stored. Data.fs is not updated, and I could not find any temporary files. (ok, I had not have yet the chance to look what's going on during, while updateWordBTree is running; during my tests, it was called in the middle of the night -- but while ZCatalog.ZopeFindAndApply is running, nothing seems to hallen) --- Catalog.py.orig Thu Apr 19 19:07:29 2001 +++ Catalog.py Fri May 25 13:46:16 2001 @@ -105,6 +105,7 @@ from SearchIndex.randid import randid import time +import popen2 def orify(seq, query_map={ @@ -379,8 +380,24 @@ self.indexes = indexes # the cataloging API + + def openPipes(self): + pipes={} + for name in self.indexes.keys(): + index = self.indexes[name] + index = index.__of__(self) + if hasattr(index, 'updateWordBTree'): + pipes[name] = popen2.Popen3("sort", 1) + return pipes + + def updateWordBTrees(self, pipes, threshold=None): + for name in self.indexes.keys(): + index = self.indexes[name] + index = index.__of__(self) + if hasattr(index, 'updateWordBTree'): + index.updateWordBTree(pipes[name], threshold) - def catalogObject(self, object, uid, threshold=None): + def catalogObject(self, object, uid, threshold=None, pipes=None): """ Adds an object to the Catalog by iteratively applying it all indexes. @@ -439,17 +456,34 @@ self.paths[index] = uid total = 0 - for x in self.indexes.values(): - ## tricky! indexes need to acquire now, and because they - ## are in a standard dict __getattr__ isn't used, so - ## acquisition doesn't kick in, we must explicitly wrap! - x = x.__of__(self) - if hasattr(x, 'index_object'): - blah = x.index_object(index, object, threshold) - total = total + blah - else: - LOG('Catalog', ERROR, ('catalogObject was passed ' - 'bad index object %s.' % str(x))) + if pipes is None: + for x in self.indexes.values(): + ## tricky! indexes need to acquire now, and because they + ## are in a standard dict __getattr__ isn't used, so + ## acquisition doesn't kick in, we must explicitly wrap! + x = x.__of__(self) + if hasattr(x, 'index_object'): + blah = x.index_object(index, object, threshold) + total = total + blah + else: + LOG('Catalog', ERROR, ('catalogObject was passed ' + 'bad index object %s.' % str(x))) + else: + for name in self.indexes.keys(): + x = self.indexes[name] + ## tricky! indexes need to acquire now, and because they + ## are in a standard dict __getattr__ isn't used, so + ## acquisition doesn't kick in, we must explicitly wrap! + x = x.__of__(self) + if hasattr(x, 'index_object'): + if pipes.has_key(name): + blah = x.index_object(index, object, threshold, pipes[name]) + else: + blah = x.index_object(index, object, threshold) + total = total + blah + else: + LOG('Catalog', ERROR, ('catalogObject was passed ' + 'bad index object %s.' % str(x))) return total --- UnTextIndex.py.orig Wed Mar 28 01:43:11 2001 +++ UnTextIndex.py Fri May 25 14:01:46 2001 @@ -305,7 +305,7 @@ # put our first entry in, and use a tuple to save space index[entry] = (documentId, score) - def index_object(self, documentId, obj, threshold=None): + def index_object(self, documentId, obj, threshold=None, pipe=None): """ Index an object: 'documentId' is the integer id of the document @@ -356,9 +356,14 @@ # Now index the words. Note that the new xIBTrees are clever # enough to do nothing when there isn't a change. Woo hoo. - insert=self.insertForwardIndexEntry - for wid, score in widScores.items(): - insert(wid, documentId, score) + if pipe is None: + insert=self.insertForwardIndexEntry + for wid, score in widScores.items(): + insert(wid, documentId, score) + else: + write = pipe.tochild.write + for wid, score in widScores.items(): + write(`wid` + " " + `documentId` + " " + `score` + "\n") # Save the unindexing info if it's changed: wids=widScores.keys() @@ -367,6 +372,80 @@ return len(wids) + def updateWordBTree(self, pipe, threshold=None): + pipe.tochild.close() + readline = pipe.fromchild.readline + print "waiting for pipe input" + + index = self._index + + s = readline() + lastentry = None + count = 0 + + while len(s) > 4: + (entry, documentId, score) = s.split(' ') + entry = int(entry) + documentId = int(documentId) + score = int(score) + + # copied from insertForwardIndexEntry: + if entry != lastentry: + indexRow = index.get(entry, None) + if threshold is not None: + count = count + 1 + if count >= threshold: + get_transaction().commit(1) + self._p_jar.cacheFullSweep(3) + print "commit indexer", s, + count = 0 + + if indexRow is not None: + if type(indexRow) is TupleType: + # Tuples are only used for rows which have only + # a single entry. Since we now need more, we'll + # promote it to a mapping object (dictionary). + + # First, make sure we're not already in it, if so + # update the score if necessary. + if indexRow[0] == documentId: + if indexRow[1] != score: + indexRow = (documentId, score) + index[entry] = indexRow + else: + indexRow={ + indexRow[0]: indexRow[1], + documentId: score, + } + index[entry] = indexRow + else: + # xxx optimization: put a "readline loop" + # into _this_ block!! + if indexRow.get(documentId, -1) != score: + # score changed (or new entry) + + if type(indexRow) is DictType: + indexRow[documentId] = score + if len(indexRow) > 3: + # Big enough to give it's own database record + indexRow=IIBTree(indexRow) + index[entry] = indexRow + else: + indexRow[documentId] = score + else: + # We don't have any information at this point, so we'll + # put our first entry in, and use a tuple to save space + indexRow = index[entry] = (documentId, score) + + lastentry = entry + s = readline() + + pipe.fromchild.close() + res = pipe.wait() + # xxx I'm new to Python and Zope... How should an error + # be handled? raise an exception? + return + def _subindex(self, source, wordScores, last, splitter): """Recursively handle multi-word synonyms""" for word in splitter(source): --- ZCatalog.py.orig Wed Mar 21 23:48:04 2001 +++ ZCatalog.py Fri May 25 13:48:04 2001 @@ -359,6 +359,7 @@ obj = REQUEST.PARENTS[1] path = string.join(obj.getPhysicalPath(), '/') + pipelist = self._catalog.openPipes() results = self.ZopeFindAndApply(obj, obj_metatypes=obj_metatypes, @@ -372,8 +373,11 @@ search_sub=1, REQUEST=REQUEST, apply_func=self.catalog_object, - apply_path=path) + apply_path=path, + pipes=pipelist) + self._catalog.updateWordBTrees(pipelist, self.threshold) + elapse = time.time() - elapse c_elapse = time.clock() - c_elapse @@ -412,7 +416,7 @@ RESPONSE.redirect(URL1 + '/manage_catalogIndexes?manage_tabs_message=Index%20Deleted') - def catalog_object(self, obj, uid=None): + def catalog_object(self, obj, uid=None, pipes=None): """ wrapper around catalog """ if uid is None: @@ -426,13 +430,12 @@ elif type(uid) is not StringType: raise CatalogError('The object unique id must be a string.') - self._catalog.catalogObject(obj, uid, None) + self._catalog.catalogObject(obj, uid, None, pipes) # None passed in to catalogObject as third argument indicates # that we shouldn't try to commit subtransactions within any # indexing code. We throw away the result of the call to # catalogObject (which is a word count), because it's # worthless to us here. - if self.threshold is not None: # figure out whether or not to commit a subtransaction. t = id(get_transaction()) @@ -449,6 +452,7 @@ # to commit a subtransaction. The semantics here mean that # we should commit a subtransaction if our threshhold is # exceeded within the boundaries of the current transaction. + if self._v_total > self.threshold: get_transaction().commit(1) self._p_jar.cacheFullSweep(3) @@ -574,7 +578,7 @@ obj_permission=None, obj_roles=None, search_sub=0, REQUEST=None, result=None, pre='', - apply_func=None, apply_path=''): + apply_func=None, apply_path='', pipes=None): """Zope Find interface and apply This is a *great* hack. Zope find just doesn't do what we @@ -649,7 +653,7 @@ ) ): if apply_func: - apply_func(ob, (apply_path+'/'+p)) + apply_func(ob, (apply_path+'/'+p), pipes) else: add_result((p, ob)) dflag=0 @@ -661,7 +665,7 @@ obj_permission, obj_roles, search_sub, REQUEST, result, p, - apply_func, apply_path) + apply_func, apply_path, pipes) if dflag: ob._p_deactivate() return result