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