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