bundles/org.simantics.db.indexing/src/org/simantics/db/indexing/IndexedRelationsSearcherBase.java

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