OAK-10341 Tree store

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java

@@ -239,6 +239,16 @@ nodeStore, getMongoDocumentStore(), traversalLog))
         return storeList;
     }
 
+    public IndexStore buildTreeStore() throws IOException, CommitFailedException {
+        String old = System.setProperty(FlatFileNodeStoreBuilder.OAK_INDEXER_SORT_STRATEGY_TYPE,
+                FlatFileNodeStoreBuilder.SortStrategyType.PIPELINED_TREE.name());
+        try {
+            return buildFlatFileStore();
+        } finally {
+            System.setProperty(FlatFileNodeStoreBuilder.OAK_INDEXER_SORT_STRATEGY_TYPE, old);
+        }
+    }
+
     public IndexStore buildStore() throws IOException, CommitFailedException {
         return buildFlatFileStore();
     }

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/AheadOfTimeBlobDownloadingFlatFileStore.java

@@ -82,7 +82,15 @@ private AheadOfTimeBlobDownloadingFlatFileStore(FlatFileStore ffs, CompositeInde
         }
     }
 
-    static boolean isEnabledForIndexes(String indexesEnabledPrefix, List<String> indexPaths) {
+    /**
+     * Whether blob downloading is needed for the given indexes.
+     *
+     * @param indexesEnabledPrefix the comma-separated list of prefixes of the index
+     *                             definitions that benefit from the download
+     * @param indexPaths           the index paths
+     * @return true if any of the indexes start with any of the prefixes
+     */
+    public static boolean isEnabledForIndexes(String indexesEnabledPrefix, List<String> indexPaths) {
         List<String> enableForIndexes = splitAndTrim(indexesEnabledPrefix);
         for (String indexPath : indexPaths) {
             if (enableForIndexes.stream().anyMatch(indexPath::startsWith)) {

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java

@@ -29,10 +29,14 @@
 import org.apache.jackrabbit.oak.index.indexer.document.CompositeException;
 import org.apache.jackrabbit.oak.index.indexer.document.CompositeIndexer;
 import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverserFactory;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.ConfigHelper;
 import org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedTreeStoreStrategy;
 import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStore;
 import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreSortStrategy;
 import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.tree.Prefetcher;
+import org.apache.jackrabbit.oak.index.indexer.document.tree.TreeStore;
 import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore;
 import org.apache.jackrabbit.oak.plugins.document.RevisionVector;
 import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore;
@@ -122,7 +126,11 @@ public enum SortStrategyType {
         /**
          * System property {@link #OAK_INDEXER_SORT_STRATEGY_TYPE} if set to this value would result in {@link PipelinedStrategy} being used.
          */
-        PIPELINED
+        PIPELINED,
+        /**
+         * System property {@link #OAK_INDEXER_SORT_STRATEGY_TYPE} if set to this value would result in {@link PipelinedTreeStoreStrategy} being used.
+         */
+        PIPELINED_TREE,
     }
 
     public FlatFileNodeStoreBuilder(File workDir) {
@@ -224,20 +232,52 @@ public IndexStore build(IndexHelper indexHelper, CompositeIndexer indexer) throw
         entryWriter = new NodeStateEntryWriter(blobStore);
         IndexStoreFiles indexStoreFiles = createdSortedStoreFiles();
         File metadataFile = indexStoreFiles.metadataFile;
-        FlatFileStore store = new FlatFileStore(blobStore, indexStoreFiles.storeFiles.get(0), metadataFile,
-                new NodeStateEntryReader(blobStore),
-                unmodifiableSet(preferredPathElements), algorithm);
+        File file = indexStoreFiles.storeFiles.get(0);
+        IndexStore store;
+        if (file.isDirectory()) {
+            store = buildTreeStoreForIndexing(indexHelper, file);
+        } else {
+            store = new FlatFileStore(blobStore, file, metadataFile,
+                    new NodeStateEntryReader(blobStore),
+                    unmodifiableSet(preferredPathElements), algorithm);
+        }
         if (entryCount > 0) {
             store.setEntryCount(entryCount);
         }
         if (indexer == null || indexHelper == null) {
             return store;
         }
-        if (withAheadOfTimeBlobDownloading) {
-            return AheadOfTimeBlobDownloadingFlatFileStore.wrap(store, indexer, indexHelper);
-        } else {
-            return store;
+        if (withAheadOfTimeBlobDownloading && store instanceof FlatFileStore) {
+            FlatFileStore ffs = (FlatFileStore) store;
+            return AheadOfTimeBlobDownloadingFlatFileStore.wrap(ffs, indexer, indexHelper);
+        }
+        return store;
+    }
+
+    public IndexStore buildTreeStoreForIndexing(IndexHelper indexHelper, File file) {
+        TreeStore indexingTreeStore = new TreeStore(
+                "indexing", file,
+                new NodeStateEntryReader(blobStore), 10);
+        indexingTreeStore.setIndexDefinitions(indexDefinitions);
+
+        // use a separate tree store (with a smaller cache)
+        // for prefetching, to avoid cache evictions
+        TreeStore prefetchTreeStore = new TreeStore(
+                "prefetch", file,
+                new NodeStateEntryReader(blobStore), 3);
+        prefetchTreeStore.setIndexDefinitions(indexDefinitions);
+        String blobPrefetchEnableForIndexes = ConfigHelper.getSystemPropertyAsString(
+                AheadOfTimeBlobDownloadingFlatFileStore.BLOB_PREFETCH_ENABLE_FOR_INDEXES_PREFIXES, "");
+        Prefetcher prefetcher = new Prefetcher(prefetchTreeStore, indexingTreeStore);
+        String blobSuffix = "";
+        if (AheadOfTimeBlobDownloadingFlatFileStore.isEnabledForIndexes(
+                blobPrefetchEnableForIndexes, indexHelper.getIndexPaths())) {
+            blobSuffix = ConfigHelper.getSystemPropertyAsString(
+                    AheadOfTimeBlobDownloadingFlatFileStore.BLOB_PREFETCH_BINARY_NODES_SUFFIX, "");
         }
+        prefetcher.setBlobSuffix(blobSuffix);
+        prefetcher.startPrefetch();
+        return indexingTreeStore;
     }
 
     public List<IndexStore> buildList(IndexHelper indexHelper, IndexerSupport indexerSupport,
@@ -351,15 +391,24 @@ IndexStoreSortStrategy createSortStrategy(File dir) {
                 log.warn("TraverseWithSortStrategy is deprecated and will be removed in the near future. Use PipelinedStrategy instead.");
                 return new TraverseWithSortStrategy(nodeStateEntryTraverserFactory, preferredPathElements, entryWriter, dir,
                         algorithm, pathPredicate, checkpoint);
-            case PIPELINED:
+            case PIPELINED: {
                 log.info("Using PipelinedStrategy");
                 List<PathFilter> pathFilters = indexDefinitions.stream().map(IndexDefinition::getPathFilter).collect(Collectors.toList());
                 List<String> indexNames = indexDefinitions.stream().map(IndexDefinition::getIndexName).collect(Collectors.toList());
                 indexingReporter.setIndexNames(indexNames);
                 return new PipelinedStrategy(mongoClientURI, mongoDocumentStore, nodeStore, rootRevision,
                         preferredPathElements, blobStore, dir, algorithm, pathPredicate, pathFilters, checkpoint,
                         statisticsProvider, indexingReporter);
-
+            }
+            case PIPELINED_TREE: {
+                log.info("Using PipelinedTreeStoreStrategy");
+                List<PathFilter> pathFilters = indexDefinitions.stream().map(IndexDefinition::getPathFilter).collect(Collectors.toList());
+                List<String> indexNames = indexDefinitions.stream().map(IndexDefinition::getIndexName).collect(Collectors.toList());
+                indexingReporter.setIndexNames(indexNames);
+                return new PipelinedTreeStoreStrategy(mongoClientURI, mongoDocumentStore, nodeStore, rootRevision,
+                        preferredPathElements, blobStore, dir, algorithm, pathPredicate, pathFilters, checkpoint,
+                        statisticsProvider, indexingReporter);
+            }
         }
         throw new IllegalStateException("Not a valid sort strategy value " + sortStrategyType);
     }

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/BoundedHistogram.java

@@ -32,7 +32,7 @@
  * histogram are correct but if the histogram overflowed, it may be missing some entries.
  */
 public class BoundedHistogram {
-    private static final Logger LOG = LoggerFactory.getLogger(PipelinedStrategy.class);
+    private static final Logger LOG = LoggerFactory.getLogger(BoundedHistogram.class);
     private final ConcurrentHashMap<String, LongAdder> histogram = new ConcurrentHashMap<>();
     private volatile boolean overflowed = false;
     private final String histogramName;

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/ConfigHelper.java

@@ -22,7 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 public class ConfigHelper {
-    private static final Logger LOG = LoggerFactory.getLogger(PipelinedStrategy.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ConfigHelper.class);
 
     public static int getSystemPropertyAsInt(String name, int defaultValue) {
         int result = Integer.getInteger(name, defaultValue);

oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTransformTask.java

@@ -160,7 +160,7 @@ public Result call() throws Exception {
                     for (NodeDocument nodeDoc : nodeDocumentBatch) {
                         statistics.incrementMongoDocumentsTraversed();
                         mongoObjectsProcessed++;
-                        if (mongoObjectsProcessed % 50000 == 0) {
+                        if (mongoObjectsProcessed % 50_000 == 0) {
                             LOG.info("Mongo objects: {}, total entries: {}, current batch: {}, Size: {}/{} MB",
                                     mongoObjectsProcessed, totalEntryCount, nseBatch.numberOfEntries(),
                                     nseBatch.sizeOfEntriesBytes() / FileUtils.ONE_MB,