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                     getLogger().error("Attempting to remove document from index of {} with key {} and unrecognized key value type {} : {}", input, key, keyValue, keyValue != null ? keyValue.getClass() : "null");
 351                     continue;
 352                 }
 353
 354                 if (IndexPolicy.TRACE_INDEX_UPDATE)
 355                     System.out.println(getDescriptor() + "Removing document with key " + removedTerm);
 356                 writer.deleteDocuments(removedTerm);
 357             }
 358
 359             if (IndexPolicy.PERF_INDEX_UPDATE) {
 360                 end = System.nanoTime();
 361                 double ms = (end - start) * 1e-6;
 362                 System.out.println(getDescriptor() + "Removed " + keyValues.size() + " documents from index at " + indexPath + " in " + ms + " ms");
 363             }
 364
 365         } finally {
 366         }
 367     }
 368
 369     void removeIndex(IProgressMonitor monitor) throws DatabaseException, CorruptIndexException, IOException {
 370         assertAccessOpen(true);
 371
 372         long start = 0, end = 0;
 373         if (IndexPolicy.PERF_INDEX_UPDATE)
 374             start = System.nanoTime();
 375
 376         try {
 377
 378             writer.deleteAll();
 379
 380             if (IndexPolicy.PERF_INDEX_UPDATE) {
 381                 end = System.nanoTime();
 382                 double ms = (end - start) * 1e-6;
 383                 System.out.println(getDescriptor() + "Removed all documents from index at " + indexPath + " in " + ms + " ms");
 384             }
 385
 386         } finally {
 387         }
 388     }
 389
 390     boolean replaceIndex(IProgressMonitor monitor, String key, Collection<Object> keyValues, GenericRelation r, int boundLength, Collection<Object[]> documentsData) throws CorruptIndexException, IOException, DatabaseException {
 391
 392         boolean didReplace = false;
 393
 394         assertAccessOpen(true);
 395         if (keyValues.size() != documentsData.size())
 396             throw new IllegalArgumentException("keyValues size does not match documents data size, " + keyValues.size() + " <> " + documentsData.size());
 397
 398         if (IndexPolicy.TRACE_INDEX_UPDATE)
 399             System.out.println(getDescriptor() + "Replacing " + keyValues.size() + " documents from index at " + indexPath);
 400
 401         long start = 0, end = 0;
 402         if (IndexPolicy.PERF_INDEX_UPDATE)
 403             start = System.nanoTime();
 404
 405         try {
 406             Iterator<Object> keyIt = keyValues.iterator();
 407             Iterator<Object[]> documentDataIt = documentsData.iterator();
 408
 409             Document document = new Document();
 410             Field[] fs = makeFieldsForRelation(r, boundLength, document);
 411
 412             nextDocument:
 413                 while (keyIt.hasNext()) {
 414                     Object keyValue = keyIt.next();
 415                     Object[] documentData = documentDataIt.next();
 416
 417                     Term removedTerm = null;
 418                     if (keyValue instanceof Long) {
 419                         removedTerm = IndexUtils.longTerm(key, (Long) keyValue);
 420                     } else if (keyValue instanceof String) {
 421                         removedTerm = new Term(key, (String) keyValue);
 422                     } else {
 423                         // FIXME: should throw an exception for illegal input data but this would leave the index in an incoherent state
 424                         System.err.println("[" + getClass().getSimpleName() + "] Unrecognized document key to remove '" + keyValue + "', only " + String.class + " and " + Resource.class + " are supported.");
 425                         continue nextDocument;
 426                     }
 427
 428                     if (setFields(fs, documentData) == null)
 429                         continue nextDocument;
 430
 431                     if (IndexPolicy.TRACE_INDEX_UPDATE)
 432                         System.out.println(getDescriptor() + "Replacing document with key " + removedTerm + " with " + document);
 433
 434                     boolean done = false;
 435                     if(requireChangeInfoOnReplace()) {
 436                             TopDocs exist = searcher.search(new TermQuery(removedTerm), null, 2);
 437                             if(exist.scoreDocs.length == 1) {
 438                                 Document doc = reader.document(exist.scoreDocs[0].doc);
 439                                 if(!areSame(doc, document)) {
 440                                     writer.deleteDocuments(removedTerm);
 441                                     writer.addDocument(document);
 442                                     didReplace |= true;
 443                                     if (IndexPolicy.TRACE_INDEX_UPDATE)
 444                                         System.out.println("-replaced single existing");
 445                                 } else {
 446                                     if (IndexPolicy.TRACE_INDEX_UPDATE)
 447                                         System.out.println("-was actually same than single existing");
 448                                 }
 449                                 done = true;
 450                             }
 451                     }
 452                     if(!done) {
 453                         writer.deleteDocuments(removedTerm);
 454                         writer.addDocument(document);
 455                         didReplace |= true;
 456                         if (IndexPolicy.TRACE_INDEX_UPDATE)
 457                                 System.out.println("-had many or none - removed all existing");
 458                     }
 459
 460                 }
 461
 462             if (IndexPolicy.PERF_INDEX_UPDATE) {
 463                 end = System.nanoTime();
 464                 double ms = (end - start) * 1e-6;
 465                 System.out.println(getDescriptor() + "Replaced " + keyValues.size() + " documents from index at " + indexPath + " in " + ms + " ms");
 466             }
 467
 468         } finally {
 469         }
 470
 471         return didReplace;
 472
 473     }
 474
 475     protected boolean requireChangeInfoOnReplace() {
 476         return true;
 477     }
 478
 479     private boolean areSame(Document d1, Document d2) {
 480         List<IndexableField> fs1 = d1.getFields();
 481         List<IndexableField> fs2 = d2.getFields();
 482         if(fs1.size() != fs2.size()) return false;
 483         for(int i=0;i<fs1.size();i++) {
 484                 IndexableField f1 = fs1.get(i);
 485                 IndexableField f2 = fs2.get(i);
 486                 String s1 = f1.stringValue();
 487                 String s2 = f2.stringValue();
 488             if (IndexPolicy.TRACE_INDEX_UPDATE)
 489                 System.err.println("areSame " + s1 + " vs. " + s2 );
 490                 if(!ObjectUtils.objectEquals(s1,s2)) return false;
 491         }
 492         return true;
 493     }
 494
 495     final RequestProcessor session;
 496
 497     final Resource         relation;
 498
 499     /**
 500      * The schema of the index, i.e. the fields that will be indexed per
 501      * document for the specified relation. Since the relation stays the same
 502      * throughout the lifetime of this class, the index schema is also assumed
 503      * to the same. This means that {@link GenericRelation#getFields()} is
 504      * assumed to stay the same.
 505      */
 506     final IndexSchema      schema;
 507
 508     Resource         input;
 509
 510     Path             indexPath;
 511
 512     Directory        directory;
 513
 514     IndexReader      reader;
 515
 516     IndexWriter      writer;
 517
 518     IndexSearcher    searcher;
 519
 520     IndexedRelationsSearcherBase(RequestProcessor session, Resource relation, Resource input) {
 521         this.session = session;
 522         this.relation = relation;
 523         this.input = input;
 524         this.indexPath = getIndexDirectory(session.getSession(), relation, input);
 525         if(isIndexAvailable()) {
 526                 state = State.READY;
 527         } else {
 528                 state = State.NONE;
 529         }
 530         this.schema = IndexSchema.readFromRelation(session, relation);
 531     }
 532
 533     Directory getDirectory(Session session) throws IOException {
 534         return FSDirectory.open(indexPath.toFile());
 535     }
 536
 537     abstract String getDescriptor();
 538
 539     /**
 540      * Ensures that searcher is in read or write state.
 541      *
 542      * @param forWriting <code>true</code> to open index for writing,
 543      *        <code>false</code> for reading
 544      * @return true is required state was reached
 545      *
 546      */
 547     boolean startAccess(IProgressMonitor monitor, Session session, boolean forWriting) {
 548         if(forWriting) {
 549                 changeState(monitor, session, State.WRITE);
 550                 return checkState(State.WRITE);
 551         } else {
 552                 changeState(monitor, session, State.READ);
 553                 return checkState(State.READ);
 554         }
 555     }
 556
 557     boolean hasAccess(boolean forWriting) {
 558
 559         if (forWriting)
 560                 return checkState(State.WRITE);
 561         else
 562                 return checkState(State.WRITE) || checkState(State.READ);
 563
 564     }
 565
 566     void assertAccessOpen(boolean forWriting) {
 567         if (forWriting)
 568                 if(!checkState(State.WRITE))
 569                 throw new IllegalStateException("index not opened for writing (directory=" + directory + ", reader=" + reader + ")");
 570         else
 571                 if(!(checkState(State.WRITE) || checkState(State.READ)))
 572                 throw new IllegalStateException("index not opened for reading (directory=" + directory + ", writer=" + writer + ")");
 573     }
 574
 575     void closeWriter(IndexWriter writer) throws CorruptIndexException, IOException {
 576         if (writer == null)
 577             return;
 578
 579         try {
 580             // May throw OOME, see IndexWriter javadoc for the correct actions.
 581             writer.close(false);
 582         } catch (OutOfMemoryError e) {
 583             writer.close();
 584             throw e;
 585         }
 586     }
 587
 588     public static String getPattern(GenericRelation relation, int boundCount) {
 589         String result = "";
 590         for (int i = 0; i < boundCount; i++)
 591             result += "b";
 592         for (int i = 0; i < relation.getFields().length - boundCount; i++)
 593             result += "f";
 594         return result;
 595     }
 596
 597     void initializeIndex(IProgressMonitor monitor, ReadGraph graph, Object[] bound, boolean overwrite)
 598             throws IOException, DatabaseException
 599     {
 600         IndexingJob.jobifyIfPossible(
 601                 monitor,
 602                 "Reindexing " + NameUtils.getSafeLabel(graph, input),
 603                 mon -> {
 604                     try {
 605                         GenericRelation r = graph.adapt(relation, GenericRelation.class);
 606                         if (r == null)
 607                             throw new IndexingException("Given resource " + relation + "could not be adapted to GenericRelation.");
 608
 609                         GenericRelation selection = r.select(getPattern(r, bound.length), bound);
 610
 611                         List<Object[]> results = selection.realize(graph);
 612                         initializeIndexImpl(new CompletableFuture<>(), mon, r, results, bound, overwrite);
 613                     } catch (IOException e) {
 614                         getLogger().error("Index is in problematic state! {}", this, e);
 615                         throw new IndexingException(e);
 616                     }
 617                 });
 618     }
 619
 620     private static final int INDEXING_THREAD_COUNT = 2; // this is quite good parallelism level for lucene
 621
 622     void initializeIndexImpl(CompletableFuture<?> result, IProgressMonitor monitor, GenericRelation r, List<Object[]> results, final Object[] bound, boolean overwrite) throws IOException {
 623         try {
 624             final SubMonitor mon = SubMonitor.convert(monitor, 100);
 625
 626             if (IndexPolicy.TRACE_INDEX_INIT)
 627                 System.out.println(getDescriptor() + "Initializing index at " + indexPath + " (overwrite = " + overwrite + ")");
 628             mon.beginTask("Initializing Index", 100);
 629
 630             if (overwrite) {
 631                 if (Files.exists(indexPath)) {
 632                     mon.subTask("Erasing previous index");
 633                     if (getLogger().isDebugEnabled())
 634                         getLogger().debug("Erasing previous index {}", indexPath.toAbsolutePath());
 635                     FileUtils.emptyDirectory(indexPath);
 636                 }
 637             }
 638
 639             final AtomicReference<FSDirectory> directory = new AtomicReference<FSDirectory>();
 640             final AtomicReference<IndexWriter> writer = new AtomicReference<IndexWriter>();
 641
 642             try {
 643                 mon.subTask("Start index write");
 644                 Files.createDirectories(indexPath);
 645
 646                 directory.set(FSDirectory.open(indexPath.toFile()));
 647                 IndexWriterConfig conf = new IndexWriterConfig(Version.LUCENE_4_9, Queries.getAnalyzer()).setOpenMode(OpenMode.CREATE);
 648                 writer.set(new IndexWriter(directory.get(), conf));
 649
 650                 mon.worked(5);
 651
 652                 long realizeStart = 0;
 653                 if (IndexPolicy.PERF_INDEX_INIT)
 654                     realizeStart = System.nanoTime();
 655
 656                 mon.subTask("Calculating indexed content");
 657                 mon.worked(5);
 658
 659                 mon.worked(40);
 660
 661                 if (IndexPolicy.PERF_INDEX_INIT)
 662                     System.out.println(getDescriptor() + "Realized index with " + results.size() + " entries at " + indexPath + " in " + (1e-9 * (System.nanoTime()-realizeStart)) + " seconds.");
 663
 664                 if (IndexPolicy.TRACE_INDEX_INIT)
 665                     System.out.println(getDescriptor() + "Indexed relation " + r + " produced " + results.size() + " results");
 666
 667                 long start = IndexPolicy.PERF_INDEX_INIT ? System.nanoTime() : 0;
 668
 669                 mon.subTask("Indexing content");
 670                 final Semaphore s = new Semaphore(0);
 671                 mon.setWorkRemaining(results.size());
 672                 for (int i = 0; i < INDEXING_THREAD_COUNT; i++) {
 673                     final int startIndex = i;
 674                     ThreadUtils.getBlockingWorkExecutor().submit(() -> {
 675                         try {
 676                             Document document = new Document();
 677                             Field[] fs = makeFieldsForRelation(r, bound.length, document);
 678
 679                             for (int index = startIndex; index < results.size(); index += INDEXING_THREAD_COUNT) {
 680                                 if (setFields(fs, results.get(index)) == null)
 681                                     continue;
 682                                 try {
 683                                     writer.get().addDocument(document);
 684                                 } catch (CorruptIndexException e) {
 685                                     getLogger().error("Index is corrupted! {}", this, e);
 686                                     throw new IllegalStateException(e);
 687                                 } catch (IOException e) {
 688                                     getLogger().error("Index is in problematic state! {}", this, e);
 689                                     throw new IllegalStateException(e);
 690                                 } finally {
 691                                     synchronized (mon) {
 692                                         mon.worked(1);
 693                                     }
 694                                 }
 695                             }
 696                         } catch (DatabaseException e) {
 697                             throw new IllegalStateException(e);
 698                         } finally {
 699                             s.release();
 700                         }
 701                     });
 702                 }
 703
 704                 try {
 705                     s.acquire(INDEXING_THREAD_COUNT);
 706                 } catch (InterruptedException e) {
 707                     getLogger().error("Could not initialize index {}", this, e);
 708                 }
 709
 710                 // http://www.gossamer-threads.com/lists/lucene/java-dev/47895
 711                 // and http://lucene.apache.org/java/docs/index.html#27+November+2011+-+Lucene+Core+3.5.0
 712                 // advise against calling optimize at all. So let's not do it anymore.
 713                 //writer.get().optimize();
 714                 //writer.get().commit();
 715
 716                 mon.subTask("Flushing");
 717
 718                 if (IndexPolicy.PERF_INDEX_INIT)
 719                     System.out.println(getDescriptor() + "Wrote index at " + indexPath + " in " + (1e-9 * (System.nanoTime()-start)) + " seconds.");
 720
 721                 result.complete(null);
 722     //        } catch (DatabaseException e) {
 723     //            getLogger().error("Could not initialize index due to db {}", this, e);
 724             } finally {
 725                 try {
 726                     closeWriter(writer.getAndSet(null));
 727                 } finally {
 728                     FileUtils.uncheckedClose(directory.getAndSet(null));
 729                 }
 730             }
 731         } catch (Throwable t) {
 732             getLogger().error("Could not initialize index", t);
 733             result.completeExceptionally(t);
 734         }
 735     }
 736
 737
 738     public List<Object[]> debugDocs(IProgressMonitor monitor) throws ParseException, IOException, IndexingException {
 739
 740             Query query = new MatchAllDocsQuery();
 741
 742             TopDocs td = searcher.search(query, Integer.MAX_VALUE);
 743
 744             ScoreDoc[ ] scoreDocs = td.scoreDocs;
 745             List<Object[]> result = new ArrayList<Object[]>(scoreDocs.length);
 746
 747             for(ScoreDoc scoreDoc:scoreDocs) {
 748
 749                 try {
 750
 751                     Document doc = reader.document(scoreDoc.doc);
 752                     List<IndexableField> fs = doc.getFields();
 753                     Object[] o = new Object[fs.size()];
 754                     int index = 0;
 755                     for (IndexableField f : fs) {
 756                     o[index++] = f.stringValue();
 757                     }
 758                     result.add(o);
 759
 760             } catch (CorruptIndexException e) {
 761                 getLogger().error("Index is corrupted! {}", this, e);
 762                 throw new IndexCorruptedException("Index is corrupted! " + this, e);
 763             } catch (IOException e) {
 764                 getLogger().error("Index is in problematic state! {}", this, e);
 765                 throw new IndexingException(e);
 766             }
 767
 768             }
 769
 770             return result;
 771
 772     }
 773
 774
 775     List<Map<String, Object>> doSearch(IProgressMonitor monitor, RequestProcessor processor, String search, int maxResultCount) throws ParseException, IOException,
 776     IndexingException {
 777
 778         // An empty search string will crash QueryParser
 779         // Just return no results for empty queries.
 780         //System.out.println("search: '" + search + "'");
 781         if (search.isEmpty())
 782             return Collections.emptyList();
 783
 784         assertAccessOpen(false);
 785
 786         Query query = Queries.parse(search, schema);
 787
 788         long start = System.nanoTime();
 789
 790         maxResultCount = Math.min(maxResultCount, searcher.getIndexReader().numDocs());
 791         if (maxResultCount == 0)
 792             return Collections.emptyList();
 793
 794         final TopDocs docs = searcher.search(query, null, maxResultCount);
 795
 796 //        for(Object[] o : debugDocs(monitor)) {
 797 //            System.err.println("-" + Arrays.toString(o));
 798 //        }
 799
 800         if (IndexPolicy.PERF_INDEX_QUERY) {
 801             long end = System.nanoTime();
 802             System.out.println(getDescriptor() + "search(" + search + ", " + maxResultCount + ") into index at " + indexPath + " took " + (1e-9 * (end-start)) + " seconds.");
 803         }
 804
 805         if (docs.totalHits == 0) {
 806             return Collections.emptyList();
 807         }
 808
 809         try {
 810             return processor.syncRequest(new Read<List<Map<String, Object>>>() {
 811
 812                 @Override
 813                 public List<Map<String, Object>> perform(ReadGraph graph) throws DatabaseException {
 814
 815                     GenericRelation r = graph.adapt(relation, GenericRelation.class);
 816                     if (r == null)
 817                         throw new IndexingException("Given resource " + graph.syncRequest(new SafeName(relation))
 818                                 + "could not be adapted to GenericRelation.");
 819
 820                     SerialisationSupport support = graph.getService(SerialisationSupport.class);
 821
 822                     List<Map<String, Object>> result = new ArrayList<Map<String, Object>>(docs.scoreDocs.length);
 823
 824                     final DocumentStoredFieldVisitor visitor = new DocumentStoredFieldVisitor();
 825
 826                     for (ScoreDoc scoreDoc : docs.scoreDocs) {
 827
 828                         try {
 829
 830                             reader.document(scoreDoc.doc, visitor);
 831
 832                             Document doc = visitor.getDocument();
 833
 834                             List<IndexableField> fs = doc.getFields();
 835                             Map<String, Object> entry = new THashMap<String, Object>(fs.size());
 836                             for (IndexableField f : fs) {
 837                                 IndexSchema.Type type = schema.typeMap.get(f.name());
 838                                 if (type == IndexSchema.Type.LONG) {
 839                                     entry.put(f.name(), support.getResource(f.numericValue().longValue()));
 840                                 } else {
 841                                     entry.put(f.name(), f.stringValue());
 842                                 }
 843                             }
 844
 845                             result.add(entry);
 846
 847                         } catch (CorruptIndexException e) {
 848                             getLogger().error("Index is corrupted! {}", this, e);
 849                             throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
 850                         } catch (IOException e) {
 851                             getLogger().error("Index is in problematic state! {}", this, e);
 852                             throw new IndexingException(e);
 853                         }
 854                     }
 855                     return result;
 856                 }
 857             });
 858         } catch (DatabaseException e) {
 859             if (e instanceof IndexingException) {
 860                 throw (IndexingException) e;
 861             } else {
 862                 throw new IndexingException(e);
 863             }
 864         }
 865     }
 866
 867     static class ResourceVisitor extends StoredFieldVisitor {
 868
 869         public long id;
 870
 871                 @Override
 872                 public Status needsField(FieldInfo fieldInfo) throws IOException {
 873                         if("Resource".equals(fieldInfo.name)) return Status.YES;
 874                         return Status.NO;
 875                 }
 876
 877                 @Override
 878                 public void longField(FieldInfo fieldInfo, long value) throws IOException {
 879                         id = value;
 880                 }
 881
 882     };
 883
 884     static class DumpVisitor extends StoredFieldVisitor {
 885
 886         public List<Object> values;
 887
 888         DumpVisitor(List<Object> values) {
 889                 this.values = values;
 890         }
 891
 892                 @Override
 893                 public Status needsField(FieldInfo fieldInfo) throws IOException {
 894                         return Status.YES;
 895                 }
 896
 897                 @Override
 898                 public void longField(FieldInfo fieldInfo, long value) throws IOException {
 899                         values.add(value);
 900                 }
 901
 902                 @Override
 903                 public void stringField(FieldInfo fieldInfo, String value) throws IOException {
 904                         values.add(value);
 905                 }
 906
 907     }
 908
 909     List<Resource> doSearchResources(IProgressMonitor monitor, RequestProcessor processor, String search, int maxResultCount) throws ParseException, IOException,
 910     IndexingException {
 911
 912         // An empty search string will crash QueryParser
 913         // Just return no results for empty queries.
 914         //System.out.println("search: '" + search + "'");
 915         if (search.isEmpty())
 916             return Collections.emptyList();
 917
 918         assertAccessOpen(false);
 919
 920         Query query = Queries.parse(search, schema);
 921
 922         long start = System.nanoTime();
 923
 924         maxResultCount = Math.min(maxResultCount, searcher.getIndexReader().numDocs());
 925         if (maxResultCount == 0)
 926             return Collections.emptyList();
 927
 928         final TopDocs docs = searcher.search(query, null, maxResultCount);
 929
 930 //        for(Object[] o : debugDocs(monitor)) {
 931 //            System.err.println("-" + Arrays.toString(o));
 932 //        }
 933
 934         if (IndexPolicy.PERF_INDEX_QUERY) {
 935             long end = System.nanoTime();
 936             System.out.println(getDescriptor() + "search(" + search + ", " + maxResultCount + ") into index at " + indexPath + " took " + (1e-9 * (end-start)) + " seconds.");
 937         }
 938
 939         if (docs.totalHits == 0) {
 940             return Collections.emptyList();
 941         }
 942
 943         try {
 944             return processor.syncRequest(new Read<List<Resource>>() {
 945
 946                 @Override
 947                 public List<Resource> perform(ReadGraph graph) throws DatabaseException {
 948
 949                         CollectionSupport cs = graph.getService(CollectionSupport.class);
 950                     SerialisationSupport support = graph.getService(SerialisationSupport.class);
 951
 952                         List<Resource> result = cs.createList();
 953
 954                         ResourceVisitor visitor = new ResourceVisitor();
 955
 956                     for (ScoreDoc scoreDoc : docs.scoreDocs) {
 957                         try {
 958                                 reader.document(scoreDoc.doc, visitor);
 959                                 result.add(support.getResource(visitor.id));
 960                         } catch (CorruptIndexException e) {
 961                             getLogger().error("Index is corrupted! {}", this, e);
 962                             throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
 963                         } catch (IOException e) {
 964                             getLogger().error("Index is in problematic state! {}", this, e);
 965                             throw new IndexingException(e);
 966                         }
 967                     }
 968                     return result;
 969                 }
 970             });
 971         } catch (DatabaseException e) {
 972             if (e instanceof IndexingException) {
 973                 throw (IndexingException) e;
 974             } else {
 975                 throw new IndexingException(e);
 976             }
 977         }
 978     }
 979
 980     List<Object> doList(IProgressMonitor monitor, RequestProcessor processor) throws ParseException, IOException,
 981     IndexingException {
 982
 983         assertAccessOpen(false);
 984
 985         Query query = new MatchAllDocsQuery();
 986
 987         final TopDocs docs = searcher.search(query, Integer.MAX_VALUE);
 988
 989         ArrayList<Object> result = new ArrayList<Object>();
 990
 991         DumpVisitor visitor = new DumpVisitor(result);
 992
 993         for (ScoreDoc scoreDoc : docs.scoreDocs) {
 994
 995                 try {
 996
 997                         reader.document(scoreDoc.doc, visitor);
 998
 999                 } catch (CorruptIndexException e) {
1000                     getLogger().error("Index is corrupted! {}", this, e);
1001                     throw new IndexCorruptedException("Index is corrupted! " + " " + this + " " + scoreDoc, e);
1002                 } catch (IOException e) {
1003                     getLogger().error("Index is in problematic state! {}", this, e);
1004                         throw new IndexingException(e);
1005                 }
1006
1007         }
1008
1009         return result;
1010
1011     }
1012
1013     protected static Path getIndexDirectory(Session session, Resource relation, Resource input) {
1014         Path path = DatabaseIndexing.getIndexLocation(session, relation, input);
1015 //        System.out.println("getIndexDirectory = " + path);
1016         return path;
1017     }
1018
1019     Path getIndexPath() {
1020         return indexPath;
1021     }
1022
1023     boolean isIndexAvailable() {
1024         return Files.isDirectory(indexPath);
1025     }
1026
1027     abstract Throwable bestEffortClear(IProgressMonitor monitor, Session session);
1028
1029     /*
1030      * Start from scratch. Clear all caches and rebuild the index.
1031      */
1032     Throwable clearDirectory(IProgressMonitor monitor, Session session) {
1033
1034                 Path file = getIndexPath();
1035
1036         try {
1037                         FileUtils.delete(file);
1038         } catch (Throwable t) {
1039                 getLogger().error("Could not delete directory {}", file.toAbsolutePath(), t);
1040                 return t;
1041         }
1042         if (Files.exists(file))
1043             return new IllegalStateException("Failed to delete directory " + file.toAbsolutePath());
1044         return null;
1045     }
1046
1047     private Field[] setFields(Field[] fs, Object[] result) {
1048         for (int i = 0; i < result.length; i++) {
1049             Object value = result[i];
1050             if (value instanceof String) {
1051                 if (IndexPolicy.DEBUG_INDEX_INIT)
1052                     System.out.println(getDescriptor() + "index " + fs[i].name() + " = " + value + " : String");
1053                 fs[i].setStringValue((String) value);
1054             } else if (value instanceof Long) {
1055                 if (IndexPolicy.DEBUG_INDEX_INIT)
1056                     System.out.println(getDescriptor() + "index " + fs[i].name() + " = " + value + " : Long");
1057                 fs[i].setLongValue((Long) value);
1058             } else {
1059                 getLogger().error("Can only index Long and String fields, encountered " + value);
1060                 return null;
1061             }
1062         }
1063         return fs;
1064     }
1065
1066     protected abstract Logger getLogger();
1067
1068     @Override
1069     public String toString() {
1070         return getClass().getSimpleName() + " [" + String.valueOf(schema) + ", " + String.valueOf(relation) + ", " + String.valueOf(input) + ", " + String.valueOf(indexPath) + ", " + String.valueOf(directory) + ", " + String.valueOf(state) + "]";
1071     }
1072 }