diff --git oak-run/pom.xml oak-run/pom.xml index 997e4ed..798ae4f 100644 --- oak-run/pom.xml +++ oak-run/pom.xml @@ -391,6 +391,12 @@ 1.10.19 test + + org.hamcrest + hamcrest-all + 1.3 + test + diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexCommand.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexCommand.java index e7e64e7..81d732f 100644 --- oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexCommand.java +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexCommand.java @@ -37,6 +37,7 @@ import joptsimple.OptionParser; import org.apache.commons.io.FileUtils; import org.apache.felix.inventory.Format; import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.index.indexer.document.DocumentStoreIndexer; import org.apache.jackrabbit.oak.plugins.index.importer.IndexDefinitionUpdater; import org.apache.jackrabbit.oak.run.cli.CommonOptions; import org.apache.jackrabbit.oak.run.cli.DocumentBuilderCustomizer; @@ -187,7 +188,7 @@ public class IndexCommand implements Command { } String checkpoint = indexOpts.getCheckpoint(); - File destDir = reindex(indexHelper, checkpoint); + File destDir = reindex(indexOpts, indexHelper, checkpoint); log.info("To complete indexing import the created index files via IndexerMBean#importIndex operation with " + "[{}] as input", getPath(destDir)); } @@ -199,7 +200,7 @@ public class IndexCommand implements Command { } } - private File reindex(IndexHelper indexHelper, String checkpoint) throws IOException, CommitFailedException { + private File reindex(IndexOptions idxOpts, IndexHelper indexHelper, String checkpoint) throws IOException, CommitFailedException { checkNotNull(checkpoint, "Checkpoint value is required for reindexing done in read only mode"); Stopwatch w = Stopwatch.createStarted(); @@ -207,8 +208,14 @@ public class IndexCommand implements Command { log.info("Proceeding to index {} upto checkpoint {} {}", indexHelper.getIndexPaths(), checkpoint, indexerSupport.getCheckpointInfo()); - try (OutOfBandIndexer indexer = new OutOfBandIndexer(indexHelper, indexerSupport)) { - indexer.reindex(); + if (opts.getCommonOpts().isDocument() && idxOpts.isDocTraversalMode()) { + try (DocumentStoreIndexer indexer = new DocumentStoreIndexer(indexHelper, indexerSupport)) { + indexer.reindex(); + } + } else { + try (OutOfBandIndexer indexer = new OutOfBandIndexer(indexHelper, indexerSupport)) { + indexer.reindex(); + } } indexerSupport.writeMetaInfo(checkpoint); @@ -245,7 +252,7 @@ public class IndexCommand implements Command { NodeStoreFixture fixture = NodeStoreFixtureProvider.create(opts, true); closer.register(fixture); IndexHelper indexHelper = createIndexHelper(fixture, indexOpts, closer); - reindex(indexHelper, checkpoint); + reindex(indexOpts, indexHelper, checkpoint); return new File(indexOpts.getOutDir(), OutOfBandIndexer.LOCAL_INDEX_ROOT_DIR); } } diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexOptions.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexOptions.java index 06f5f24..7bb4c62 100644 --- oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexOptions.java +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/IndexOptions.java @@ -51,6 +51,7 @@ public class IndexOptions implements OptionsBean { private final OptionSpec dumpIndex; private final OptionSpec reindex; private final OptionSpec importIndex; + private final OptionSpec docTraversal; private final OptionSpec consistencyCheck; private OptionSet options; private final Set actionOpts; @@ -90,6 +91,7 @@ public class IndexOptions implements OptionsBean { reindex = parser.accepts("reindex", "Reindex the indexes specified by --index-paths or --index-definitions-file"); importIndex = parser.accepts("index-import", "Imports index"); + docTraversal = parser.accepts("doc-traversal-mode", "Use Document traversal mode for reindex in DocumentNodeStore setups. This may provide better performance in some cases"); indexImportDir = parser.accepts("index-import-dir", "Directory containing index files. This " + "is required when --index-import operation is selected") @@ -179,6 +181,10 @@ public class IndexOptions implements OptionsBean { return options.has(importIndex); } + public boolean isDocTraversalMode() { + return options.has(docTraversal); + } + public String getCheckpoint(){ return checkpoint.value(options); } diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/CompositeIndexer.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/CompositeIndexer.java new file mode 100644 index 0000000..d7cbbc9 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/CompositeIndexer.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.IOException; +import java.util.List; + +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; + +import static com.google.common.base.Preconditions.checkNotNull; + +public class CompositeIndexer implements NodeStateIndexer { + private final List indexers; + + public CompositeIndexer(List indexers) { + this.indexers = checkNotNull(indexers); + } + + public boolean isEmpty() { + return indexers.isEmpty(); + } + + @Override + public boolean shouldInclude(String path) { + return indexers.stream().anyMatch(indexer -> indexer.shouldInclude(path)); + } + + @Override + public boolean shouldInclude(NodeDocument doc) { + return indexers.stream().anyMatch(indexer -> indexer.shouldInclude(doc)); + } + + @Override + public void index(NodeStateEntry entry) throws IOException, CommitFailedException { + for (NodeStateIndexer indexer : indexers) { + indexer.index(entry); + } + } + + @Override + public void close() throws IOException { + + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexer.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexer.java new file mode 100644 index 0000000..115d71d --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexer.java @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +import com.codahale.metrics.MetricRegistry; +import com.google.common.collect.FluentIterable; +import com.google.common.collect.ImmutableList; +import com.google.common.io.Closer; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.index.IndexHelper; +import org.apache.jackrabbit.oak.index.IndexerSupport; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentTraverser; +import org.apache.jackrabbit.oak.plugins.document.util.CloseableIterable; +import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection; +import org.apache.jackrabbit.oak.plugins.document.util.Utils; +import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback; +import org.apache.jackrabbit.oak.plugins.index.NodeTraversalCallback; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.apache.jackrabbit.oak.plugins.index.progress.MetricRateEstimator; +import org.apache.jackrabbit.oak.plugins.metric.MetricStatisticsProvider; +import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.apache.jackrabbit.oak.spi.state.NodeStateUtils; +import org.apache.jackrabbit.oak.stats.StatisticsProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static com.google.common.base.Preconditions.checkNotNull; +import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME; + +public class DocumentStoreIndexer implements Closeable{ + private final Logger log = LoggerFactory.getLogger(getClass()); + private final Logger traversalLog = LoggerFactory.getLogger(DocumentStoreIndexer.class.getName()+".traversal"); + private final Closer closer = Closer.create(); + private final IndexHelper indexHelper; + private final List indexerProviders; + private final IndexerSupport indexerSupport; + private final IndexingProgressReporter progressReporter = + new IndexingProgressReporter(IndexUpdateCallback.NOOP, NodeTraversalCallback.NOOP); + + public DocumentStoreIndexer(IndexHelper indexHelper, IndexerSupport indexerSupport) throws IOException { + this.indexHelper = indexHelper; + this.indexerSupport = indexerSupport; + this.indexerProviders = createProviders(); + } + + public void reindex() throws CommitFailedException, IOException { + configureEstimators(); + //TODO Support for adding new index definition + CompositeIndexer indexer = prepareIndexers(); + if (indexer.isEmpty()) { + return; + } + + closer.register(indexer); + + //TODO How to ensure we can safely read from secondary + NodeState checkpointedState = indexerSupport.retrieveNodeStateForCheckpoint(); + DocumentNodeState rootDocumentState = (DocumentNodeState) checkpointedState; + DocumentNodeStore nodeStore = (DocumentNodeStore) indexHelper.getNodeStore(); + + progressReporter.reindexingTraversalStart("/"); + for (NodeDocument doc : getIncludedDocs(indexer)) { + String path = doc.getPath(); + + DocumentNodeState nodeState = nodeStore.getNode(path, rootDocumentState.getRootRevision()); + + //At DocumentNodeState api level the nodeState can be null + if (nodeState == null || !nodeState.exists()) { + continue; + } + + //TODO Handle bundled NodeStates + + NodeStateEntry entry = new NodeStateEntry(nodeState, path); + indexer.index(entry); + } + + progressReporter.reindexingTraversalEnd(); + progressReporter.logReport(); + } + + private void configureEstimators() { + StatisticsProvider statsProvider = indexHelper.getStatisticsProvider(); + if (statsProvider instanceof MetricStatisticsProvider) { + MetricRegistry registry = ((MetricStatisticsProvider) statsProvider).getRegistry(); + progressReporter.setTraversalRateEstimator(new MetricRateEstimator("async", registry)); + } + + MongoConnection mongoConnection = indexHelper.getService(MongoConnection.class); + if (mongoConnection != null) { + long nodesCount = mongoConnection.getDB().getCollection("nodes").count(); + progressReporter.setNodeCountEstimator((String basePath, Set indexPaths) -> nodesCount); + log.info("Estimated number of documents in Mongo are {}", nodesCount); + } + } + + @Override + public void close() throws IOException { + closer.close(); + } + + @SuppressWarnings("Guava") + private Iterable getIncludedDocs(CompositeIndexer indexer) { + return FluentIterable.from(getDocsFilteredByPath(indexer)) + .filter(doc -> !doc.isSplitDocument()) + .filter(indexer::shouldInclude); + } + + private Iterable getDocsFilteredByPath(CompositeIndexer indexer) { + CloseableIterable docs = findAllDocuments(indexer); + closer.register(docs); + return docs; + } + + private CloseableIterable findAllDocuments(CompositeIndexer indexer) { + MongoDocumentStore mds = indexHelper.getService(MongoDocumentStore.class); + checkNotNull(mds); + return new MongoDocumentTraverser(mds).getAllDocuments(Collection.NODES, id -> includeId(id, indexer)); + } + + private boolean includeId(String id, NodeStateIndexer indexer) { + reportDocumentRead(id); + //Cannot interpret long paths as they are hashed. So let them + //be included + if (Utils.isIdFromLongPath(id)){ + return true; + } + + //Not easy to determine path for previous docs + //Given there count is pretty low compared to others + //include them all + if (Utils.isPreviousDocId(id)){ + return true; + } + + String path = Utils.getPathFromId(id); + + //Exclude hidden nodes from index data + if (NodeStateUtils.isHiddenPath(path)){ + return false; + } + + return indexer.shouldInclude(path); + } + + private void reportDocumentRead(String id) { + try { + progressReporter.traversedNode(() -> id); + } catch (CommitFailedException e) { + throw new RuntimeException(e); + } + traversalLog.trace(id); + } + + private CompositeIndexer prepareIndexers() { + NodeState root = indexHelper.getNodeStore().getRoot(); + List indexers = new ArrayList<>(); + for (String indexPath : indexHelper.getIndexPaths()) { + NodeState indexState = NodeStateUtils.getNode(root, indexPath); + String type = indexState.getString(TYPE_PROPERTY_NAME); + if (type == null) { + log.warn("No 'type' property found on indexPath [{}]. Skipping it", indexPath); + continue; + } + for (NodeStateIndexerProvider indexerProvider : indexerProviders) { + NodeStateIndexer indexer = indexerProvider.getIndexer(type, indexPath, indexState, root, progressReporter); + if (indexer != null) { + indexers.add(indexer); + closer.register(indexer); + progressReporter.registerIndex(indexPath, true, -1); + } + } + } + + return new CompositeIndexer(indexers); + } + + private List createProviders() throws IOException { + List providers = ImmutableList.of( + createLuceneIndexProvider() + ); + + providers.forEach(closer::register); + return providers; + } + + private NodeStateIndexerProvider createLuceneIndexProvider() throws IOException { + return new LuceneIndexerProvider(indexHelper, indexerSupport); + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java new file mode 100644 index 0000000..1cb9294 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexer.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.IOException; + +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.index.PathFilter; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexDefinition; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexDefinition.IndexingRule; +import org.apache.jackrabbit.oak.plugins.index.lucene.LuceneDocumentMaker; +import org.apache.jackrabbit.oak.plugins.index.lucene.binary.BinaryTextExtractor; +import org.apache.jackrabbit.oak.plugins.index.lucene.util.FacetHelper; +import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.apache.jackrabbit.oak.spi.state.NodeBuilder; +import org.apache.lucene.document.Document; + +public class LuceneIndexer implements NodeStateIndexer { + private final IndexDefinition definition; + private final BinaryTextExtractor binaryTextExtractor; + private final NodeBuilder definitionBuilder; + private final LuceneIndexWriter indexWriter; + private final IndexingProgressReporter progressReporter; + + public LuceneIndexer(IndexDefinition definition, LuceneIndexWriter indexWriter, + NodeBuilder builder, BinaryTextExtractor binaryTextExtractor, + IndexingProgressReporter progressReporter) { + this.definition = definition; + this.binaryTextExtractor = binaryTextExtractor; + this.indexWriter = indexWriter; + this.definitionBuilder = builder; + this.progressReporter = progressReporter; + } + + @Override + public boolean shouldInclude(String path) { + return definition.getPathFilter().filter(path) != PathFilter.Result.EXCLUDE; + } + + @Override + public boolean shouldInclude(NodeDocument doc) { + //TODO possible optimization for NodeType based filtering + return true; + } + + @Override + public void index(NodeStateEntry entry) throws IOException, CommitFailedException { + IndexingRule indexingRule = definition.getApplicableIndexingRule(entry.getNodeState()); + + if (indexingRule == null) { + return; + } + + LuceneDocumentMaker maker = newDocumentMaker(indexingRule, entry.getPath()); + Document doc = maker.makeDocument(entry.getNodeState()); + if (doc != null) { + writeToIndex(doc, entry.getPath()); + progressReporter.indexUpdate(definition.getIndexPath()); + } + } + + @Override + public void close() throws IOException { + indexWriter.close(System.currentTimeMillis()); + } + + private void writeToIndex(Document doc, String path) throws IOException { + indexWriter.updateDocument(path, doc); + } + + private LuceneDocumentMaker newDocumentMaker(IndexingRule indexingRule, String path) { + return new LuceneDocumentMaker( + binaryTextExtractor, + () -> FacetHelper.getFacetsConfig(definitionBuilder), //TODO FacetsConfig handling + null, //TODO augmentorFactory + definition, + indexingRule, + path + ); + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexerProvider.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexerProvider.java new file mode 100644 index 0000000..449cd32 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/LuceneIndexerProvider.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Nonnull; + +import org.apache.commons.io.FileUtils; +import org.apache.jackrabbit.oak.index.IndexHelper; +import org.apache.jackrabbit.oak.index.IndexerSupport; +import org.apache.jackrabbit.oak.plugins.index.lucene.ExtractedTextCache; +import org.apache.jackrabbit.oak.plugins.index.lucene.IndexDefinition; +import org.apache.jackrabbit.oak.plugins.index.lucene.binary.BinaryTextExtractor; +import org.apache.jackrabbit.oak.plugins.index.lucene.directory.DirectoryFactory; +import org.apache.jackrabbit.oak.plugins.index.lucene.directory.FSDirectoryFactory; +import org.apache.jackrabbit.oak.plugins.index.lucene.writer.DefaultIndexWriterFactory; +import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter; +import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriterFactory; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.apache.jackrabbit.oak.spi.state.NodeBuilder; +import org.apache.jackrabbit.oak.spi.state.NodeState; + +import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME; +import static org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexConstants.TYPE_LUCENE; + +public class LuceneIndexerProvider implements NodeStateIndexerProvider { + private final ExtractedTextCache textCache = + new ExtractedTextCache(FileUtils.ONE_MB * 5, TimeUnit.HOURS.toSeconds(5)); + private final IndexHelper indexHelper; + private final DirectoryFactory dirFactory; + private final LuceneIndexWriterFactory indexWriterFactory; + + public LuceneIndexerProvider(IndexHelper indexHelper, IndexerSupport indexerSupport) throws IOException { + this.indexHelper = indexHelper; + this.dirFactory = new FSDirectoryFactory(indexerSupport.getLocalIndexDir()); + this.indexWriterFactory = new DefaultIndexWriterFactory(indexHelper.getMountInfoProvider(), dirFactory); + } + + @Override + public NodeStateIndexer getIndexer(@Nonnull String type, @Nonnull String indexPath, + @Nonnull NodeState definition, @Nonnull NodeState root, + IndexingProgressReporter progressReporter) { + if (!TYPE_LUCENE.equals(definition.getString(TYPE_PROPERTY_NAME))) { + return null; + } + + NodeBuilder builder = definition.builder(); + IndexDefinition idxDefinition = IndexDefinition.newBuilder(root, definition, indexPath).reindex().build(); + + LuceneIndexWriter indexWriter = indexWriterFactory.newInstance(idxDefinition, builder, true); + BinaryTextExtractor textExtractor = new BinaryTextExtractor(textCache, idxDefinition, true); + return new LuceneIndexer( + idxDefinition, + indexWriter, + builder, + textExtractor, + progressReporter + ); + } + + @Override + public void close() throws IOException { + + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateEntry.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateEntry.java new file mode 100644 index 0000000..78b27c1 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateEntry.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import org.apache.jackrabbit.oak.spi.state.NodeState; + +public class NodeStateEntry { + private final NodeState nodeState; + private final String path; + + public NodeStateEntry(NodeState nodeState, String path) { + this.nodeState = nodeState; + this.path = path; + } + + public NodeState getNodeState() { + return nodeState; + } + + public String getPath() { + return path; + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexer.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexer.java new file mode 100644 index 0000000..da295c3 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexer.java @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.Closeable; +import java.io.IOException; + +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; + +public interface NodeStateIndexer extends Closeable{ + + boolean shouldInclude(String path); + + boolean shouldInclude(NodeDocument doc); + + void index(NodeStateEntry entry) throws IOException, CommitFailedException; +} + diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexerProvider.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexerProvider.java new file mode 100644 index 0000000..cbd964b --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/NodeStateIndexerProvider.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index.indexer.document; + +import java.io.Closeable; + +import javax.annotation.CheckForNull; +import javax.annotation.Nonnull; + +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.apache.jackrabbit.oak.spi.state.NodeState; + +public interface NodeStateIndexerProvider extends Closeable { + + @CheckForNull + NodeStateIndexer getIndexer(@Nonnull String type, + @Nonnull String indexPath, + @Nonnull NodeState definition, + @Nonnull NodeState root, IndexingProgressReporter progressReporter); +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentTraverser.java oak-run/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentTraverser.java new file mode 100644 index 0000000..434826a --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentTraverser.java @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.plugins.document.mongo; + +import java.util.function.Predicate; + +import com.google.common.collect.FluentIterable; +import com.google.common.io.Closer; +import com.mongodb.DBCollection; +import com.mongodb.DBCursor; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.Document; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.cache.CacheChangesTracker; +import org.apache.jackrabbit.oak.plugins.document.cache.NodeDocumentCache; +import org.apache.jackrabbit.oak.plugins.document.util.CloseableIterable; + +import static java.util.Collections.singletonList; + +public class MongoDocumentTraverser { + private final MongoDocumentStore mongoStore; + + public MongoDocumentTraverser(MongoDocumentStore mongoStore) { + this.mongoStore = mongoStore; + } + + public CloseableIterable getAllDocuments(Collection collection, Predicate filter) { + //TODO Handle readOnly + boolean readOnly = true; + DBCollection dbCollection = mongoStore.getDBCollection(collection); + Closer closer = Closer.create(); + CacheChangesTracker cacheChangesTracker; + if (collection == Collection.NODES && !readOnly) { + cacheChangesTracker = getNodeDocCache().registerTracker(NodeDocument.MIN_ID_VALUE, NodeDocument.MAX_ID_VALUE); + closer.register(cacheChangesTracker::close); + } else { + cacheChangesTracker = null; + } + + DBCursor cursor = dbCollection.find(); + //TODO This may lead to reads being routed to secondary depending on MongoURI + //So caller must ensure that its safe to read from secondary + cursor.setReadPreference(mongoStore.getConfiguredReadPreference(collection)); + closer.register(cursor); + + @SuppressWarnings("Guava") + Iterable result = FluentIterable.from(cursor) + .filter(o -> filter.test((String) o.get("_id"))) + .transform(o -> { + T doc = mongoStore.convertFromDBObject(collection, o); + //TODO Review the cache update approach where tracker has to track *all* docs + if (collection == Collection.NODES) { + NodeDocument nodeDoc = (NodeDocument) doc; + if (readOnly) { + getNodeDocCache().put(nodeDoc); + } + getNodeDocCache().putNonConflictingDocs(cacheChangesTracker, singletonList(nodeDoc)); + } + return doc; + }); + return CloseableIterable.wrap(result, closer); + } + + private NodeDocumentCache getNodeDocCache() { + return mongoStore.getNodeDocumentCache(); + } +} diff --git oak-run/src/test/java/org/apache/jackrabbit/oak/index/DocumentStoreIndexerIT.java oak-run/src/test/java/org/apache/jackrabbit/oak/index/DocumentStoreIndexerIT.java new file mode 100644 index 0000000..7a08a26 --- /dev/null +++ oak-run/src/test/java/org/apache/jackrabbit/oak/index/DocumentStoreIndexerIT.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.index; + +import java.io.File; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.apache.jackrabbit.oak.plugins.document.DocumentMKBuilderProvider; +import org.apache.jackrabbit.oak.plugins.document.MongoConnectionFactory; +import org.apache.jackrabbit.oak.plugins.document.MongoUtils; +import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection; +import org.apache.jackrabbit.oak.plugins.index.lucene.directory.IndexRootDirectory; +import org.apache.jackrabbit.oak.plugins.index.lucene.directory.LocalIndexDir; +import org.apache.jackrabbit.oak.spi.state.NodeStore; +import org.junit.Rule; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeNotNull; + +public class DocumentStoreIndexerIT extends AbstractIndexCommandTest { + @Rule + public MongoConnectionFactory connectionFactory = new MongoConnectionFactory(); + + @Rule + public DocumentMKBuilderProvider builderProvider = new DocumentMKBuilderProvider(); + + @Test + public void indexMongoRepo() throws Exception{ + fixture = new RepositoryFixture(temporaryFolder.getRoot(), getNodeStore()); + createTestData(false); + String checkpoint = fixture.getNodeStore().checkpoint(TimeUnit.HOURS.toMillis(24)); + fixture.close(); + + IndexCommand command = new IndexCommand(); + + File outDir = temporaryFolder.newFolder(); + String[] args = { + "--index-temp-dir=" + temporaryFolder.newFolder().getAbsolutePath(), + "--index-out-dir=" + outDir.getAbsolutePath(), + "--index-paths=/oak:index/fooIndex", + "--doc-traversal-mode", + "--checkpoint="+checkpoint, + "--reindex", + "--", // -- indicates that options have ended and rest needs to be treated as non option + MongoUtils.URL + }; + + command.execute(args); + + File indexes = new File(outDir, IndexerSupport.LOCAL_INDEX_ROOT_DIR); + assertTrue(indexes.exists()); + + IndexRootDirectory idxRoot = new IndexRootDirectory(indexes); + List idxDirs = idxRoot.getAllLocalIndexes(); + + assertEquals(1, idxDirs.size()); + } + + private NodeStore getNodeStore(){ + return builderProvider.newBuilder().setMongoDB(getConnection().getDB()).getNodeStore(); + } + + private MongoConnection getConnection(){ + MongoConnection conn = connectionFactory.getConnection(); + assumeNotNull(conn); + conn.getDB().dropDatabase(); + return conn; + } + +} diff --git oak-run/src/test/java/org/apache/jackrabbit/oak/plugins/document/mongo/DocumentTraverserTest.java oak-run/src/test/java/org/apache/jackrabbit/oak/plugins/document/mongo/DocumentTraverserTest.java new file mode 100644 index 0000000..8db3621 --- /dev/null +++ oak-run/src/test/java/org/apache/jackrabbit/oak/plugins/document/mongo/DocumentTraverserTest.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.jackrabbit.oak.plugins.document.mongo; + +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; + +import org.apache.jackrabbit.oak.plugins.document.AbstractDocumentStoreTest; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.DocumentStoreFixture; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.UpdateOp; +import org.apache.jackrabbit.oak.plugins.document.util.CloseableIterable; +import org.apache.jackrabbit.oak.plugins.document.util.Utils; +import org.junit.Test; + +import static java.util.Arrays.asList; +import static org.apache.jackrabbit.oak.plugins.document.util.Utils.getPathFromId; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assume.assumeTrue; + +public class DocumentTraverserTest extends AbstractDocumentStoreTest { + + public DocumentTraverserTest(DocumentStoreFixture dsf) { + super(dsf); + } + + @Test + public void getAllDocuments() throws Exception{ + assumeTrue(ds instanceof MongoDocumentStore); + ds.create(Collection.NODES, asList( + newDocument("/a/b", 1), + newDocument("/a/c", 1), + newDocument("/d", 1) + )); + + ds.invalidateCache(); + + MongoDocumentTraverser traverser = new MongoDocumentTraverser((MongoDocumentStore) ds); + CloseableIterable itr = traverser.getAllDocuments(Collection.NODES, id -> getPathFromId(id).startsWith("/a")); + Set paths = StreamSupport.stream(itr.spliterator(), false) + .map(NodeDocument::getPath) + .collect(Collectors.toSet()); + + itr.close(); + assertThat(paths, containsInAnyOrder("/a/b", "/a/c")); + + + assertNotNull(ds.getIfCached(Collection.NODES, Utils.getIdFromPath("/a/b"))); + assertNotNull(ds.getIfCached(Collection.NODES, Utils.getIdFromPath("/a/c"))); + + // Excluded id should not be cached + assertNull(ds.getIfCached(Collection.NODES, Utils.getIdFromPath("/d"))); + } + + private static UpdateOp newDocument(String path, long modified) { + String id = Utils.getIdFromPath(path); + UpdateOp op = new UpdateOp(id, true); + op.set(NodeDocument.MODIFIED_IN_SECS, modified); + return op; + } +}