afcaafd51858229f8ac153c535493c1711717cef
[simantics/platform.git] / bundles / org.simantics.db.indexing / src / org / simantics / db / indexing / IndexedRelationsSearcherBase.java
1 /*******************************************************************************
2  * Copyright (c) 2007, 2015 Association for Decentralized Information Management
3  * in Industry THTH ry.
4  * All rights reserved. This program and the accompanying materials
5  * are made available under the terms of the Eclipse Public License v1.0
6  * which accompanies this distribution, and is available at
7  * http://www.eclipse.org/legal/epl-v10.html
8  *
9  * Contributors:
10  *     VTT Technical Research Centre of Finland - initial API and implementation
11  *     Semantum Oy - Fix for simantics issue #6053
12  *******************************************************************************/
13 package org.simantics.db.indexing;
14
15 import java.io.IOException;
16 import java.nio.file.Files;
17 import java.nio.file.Path;
18 import java.util.ArrayList;
19 import java.util.Collection;
20 import java.util.Collections;
21 import java.util.Iterator;
22 import java.util.List;
23 import java.util.Map;
24 import java.util.concurrent.CompletableFuture;
25 import java.util.concurrent.Semaphore;
26 import java.util.concurrent.atomic.AtomicReference;
27
28 import org.apache.lucene.document.Document;
29 import org.apache.lucene.document.DocumentStoredFieldVisitor;
30 import org.apache.lucene.document.Field;
31 import org.apache.lucene.document.FieldType;
32 import org.apache.lucene.document.LongField;
33 import org.apache.lucene.document.TextField;
34 import org.apache.lucene.index.CorruptIndexException;
35 import org.apache.lucene.index.DirectoryReader;
36 import org.apache.lucene.index.FieldInfo;
37 import org.apache.lucene.index.IndexNotFoundException;
38 import org.apache.lucene.index.IndexReader;
39 import org.apache.lucene.index.IndexWriter;
40 import org.apache.lucene.index.IndexWriterConfig;
41 import org.apache.lucene.index.IndexWriterConfig.OpenMode;
42 import org.apache.lucene.index.IndexableField;
43 import org.apache.lucene.index.StoredFieldVisitor;
44 import org.apache.lucene.index.Term;
45 import org.apache.lucene.queryparser.classic.ParseException;
46 import org.apache.lucene.search.IndexSearcher;
47 import org.apache.lucene.search.MatchAllDocsQuery;
48 import org.apache.lucene.search.Query;
49 import org.apache.lucene.search.ScoreDoc;
50 import org.apache.lucene.search.TermQuery;
51 import org.apache.lucene.search.TopDocs;
52 import org.apache.lucene.store.Directory;
53 import org.apache.lucene.store.FSDirectory;
54 import org.apache.lucene.util.Version;
55 import org.eclipse.core.runtime.IProgressMonitor;
56 import org.eclipse.core.runtime.SubMonitor;
57 import org.simantics.databoard.util.ObjectUtils;
58 import org.simantics.db.ReadGraph;
59 import org.simantics.db.RequestProcessor;
60 import org.simantics.db.Resource;
61 import org.simantics.db.Session;
62 import org.simantics.db.common.request.SafeName;
63 import org.simantics.db.common.utils.NameUtils;
64 import org.simantics.db.exception.DatabaseException;
65 import org.simantics.db.indexing.exception.IndexCorruptedException;
66 import org.simantics.db.indexing.exception.IndexingException;
67 import org.simantics.db.indexing.internal.IndexingJob;
68 import org.simantics.db.layer0.adapter.GenericRelation;
69 import org.simantics.db.layer0.genericrelation.IndexException;
70 import org.simantics.db.request.Read;
71 import org.simantics.db.service.CollectionSupport;
72 import org.simantics.db.service.SerialisationSupport;
73 import org.simantics.utils.FileUtils;
74 import org.simantics.utils.datastructures.Pair;
75 import org.simantics.utils.threads.ThreadUtils;
76 import org.slf4j.Logger;
77
78 import gnu.trove.map.hash.THashMap;
79
80 /**
81  * @author Tuukka Lehtonen
82  * @author Antti Villberg
83  */
84 abstract public class IndexedRelationsSearcherBase {
85
86     protected enum State {
87         // No index is available
88         NONE, 
89         // An index is available, but there is a problem with it
90         PROBLEM, 
91         // An index is available but no reader or writer is ready
92         READY,
93         // A reader is ready
94         READ, 
95         // A writer (and a reader) is ready
96         WRITE
97     }
98     
99     private State state = State.READY;
100     private Throwable exception;
101     
102     public Throwable getException() {
103         return exception;
104     }
105
106     public void setProblem(Throwable t) {
107         if (t != null)
108             getLogger().error("Setting problem for {} and previous state {}", this, this.state, t);
109         this.state = State.PROBLEM;
110         this.exception = t;
111     }
112     
113     public void setNone() {
114         this.state = State.NONE;
115     }
116     
117     public void setReady() {
118         this.state = State.READY;
119     }
120     
121     public State state() {
122         return state;
123     }
124     
125     protected boolean checkState(State state) {
126         return this.state == state;
127     }
128     
129     protected void assertState(State state) throws IndexException {
130         State s = this.state;
131         if (s != state)
132             throw new IndexException("Illegal index searcher state, expected " + state.name() + " but state was " + s.name());
133     }
134     
135     public void changeState(IProgressMonitor monitor, Session session, State state) {
136         changeState(monitor, session, state, 0);
137     }
138
139     protected void changeState(IProgressMonitor monitor, Session session, State state, int depth) {
140
141         if (this.state == state) {
142             if (getLogger().isDebugEnabled())
143                 getLogger().debug("Trying to change state {} to the same as previous state {} in depth {} with {}", state, this.state, depth, this);
144             return;
145         }
146
147         if (IndexPolicy.TRACE_INDEX_MANAGEMENT)
148                 System.err.println("Index state " + this.state.name() + " => " + state.name() + " " + this);
149
150         // Check transitions
151         
152         // Try to exit problem state
153         if (State.PROBLEM == this.state && depth > 0) {
154             getLogger().info("Try to exit problem state for {} and state {}", this, state);
155                 Throwable t = bestEffortClear(monitor, session);
156                 if(t != null) {
157                     getLogger().error("Best effort clear has failed for state {} and this {}", state, this, t);
158                                 exception = t;
159                                 return;
160                 }
161                 // Managed to get into initial state
162                 this.state = State.NONE;
163                 getLogger().info("Managed to get into initial state {}", this.state);
164                 return;
165         }
166
167         // Cannot move into read from no index
168         if (State.NONE ==  this.state && State.READ == state) {
169             if (getLogger().isDebugEnabled())
170                 getLogger().debug("Cannot move into read from no index in {} with state {}", this, state);
171             return;
172         }
173         // Cannot move into write from no index
174         if (State.NONE ==  this.state && State.WRITE == state) {
175             if (getLogger().isDebugEnabled())
176                 getLogger().debug("Cannot move into write from no index in {} with state {}", this, state);
177             return;
178         }
179         
180                 boolean success = false;
181
182         try {
183
184                 if (searcher != null) {
185                         searcher = null;
186                 }
187                 if (reader != null) {
188                         reader.close();
189                         reader = null;
190                 }
191                         closeWriter(writer);
192                         directory = null;
193                         
194                         success = true;
195
196                 // Enter new state
197                 if (State.READ == state || State.WRITE == state) {
198                         
199                         success = false;
200                         
201                         boolean forWriting = State.WRITE == state;
202
203                         if (directory != null)
204                                 throw new IllegalStateException(getDescriptor() + "Index already loaded");
205
206                         SubMonitor mon = SubMonitor.convert(monitor, 100);
207
208                         mon.beginTask("Loading index", 100);
209
210                         if (IndexPolicy.TRACE_INDEX_LOAD)
211                     System.out.println(getDescriptor() + "Loading Lucene index from " + indexPath + " for " + (forWriting ? "writing" : "reading"));
212
213                 long start = System.nanoTime();
214
215                 directory = getDirectory(session);
216
217                 if (forWriting) {
218                     // Never overwrite an index that is about to be loaded.
219                     // TODO: could use OpenMode.CREATE_OR_APPEND but must test first
220                     IndexWriterConfig config = new IndexWriterConfig(Version.LUCENE_4_9, Queries.getAnalyzer()).setOpenMode(OpenMode.APPEND);
221                     try {
222                         // FIXME: platform #4676
223                         writer = new IndexWriter(directory, config);
224                     } catch (IndexNotFoundException e) {
225                         // There was no pre-existing index on disk. Create it now.
226                         writer = new IndexWriter(directory, config.setOpenMode(OpenMode.CREATE));
227                         writer.commit();
228                     }
229                     reader = DirectoryReader.open(directory);
230                     searcher = new IndexSearcher(reader);
231                 } else {
232                     reader = DirectoryReader.open(directory);
233                     searcher = new IndexSearcher(reader);
234                 }
235
236                 long end = System.nanoTime();
237
238                 mon.worked(100);
239
240                 if (IndexPolicy.PERF_INDEX_LOAD) {
241                     double time = (end - start) * 1e-6;
242                     System.out.println(getDescriptor() + "Loaded Lucene index from " + indexPath + " for " + (forWriting ? "writing" : "reading") + " in " + time + " ms");
243                 }
244
245                 success = true;
246                 
247                 }
248                 
249         } catch (Throwable t) {
250                 setProblem(t);
251         } finally {
252
253                 if(!success) {
254                         this.state = State.PROBLEM;
255                         changeState(monitor, session, State.NONE, depth+1);
256                         return;
257                 }
258
259         }
260
261         this.state = state;
262         
263     }
264
265     public static final FieldType STRING_TYPE = new FieldType();
266
267     static {
268       STRING_TYPE.setIndexed(true);
269       STRING_TYPE.setStored(true);
270       STRING_TYPE.setTokenized(true);
271       STRING_TYPE.freeze();
272     }
273
274     protected static Field makeField(String fieldName, String fieldClass) throws IndexingException {
275         switch (fieldClass) {
276         case "Long":   return new LongField(fieldName, 0L, Field.Store.YES);
277         case "String": return new Field    (fieldName, "", STRING_TYPE);
278         case "Text":   return new TextField(fieldName, "", Field.Store.YES);
279         default:
280             throw new IndexingException("Can only index Long, String and Text fields, encountered field type " + fieldClass);
281         }
282     }
283
284     protected static Field[] makeFieldsForRelation(GenericRelation r, int boundLength, Document document) throws DatabaseException {
285         Pair<String, String>[] fields = r.getFields();
286         Field[] fs = new Field[Math.max(0, fields.length - boundLength)];
287         for (int i = boundLength; i < fields.length; i++) {
288             Field f = makeField(fields[i].first, fields[i].second);
289             fs[i - boundLength] = f;
290             if (document != null)
291                 document.add(f);
292         }
293         return fs;
294     }
295
296     void insertIndex(IProgressMonitor monitor, GenericRelation r, int boundLength, Collection<Object[]> documentsData)
297     throws CorruptIndexException, IOException, DatabaseException {
298         assertAccessOpen(true);
299
300         if (IndexPolicy.TRACE_INDEX_UPDATE)
301             System.out.println(getDescriptor() + "Inserting " + documentsData.size() + " documents into index at " + indexPath);
302
303         long start = 0, end = 0;
304         if (IndexPolicy.PERF_INDEX_UPDATE)
305             start = System.nanoTime();
306
307         try {
308             Document document = new Document();
309             Field[] fs = makeFieldsForRelation(r, boundLength, document);
310
311             for (Object[] documentData : documentsData) {
312                 if (setFields(fs, documentData) == null)
313                     continue;
314
315                 if (IndexPolicy.TRACE_INDEX_UPDATE)
316                     System.out.println(getDescriptor() + "Inserting document " + document);
317
318                 writer.addDocument(document);
319             }
320
321             if (IndexPolicy.PERF_INDEX_UPDATE) {
322                 end = System.nanoTime();
323                 double ms = (end - start) * 1e-6;
324                 System.out.println(getDescriptor() + "Inserted " + documentsData.size() + " documents into index at " + indexPath + " in " + ms + " ms");
325             }
326
327         } finally {
328         }
329     }
330
331     void removeIndex(IProgressMonitor monitor, GenericRelation r, RequestProcessor processor, String key, Collection<Object> keyValues) throws DatabaseException, CorruptIndexException, IOException {
332         assertAccessOpen(true);
333
334         if (IndexPolicy.TRACE_INDEX_UPDATE)
335             System.out.println(getDescriptor() + "Removing " + keyValues.size() + " documents from index at " + indexPath);
336
337         long start = 0, end = 0;
338         if (IndexPolicy.PERF_INDEX_UPDATE)
339             start = System.nanoTime();
340
341         try {
342             for (Object keyValue : keyValues) {
343                 Term removedTerm = null;
344                 if (keyValue instanceof Long) {
345                     removedTerm = IndexUtils.longTerm(key, (Long) keyValue);
346                 } else if (keyValue instanceof String) {
347                     removedTerm = new Term(key, (String) keyValue);
348                 } else {
349                     // FIXME: should throw an exception for illegal input data but this would leave the index in an incoherent state
350                     continue;
351                 }
352
353                 if (IndexPolicy.TRACE_INDEX_UPDATE)
354                     System.out.println(getDescriptor() + "Removing document with key " + removedTerm);
355                 writer.deleteDocuments(removedTerm);
356             }
357
358             if (IndexPolicy.PERF_INDEX_UPDATE) {
359                 end = System.nanoTime();
360                 double ms = (end - start) * 1e-6;
361                 System.out.println(getDescriptor() + "Removed " + keyValues.size() + " documents from index at " + indexPath + " in " + ms + " ms");
362             }
363
364         } finally {
365         }
366     }
367
368     void removeIndex(IProgressMonitor monitor) throws DatabaseException, CorruptIndexException, IOException {
369         assertAccessOpen(true);
370
371         long start = 0, end = 0;
372         if (IndexPolicy.PERF_INDEX_UPDATE)
373             start = System.nanoTime();
374
375         try {
376
377             writer.deleteAll();
378
379             if (IndexPolicy.PERF_INDEX_UPDATE) {
380                 end = System.nanoTime();
381                 double ms = (end - start) * 1e-6;
382                 System.out.println(getDescriptor() + "Removed all documents from index at " + indexPath + " in " + ms + " ms");
383             }
384
385         } finally {
386         }
387     }
388     
389     boolean replaceIndex(IProgressMonitor monitor, String key, Collection<Object> keyValues, GenericRelation r, int boundLength, Collection<Object[]> documentsData) throws CorruptIndexException, IOException, DatabaseException {
390
391         boolean didReplace = false;
392         
393         assertAccessOpen(true);
394         if (keyValues.size() != documentsData.size())
395             throw new IllegalArgumentException("keyValues size does not match documents data size, " + keyValues.size() + " <> " + documentsData.size());
396
397         if (IndexPolicy.TRACE_INDEX_UPDATE)
398             System.out.println(getDescriptor() + "Replacing " + keyValues.size() + " documents from index at " + indexPath);
399
400         long start = 0, end = 0;
401         if (IndexPolicy.PERF_INDEX_UPDATE)
402             start = System.nanoTime();
403
404         try {
405             Iterator<Object> keyIt = keyValues.iterator();
406             Iterator<Object[]> documentDataIt = documentsData.iterator();
407
408             Document document = new Document();
409             Field[] fs = makeFieldsForRelation(r, boundLength, document);
410
411             nextDocument:
412                 while (keyIt.hasNext()) {
413                     Object keyValue = keyIt.next();
414                     Object[] documentData = documentDataIt.next();
415
416                     Term removedTerm = null;
417                     if (keyValue instanceof Long) {
418                         removedTerm = IndexUtils.longTerm(key, (Long) keyValue);
419                     } else if (keyValue instanceof String) {
420                         removedTerm = new Term(key, (String) keyValue);
421                     } else {
422                         // FIXME: should throw an exception for illegal input data but this would leave the index in an incoherent state
423                         System.err.println("[" + getClass().getSimpleName() + "] Unrecognized document key to remove '" + keyValue + "', only " + String.class + " and " + Resource.class + " are supported.");
424                         continue nextDocument;
425                     }
426
427                     if (setFields(fs, documentData) == null)
428                         continue nextDocument;
429
430                     if (IndexPolicy.TRACE_INDEX_UPDATE)
431                         System.out.println(getDescriptor() + "Replacing document with key " + removedTerm + " with " + document);
432
433                     boolean done = false;
434                     if(requireChangeInfoOnReplace()) {
435                             TopDocs exist = searcher.search(new TermQuery(removedTerm), null, 2);
436                             if(exist.scoreDocs.length == 1 && requireChangeInfoOnReplace()) {
437                                 Document doc = reader.document(exist.scoreDocs[0].doc);
438                                 if(!areSame(doc, document)) {
439                                     writer.deleteDocuments(removedTerm);
440                                     writer.addDocument(document);
441                                     didReplace |= true;
442                                     if (IndexPolicy.TRACE_INDEX_UPDATE)
443                                         System.out.println("-replaced single existing");
444                                 } else {
445                                     if (IndexPolicy.TRACE_INDEX_UPDATE)
446                                         System.out.println("-was actually same than single existing");
447                                 }
448                                 done = true;
449                             } 
450                     }
451                     if(!done) {
452                         writer.deleteDocuments(removedTerm);
453                         writer.addDocument(document);
454                         didReplace |= true;
455                         if (IndexPolicy.TRACE_INDEX_UPDATE)
456                                 System.out.println("-had many or none - removed all existing");
457                     }
458                     
459                 }
460
461             if (IndexPolicy.PERF_INDEX_UPDATE) {
462                 end = System.nanoTime();
463                 double ms = (end - start) * 1e-6;
464                 System.out.println(getDescriptor() + "Replaced " + keyValues.size() + " documents from index at " + indexPath + " in " + ms + " ms");
465             }
466
467         } finally {
468         }
469         
470         return didReplace;
471         
472     }
473     
474     protected boolean requireChangeInfoOnReplace() {
475         return true;
476     }
477     
478     private boolean areSame(Document d1, Document d2) {
479         List<IndexableField> fs1 = d1.getFields();
480         List<IndexableField> fs2 = d2.getFields();
481         if(fs1.size() != fs2.size()) return false;
482         for(int i=0;i<fs1.size();i++) {
483                 IndexableField f1 = fs1.get(i);
484                 IndexableField f2 = fs2.get(i);
485                 String s1 = f1.stringValue();
486                 String s2 = f2.stringValue();
487             if (IndexPolicy.TRACE_INDEX_UPDATE)
488                 System.err.println("areSame " + s1 + " vs. " + s2 );
489                 if(!ObjectUtils.objectEquals(s1,s2)) return false;
490         }
491         return true;
492     }
493
494     final RequestProcessor session;
495
496     final Resource         relation;
497
498     /**
499      * The schema of the index, i.e. the fields that will be indexed per
500      * document for the specified relation. Since the relation stays the same
501      * throughout the lifetime of this class, the index schema is also assumed
502      * to the same. This means that {@link GenericRelation#getFields()} is
503      * assumed to stay the same.
504      */
505     final IndexSchema      schema;
506
507     Resource         input;
508
509     Path             indexPath;
510
511     Directory        directory;
512
513     IndexReader      reader;
514
515     IndexWriter      writer;
516
517     IndexSearcher    searcher;
518
519     IndexedRelationsSearcherBase(RequestProcessor session, Resource relation, Resource input) {
520         this.session = session;
521         this.relation = relation;
522         this.input = input;
523         this.indexPath = getIndexDirectory(session.getSession(), relation, input);
524         if(isIndexAvailable()) {
525                 state = State.READY;
526         } else {
527                 state = State.NONE;
528         }
529         this.schema = IndexSchema.readFromRelation(session, relation);
530     }
531
532     Directory getDirectory(Session session) throws IOException {
533         return FSDirectory.open(indexPath.toFile());
534     }
535
536     abstract String getDescriptor();
537     
538     /**
539      * Ensures that searcher is in read or write state.
540      * 
541      * @param forWriting <code>true</code> to open index for writing,
542      *        <code>false</code> for reading
543      * @return true is required state was reached       
544      *        
545      */
546     boolean startAccess(IProgressMonitor monitor, Session session, boolean forWriting) {
547         if(forWriting) {
548                 changeState(monitor, session, State.WRITE);
549                 return checkState(State.WRITE);
550         } else {
551                 changeState(monitor, session, State.READ);
552                 return checkState(State.READ);
553         }
554     }
555
556     boolean hasAccess(boolean forWriting) {
557         
558         if (forWriting)
559                 return checkState(State.WRITE); 
560         else
561                 return checkState(State.WRITE) || checkState(State.READ);
562         
563     }
564     
565     void assertAccessOpen(boolean forWriting) {
566         if (forWriting)
567                 if(!checkState(State.WRITE)) 
568                 throw new IllegalStateException("index not opened for writing (directory=" + directory + ", reader=" + reader + ")");
569         else
570                 if(!(checkState(State.WRITE) || checkState(State.READ))) 
571                 throw new IllegalStateException("index not opened for reading (directory=" + directory + ", writer=" + writer + ")");
572     }
573     
574     void closeWriter(IndexWriter writer) throws CorruptIndexException, IOException {
575         if (writer == null)
576             return;
577
578         try {
579             // May throw OOME, see IndexWriter javadoc for the correct actions.
580             writer.close(false);
581         } catch (OutOfMemoryError e) {
582             writer.close();
583             throw e;
584         }
585     }
586
587     public static String getPattern(GenericRelation relation, int boundCount) {
588         String result = "";
589         for (int i = 0; i < boundCount; i++)
590             result += "b";
591         for (int i = 0; i < relation.getFields().length - boundCount; i++)
592             result += "f";
593         return result;
594     }
595
596     void initializeIndex(IProgressMonitor monitor, ReadGraph graph, Object[] bound, boolean overwrite)
597             throws IOException, DatabaseException
598     {
599         IndexingJob.jobifyIfPossible(
600                 monitor,
601                 "Reindexing " + NameUtils.getSafeLabel(graph, input),
602                 mon -> {
603                     try {
604                         GenericRelation r = graph.adapt(relation, GenericRelation.class);
605                         if (r == null)
606                             throw new IndexingException("Given resource " + relation + "could not be adapted to GenericRelation.");
607
608                         GenericRelation selection = r.select(getPattern(r, bound.length), bound);
609
610                         List<Object[]> results = selection.realize(graph);
611                         initializeIndexImpl(new CompletableFuture<>(), mon, r, results, bound, overwrite);
612                     } catch (IOException e) {
613                         getLogger().error("Index is in problematic state! {}", this, e);
614                         throw new IndexingException(e);
615                     }
616                 });
617     }
618
619     private static final int INDEXING_THREAD_COUNT = 2; // this is quite good parallelism level for lucene
620
621     void initializeIndexImpl(CompletableFuture<?> result, IProgressMonitor monitor, GenericRelation r, List<Object[]> results, final Object[] bound, boolean overwrite) throws IOException {
622         try {
623             final SubMonitor mon = SubMonitor.convert(monitor, 100);
624     
625             if (IndexPolicy.TRACE_INDEX_INIT)
626                 System.out.println(getDescriptor() + "Initializing index at " + indexPath + " (overwrite = " + overwrite + ")");
627             mon.beginTask("Initializing Index", 100);
628     
629             if (overwrite) {
630                 if (Files.exists(indexPath)) {
631                     mon.subTask("Erasing previous index");
632                     if (getLogger().isDebugEnabled())
633                         getLogger().debug("Erasing previous index {}", indexPath.toAbsolutePath());
634                     FileUtils.emptyDirectory(indexPath);
635                 }
636             }
637     
638             final AtomicReference<FSDirectory> directory = new AtomicReference<FSDirectory>();
639             final AtomicReference<IndexWriter> writer = new AtomicReference<IndexWriter>();
640     
641             try {
642                 mon.subTask("Start index write");
643                 Files.createDirectories(indexPath);
644     
645                 directory.set(FSDirectory.open(indexPath.toFile()));
646                 IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_4_9, Queries.getAnalyzer()).setOpenMode(OpenMode.CREATE);
647                 writer.set(new IndexWriter(directory.get(), conf));
648     
649                 mon.worked(5);
650     
651                 long realizeStart = 0;
652                 if (IndexPolicy.PERF_INDEX_INIT)
653                     realizeStart = System.nanoTime();
654     
655                 mon.subTask("Calculating indexed content");
656                 mon.worked(5);
657                 
658                 mon.worked(40);
659     
660                 if (IndexPolicy.PERF_INDEX_INIT)
661                     System.out.println(getDescriptor() + "Realized index with " + results.size() + " entries at " + indexPath + " in " + (1e-9 * (System.nanoTime()-realizeStart)) + " seconds.");
662     
663                 if (IndexPolicy.TRACE_INDEX_INIT)
664                     System.out.println(getDescriptor() + "Indexed relation " + r + " produced " + results.size() + " results");
665                 
666                 long start = IndexPolicy.PERF_INDEX_INIT ? System.nanoTime() : 0;
667     
668                 mon.subTask("Indexing content");
669                 final Semaphore s = new Semaphore(0);
670                 mon.setWorkRemaining(results.size());
671                 for (int i = 0; i < INDEXING_THREAD_COUNT; i++) {
672                     final int startIndex = i;
673                     ThreadUtils.getBlockingWorkExecutor().submit(() -> {
674                         try {
675                             Document document = new Document();
676                             Field[] fs = makeFieldsForRelation(r, bound.length, document);
677     
678                             for (int index = startIndex; index < results.size(); index += INDEXING_THREAD_COUNT) {
679                                 if (setFields(fs, results.get(index)) == null)
680                                     continue;
681                                 try {
682                                     writer.get().addDocument(document);
683                                 } catch (CorruptIndexException e) {
684                                     getLogger().error("Index is corrupted! {}", this, e);
685                                     throw new IllegalStateException(e);
686                                 } catch (IOException e) {
687                                     getLogger().error("Index is in problematic state! {}", this, e);
688                                     throw new IllegalStateException(e);
689                                 } finally {
690                                     synchronized (mon) {
691                                         mon.worked(1);
692                                     }
693                                 }
694                             }
695                         } catch (DatabaseException e) {
696                             throw new IllegalStateException(e);
697                         } finally {
698                             s.release();
699                         }
700                     });
701                 }
702     
703                 try {
704                     s.acquire(INDEXING_THREAD_COUNT);
705                 } catch (InterruptedException e) {
706                     getLogger().error("Could not initialize index {}", this, e);
707                 }
708     
709                 // http://www.gossamer-threads.com/lists/lucene/java-dev/47895
710                 // and http://lucene.apache.org/java/docs/index.html#27+November+2011+-+Lucene+Core+3.5.0
711                 // advise against calling optimize at all. So let's not do it anymore.
712                 //writer.get().optimize();
713                 //writer.get().commit();
714     
715                 mon.subTask("Flushing");
716     
717                 if (IndexPolicy.PERF_INDEX_INIT)
718                     System.out.println(getDescriptor() + "Wrote index at " + indexPath + " in " + (1e-9 * (System.nanoTime()-start)) + " seconds.");
719                 
720                 result.complete(null);
721     //        } catch (DatabaseException e) {
722     //            getLogger().error("Could not initialize index due to db {}", this, e);
723             } finally {
724                 try {
725                     closeWriter(writer.getAndSet(null));
726                 } finally {
727                     FileUtils.uncheckedClose(directory.getAndSet(null));
728                 }
729             }
730         } catch (Throwable t) {
731             getLogger().error("Could not initialize index", t);
732             result.completeExceptionally(t);
733         }
734     }
735
736     
737     public List<Object[]> debugDocs(IProgressMonitor monitor) throws ParseException, IOException, IndexingException {
738     
739             Query query = new MatchAllDocsQuery(); 
740         
741             TopDocs td = searcher.search(query, Integer.MAX_VALUE);
742     
743             ScoreDoc[ ] scoreDocs = td.scoreDocs; 
744             List<Object[]> result = new ArrayList<Object[]>(scoreDocs.length);
745         
746             for(ScoreDoc scoreDoc:scoreDocs) {
747         
748                 try {
749         
750                     Document doc = reader.document(scoreDoc.doc);
751                     List<IndexableField> fs = doc.getFields();
752                     Object[] o = new Object[fs.size()];
753                     int index = 0; 
754                     for (IndexableField f : fs) {
755                     o[index++] = f.stringValue();
756                     }
757                     result.add(o);
758         
759             } catch (CorruptIndexException e) {
760                 getLogger().error("Index is corrupted! {}", this, e);
761                 throw new IndexCorruptedException("Index is corrupted! " + this, e);
762             } catch (IOException e) {
763                 getLogger().error("Index is in problematic state! {}", this, e);
764                 throw new IndexingException(e);
765             }
766
767             }
768             
769             return result;
770             
771     }
772
773     
774     List<Map<String, Object>> doSearch(IProgressMonitor monitor, RequestProcessor processor, String search, int maxResultCount) throws ParseException, IOException,
775     IndexingException {
776
777         // An empty search string will crash QueryParser
778         // Just return no results for empty queries.
779         //System.out.println("search: '" + search + "'");
780         if (search.isEmpty())
781             return Collections.emptyList();
782
783         assertAccessOpen(false);
784
785         Query query = Queries.parse(search, schema);
786
787         long start = System.nanoTime();
788
789         maxResultCount = Math.min(maxResultCount, searcher.getIndexReader().numDocs());
790         if (maxResultCount == 0)
791             return Collections.emptyList();
792         
793         final TopDocs docs = searcher.search(query, null, maxResultCount);
794         
795 //        for(Object[] o : debugDocs(monitor)) {
796 //            System.err.println("-" + Arrays.toString(o));
797 //        }
798         
799         if (IndexPolicy.PERF_INDEX_QUERY) {
800             long end = System.nanoTime();
801             System.out.println(getDescriptor() + "search(" + search + ", " + maxResultCount + ") into index at " + indexPath + " took " + (1e-9 * (end-start)) + " seconds.");
802         }
803
804         if (docs.totalHits == 0) {
805             return Collections.emptyList();
806         }
807
808         try {
809             return processor.syncRequest(new Read<List<Map<String, Object>>>() {
810
811                 @Override
812                 public List<Map<String, Object>> perform(ReadGraph graph) throws DatabaseException {
813
814                     GenericRelation r = graph.adapt(relation, GenericRelation.class);
815                     if (r == null)
816                         throw new IndexingException("Given resource " + graph.syncRequest(new SafeName(relation))
817                                 + "could not be adapted to GenericRelation.");
818
819                     SerialisationSupport support = graph.getService(SerialisationSupport.class);
820
821                     List<Map<String, Object>> result = new ArrayList<Map<String, Object>>(docs.scoreDocs.length);
822                     
823                     final DocumentStoredFieldVisitor visitor = new DocumentStoredFieldVisitor();
824                     
825                     for (ScoreDoc scoreDoc : docs.scoreDocs) {
826
827                         try {
828
829                             reader.document(scoreDoc.doc, visitor);
830                             
831                             Document doc = visitor.getDocument();
832
833                             List<IndexableField> fs = doc.getFields();
834                             Map<String, Object> entry = new THashMap<String, Object>(fs.size());
835                             for (IndexableField f : fs) {
836                                 IndexSchema.Type type = schema.typeMap.get(f.name());
837                                 if (type == IndexSchema.Type.LONG) {
838                                     entry.put(f.name(), support.getResource(f.numericValue().longValue()));
839                                 } else {
840                                     entry.put(f.name(), f.stringValue());
841                                 }
842                             }
843                             
844                             result.add(entry);
845
846                         } catch (CorruptIndexException e) {
847                             getLogger().error("Index is corrupted! {}", this, e);
848                             throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
849                         } catch (IOException e) {
850                             getLogger().error("Index is in problematic state! {}", this, e);
851                             throw new IndexingException(e);
852                         }
853                     }
854                     return result;
855                 }
856             });
857         } catch (DatabaseException e) {
858             if (e instanceof IndexingException) {
859                 throw (IndexingException) e;
860             } else {
861                 throw new IndexingException(e);
862             }
863         }
864     }
865
866     static class ResourceVisitor extends StoredFieldVisitor {
867         
868         public long id;
869
870                 @Override
871                 public Status needsField(FieldInfo fieldInfo) throws IOException {
872                         if("Resource".equals(fieldInfo.name)) return Status.YES;
873                         return Status.NO;
874                 }
875                 
876                 @Override
877                 public void longField(FieldInfo fieldInfo, long value) throws IOException {
878                         id = value;
879                 }
880         
881     };
882     
883     static class DumpVisitor extends StoredFieldVisitor {
884
885         public List<Object> values;
886         
887         DumpVisitor(List<Object> values) {
888                 this.values = values;
889         }
890
891                 @Override
892                 public Status needsField(FieldInfo fieldInfo) throws IOException {
893                         return Status.YES;
894                 }
895                 
896                 @Override
897                 public void longField(FieldInfo fieldInfo, long value) throws IOException {
898                         values.add(value);
899                 }
900                 
901                 @Override
902                 public void stringField(FieldInfo fieldInfo, String value) throws IOException {
903                         values.add(value);
904                 }
905
906     }
907
908     List<Resource> doSearchResources(IProgressMonitor monitor, RequestProcessor processor, String search, int maxResultCount) throws ParseException, IOException,
909     IndexingException {
910
911         // An empty search string will crash QueryParser
912         // Just return no results for empty queries.
913         //System.out.println("search: '" + search + "'");
914         if (search.isEmpty())
915             return Collections.emptyList();
916
917         assertAccessOpen(false);
918
919         Query query = Queries.parse(search, schema);
920
921         long start = System.nanoTime();
922
923         maxResultCount = Math.min(maxResultCount, searcher.getIndexReader().numDocs());
924         if (maxResultCount == 0)
925             return Collections.emptyList();
926         
927         final TopDocs docs = searcher.search(query, null, maxResultCount);
928         
929 //        for(Object[] o : debugDocs(monitor)) {
930 //            System.err.println("-" + Arrays.toString(o));
931 //        }
932         
933         if (IndexPolicy.PERF_INDEX_QUERY) {
934             long end = System.nanoTime();
935             System.out.println(getDescriptor() + "search(" + search + ", " + maxResultCount + ") into index at " + indexPath + " took " + (1e-9 * (end-start)) + " seconds.");
936         }
937
938         if (docs.totalHits == 0) {
939             return Collections.emptyList();
940         }
941         
942         try {
943             return processor.syncRequest(new Read<List<Resource>>() {
944
945                 @Override
946                 public List<Resource> perform(ReadGraph graph) throws DatabaseException {
947
948                         CollectionSupport cs = graph.getService(CollectionSupport.class);
949                     SerialisationSupport support = graph.getService(SerialisationSupport.class);
950                     
951                         List<Resource> result = cs.createList();
952                     
953                         ResourceVisitor visitor = new ResourceVisitor();
954                     
955                     for (ScoreDoc scoreDoc : docs.scoreDocs) {
956                         try {
957                                 reader.document(scoreDoc.doc, visitor);
958                                 result.add(support.getResource(visitor.id));
959                         } catch (CorruptIndexException e) {
960                             getLogger().error("Index is corrupted! {}", this, e);
961                             throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
962                         } catch (IOException e) {
963                             getLogger().error("Index is in problematic state! {}", this, e);
964                             throw new IndexingException(e);
965                         }
966                     }
967                     return result;
968                 }
969             });
970         } catch (DatabaseException e) {
971             if (e instanceof IndexingException) {
972                 throw (IndexingException) e;
973             } else {
974                 throw new IndexingException(e);
975             }
976         }
977     }
978
979     List<Object> doList(IProgressMonitor monitor, RequestProcessor processor) throws ParseException, IOException,
980     IndexingException {
981
982         assertAccessOpen(false);
983
984         Query query = new MatchAllDocsQuery(); 
985
986         final TopDocs docs = searcher.search(query, Integer.MAX_VALUE);
987         
988         ArrayList<Object> result = new ArrayList<Object>();
989         
990         DumpVisitor visitor = new DumpVisitor(result);
991                 
992         for (ScoreDoc scoreDoc : docs.scoreDocs) {
993
994                 try {
995
996                         reader.document(scoreDoc.doc, visitor);
997
998                 } catch (CorruptIndexException e) {
999                     getLogger().error("Index is corrupted! {}", this, e);
1000                     throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
1001                 } catch (IOException e) {
1002                     getLogger().error("Index is in problematic state! {}", this, e);
1003                         throw new IndexingException(e);
1004                 }
1005
1006         }
1007
1008         return result;
1009
1010     }
1011     
1012     protected static Path getIndexDirectory(Session session, Resource relation, Resource input) {
1013         Path path = DatabaseIndexing.getIndexLocation(session, relation, input);
1014 //        System.out.println("getIndexDirectory = " + path);
1015         return path;
1016     }
1017
1018     Path getIndexPath() {
1019         return indexPath;
1020     }
1021
1022     boolean isIndexAvailable() {
1023         return Files.isDirectory(indexPath);
1024     }
1025
1026     abstract Throwable bestEffortClear(IProgressMonitor monitor, Session session);
1027
1028     /*
1029      * Start from scratch. Clear all caches and rebuild the index. 
1030      */
1031     Throwable clearDirectory(IProgressMonitor monitor, Session session) {
1032         
1033                 Path file = getIndexPath();
1034
1035         try {
1036                         FileUtils.delete(file);
1037         } catch (Throwable t) {
1038                 getLogger().error("Could not delete directory {}", file.toAbsolutePath(), t);
1039                 return t;
1040         }
1041         if (Files.exists(file))
1042             return new IllegalStateException("Failed to delete directory " + file.toAbsolutePath());
1043         return null;
1044     }
1045
1046     private Field[] setFields(Field[] fs, Object[] result) {
1047         for (int i = 0; i < result.length; i++) {
1048             Object value = result[i];
1049             if (value instanceof String) {
1050                 if (IndexPolicy.DEBUG_INDEX_INIT)
1051                     System.out.println(getDescriptor() + "index " + fs[i].name() + " = " + value + " : String");
1052                 fs[i].setStringValue((String) value);
1053             } else if (value instanceof Long) {
1054                 if (IndexPolicy.DEBUG_INDEX_INIT)
1055                     System.out.println(getDescriptor() + "index " + fs[i].name() + " = " + value + " : Long");
1056                 fs[i].setLongValue((Long) value);
1057             } else {
1058                 getLogger().error("Can only index Long and String fields, encountered " + value);
1059                 return null;
1060             }
1061         }
1062         return fs;
1063     }
1064
1065     protected abstract Logger getLogger();
1066     
1067     @Override
1068     public String toString() {
1069         return getClass().getSimpleName() + " [" + String.valueOf(schema) + ", " + String.valueOf(relation) + ", " + String.valueOf(input) + ", " + String.valueOf(indexPath) + ", " + String.valueOf(directory) + ", " + String.valueOf(state) + "]";
1070     }
1071 }