Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexNodeManager.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexNodeManager.java?rev=1841926&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexNodeManager.java
(added)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexNodeManager.java
Tue Sep 25 12:24:15 2018
@@ -0,0 +1,376 @@
+/*
+ * 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.index.lucene;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.Iterables;
+import org.apache.jackrabbit.oak.commons.PathUtils;
+import org.apache.jackrabbit.oak.commons.PerfLogger;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndex;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReader;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.reader.LuceneIndexReaderFactory;
+import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter;
+import
org.apache.jackrabbit.oak.plugins.index.search.update.ReaderRefreshPolicy;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.MultiReader;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.suggest.analyzing.AnalyzingInfixSuggester;
+import org.apache.lucene.store.Directory;
+import org.jetbrains.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static
org.apache.jackrabbit.oak.plugins.index.IndexConstants.ASYNC_PROPERTY_NAME;
+import static
org.apache.jackrabbit.oak.plugins.index.IndexUtils.getAsyncLaneName;
+
+/**
+ * Keeps track of the open read sessions for an index.
+ */
+public class LuceneIndexNodeManager {
+ // TODO oak-search: should extend LuceneIndexNodeManager
+ /**
+ * Name of the hidden node under which information about the checkpoints
+ * seen and indexed by each async indexer is kept.
+ */
+ static final String ASYNC = ":async";
+
+ private static final AtomicInteger SEARCHER_ID_COUNTER = new
AtomicInteger();
+
+ private static final PerfLogger PERF_LOGGER =
+ new
PerfLogger(LoggerFactory.getLogger(LuceneIndexNodeManager.class.getName() +
".perf"));
+
+ static LuceneIndexNodeManager open(String indexPath, NodeState root,
NodeState defnNodeState,
+ LuceneIndexReaderFactory readerFactory,
@Nullable NRTIndexFactory nrtFactory)
+ throws IOException {
+ LuceneIndexDefinition definition = new LuceneIndexDefinition(root,
defnNodeState, indexPath);
+ List<LuceneIndexReader> readers =
readerFactory.createReaders(definition, defnNodeState, indexPath);
+ NRTIndex nrtIndex = nrtFactory != null ?
nrtFactory.createIndex(definition) : null;
+ if (!readers.isEmpty() || (nrtIndex != null &&
!hasAsyncIndexerRun(root, indexPath, defnNodeState))){
+ return new LuceneIndexNodeManager(PathUtils.getName(indexPath),
definition, readers, nrtIndex);
+ }
+ return null;
+ }
+
+ static boolean hasAsyncIndexerRun(NodeState root, String indexPath,
NodeState defnNodeState) {
+ boolean hasAsyncNode = root.hasChildNode(ASYNC);
+
+ String asyncLaneName = getAsyncLaneName(defnNodeState, indexPath,
defnNodeState.getProperty(ASYNC_PROPERTY_NAME));
+
+ if (asyncLaneName != null) {
+ return hasAsyncNode &&
root.getChildNode(ASYNC).hasProperty(asyncLaneName);
+ } else {
+ // useful only for tests - basically non-async index defs which
don't rely on /:async
+ // hence either readers are there (and this method doesn't come
into play during open)
+ // OR there is no cycle (where we return false correctly)
+ return false;
+ }
+ }
+
+ private static final Logger log =
LoggerFactory.getLogger(LuceneIndexNodeManager.class);
+
+ private final List<LuceneIndexReader> readers;
+
+ private final String name;
+
+ private final LuceneIndexDefinition definition;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private volatile SearcherHolder searcherHolder;
+
+ private final NRTIndex nrtIndex;
+
+ private final ReaderRefreshPolicy refreshPolicy;
+
+ private final Semaphore refreshLock = new Semaphore(1);
+
+ private final Runnable refreshCallback = new Runnable() {
+ @Override
+ public void run() {
+ if (refreshLock.tryAcquire()) {
+ try {
+ refreshReaders();
+ }finally {
+ refreshLock.release();
+ }
+ }
+ }
+ };
+
+ private boolean closed = false;
+
+ LuceneIndexNodeManager(String name, LuceneIndexDefinition definition,
List<LuceneIndexReader> readers, @Nullable NRTIndex nrtIndex)
+ throws IOException {
+ checkArgument(!readers.isEmpty() || nrtIndex != null);
+ this.name = name;
+ this.definition = definition;
+ this.readers = readers;
+ this.nrtIndex = nrtIndex;
+ this.searcherHolder = createHolder(getNRTReaders());
+ this.refreshPolicy = nrtIndex != null ? nrtIndex.getRefreshPolicy() :
ReaderRefreshPolicy.NEVER;
+ }
+
+ private String getName() {
+ return name;
+ }
+
+ LuceneIndexDefinition getDefinition() {
+ return definition;
+ }
+
+ @Nullable
+ private Directory getSuggestDirectory() {
+ return readers.isEmpty() ? null :
getDefaultReader().getSuggestDirectory();
+ }
+
+ @Nullable
+ private AnalyzingInfixSuggester getLookup() {
+ return readers.isEmpty() ? null : getDefaultReader().getLookup();
+ }
+
+ @Nullable
+ LuceneIndexNode acquire() {
+ lock.readLock().lock();
+ if (closed) {
+ lock.readLock().unlock();
+ return null;
+ } else {
+ boolean success = false;
+ try {
+ refreshPolicy.refreshOnReadIfRequired(refreshCallback);
+ SearcherHolder local = searcherHolder;
+ int tryCount = 0;
+ while (!local.searcher.getIndexReader().tryIncRef()) {
+ checkState(++tryCount < 10, "Not able to " +
+ "get open searcher in %s attempts", tryCount);
+ local = searcherHolder;
+ }
+ LuceneIndexNode indexNode = new IndexNodeImpl(local);
+ success = true;
+ return indexNode;
+ } finally {
+ if (!success) {
+ lock.readLock().unlock();
+ }
+ }
+ }
+ }
+
+ private void release() {
+ lock.readLock().unlock();
+ }
+
+ void close() throws IOException {
+ lock.writeLock().lock();
+ try {
+ checkState(!closed);
+ closed = true;
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ releaseHolder(searcherHolder);
+ closeReaders(readers);
+ }
+
+ private List<LuceneIndexReader> getPrimaryReaders() {
+ return readers;
+ }
+
+ @Nullable
+ private LuceneIndexWriter getLocalWriter() throws IOException{
+ return nrtIndex != null ? nrtIndex.getWriter() : null;
+ }
+
+ private void refreshReadersOnWriteIfRequired() {
+ refreshPolicy.refreshOnWriteIfRequired(refreshCallback);
+ }
+
+ private void refreshReaders(){
+ long start = PERF_LOGGER.start();
+ List<LuceneIndexReader> newNRTReaders = getNRTReaders();
+ //The list reference would differ if index got updated
+ //so if they are same no need to reinitialize the searcher
+ if (newNRTReaders != searcherHolder.nrtReaders) {
+ SearcherHolder old = searcherHolder;
+ searcherHolder = createHolder(newNRTReaders);
+ releaseHolder(old);
+ PERF_LOGGER.end(start, 0, "Refreshed reader for index [{}]",
definition);
+ }
+ }
+
+ private LuceneIndexReader getDefaultReader(){
+ //TODO This is still required to support Suggester, Spellcheck etc
OAK-4643
+ return readers.get(0);
+ }
+
+ private IndexReader createReader(List<LuceneIndexReader> nrtReaders) {
+ //Increment count by 1. MultiReader does it for all readers
+ //So no need for an explicit increment for MultiReader
+
+ if (readers.size() == 1 && nrtReaders.isEmpty()){
+ IndexReader reader = readers.get(0).getReader();
+ reader.incRef();
+ return reader;
+ }
+ if (nrtReaders.size() == 1 && readers.isEmpty()){
+ IndexReader reader = nrtReaders.get(0).getReader();
+ reader.incRef();
+ return reader;
+ }
+
+ IndexReader[] readerArr = new IndexReader[readers.size() +
nrtReaders.size()];
+ int i = 0;
+ for (LuceneIndexReader r : Iterables.concat(readers, nrtReaders)){
+ readerArr[i++] = r.getReader();
+ }
+ return new MultiReader(readerArr, false);
+ }
+
+ private List<LuceneIndexReader> getNRTReaders() {
+ return nrtIndex != null ? nrtIndex.getReaders() :
Collections.<LuceneIndexReader>emptyList();
+ }
+
+ private SearcherHolder createHolder(List<LuceneIndexReader> newNRTReaders)
{
+ return new SearcherHolder(new
IndexSearcher(createReader(newNRTReaders)), newNRTReaders);
+ }
+
+ private void closeReaders(Iterable<LuceneIndexReader> readers) {
+ for (LuceneIndexReader r : readers){
+ try {
+ r.close();
+ } catch (IOException e) {
+ log.warn("Error occurred while releasing reader for index
[{}]", definition.getIndexPath(), e);
+ }
+ }
+ }
+
+ private void releaseHolder(SearcherHolder holder) {
+ decrementSearcherUsageCount(holder.searcher);
+ }
+
+ private void decrementSearcherUsageCount(IndexSearcher searcher) {
+ try {
+ //Decrement the count by 1 as we increased it while creating the
searcher
+ //in createReader
+ searcher.getIndexReader().decRef();
+ } catch (IOException e) {
+ log.warn("Error occurred while releasing reader for index [{}]",
definition.getIndexPath(), e);
+ }
+ }
+
+ private static class SearcherHolder {
+ final IndexSearcher searcher;
+ final List<LuceneIndexReader> nrtReaders;
+ final int searcherId = SEARCHER_ID_COUNTER.incrementAndGet();
+ final LuceneIndexStatistics indexStatistics;
+
+ public SearcherHolder(IndexSearcher searcher, List<LuceneIndexReader>
nrtReaders) {
+ this.searcher = searcher;
+ this.nrtReaders = nrtReaders;
+ this.indexStatistics = new
LuceneIndexStatistics(searcher.getIndexReader());
+ }
+
+ public LuceneIndexStatistics getIndexStatistics() {
+ return indexStatistics;
+ }
+ }
+
+ private class IndexNodeImpl implements LuceneIndexNode {
+ private final SearcherHolder holder;
+ private final AtomicBoolean released = new AtomicBoolean();
+
+ private IndexNodeImpl(SearcherHolder searcherHolder) {
+ this.holder = searcherHolder;
+ }
+
+ @Override
+ public void release() {
+ if (released.compareAndSet(false, true)) {
+ try {
+ //Decrement on each release
+ decrementSearcherUsageCount(holder.searcher);
+ } finally {
+ LuceneIndexNodeManager.this.release();
+ }
+ }
+ }
+
+ @Override
+ public IndexSearcher getSearcher() {
+ return holder.searcher;
+ }
+
+ @Override
+ public LuceneIndexStatistics getIndexStatistics() {
+ return holder.getIndexStatistics();
+ }
+
+ @Override
+ public LuceneIndexDefinition getDefinition() {
+ return definition;
+ }
+
+ @Override
+ public List<LuceneIndexReader> getPrimaryReaders() {
+ return LuceneIndexNodeManager.this.getPrimaryReaders();
+ }
+
+ @Override
+ public Directory getSuggestDirectory() {
+ return LuceneIndexNodeManager.this.getSuggestDirectory();
+ }
+
+ @Override
+ public List<LuceneIndexReader> getNRTReaders() {
+ return holder.nrtReaders;
+ }
+
+ @Override
+ public AnalyzingInfixSuggester getLookup() {
+ return LuceneIndexNodeManager.this.getLookup();
+ }
+
+ @Override
+ public int getIndexNodeId() {
+ return holder.searcherId;
+ }
+
+ @Override
+ public LuceneIndexWriter getLocalWriter() throws IOException {
+ return LuceneIndexNodeManager.this.getLocalWriter();
+ }
+
+ @Override
+ public void refreshReadersOnWriteIfRequired() {
+ LuceneIndexNodeManager.this.refreshReadersOnWriteIfRequired();
+ }
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexNodeManager.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProvider.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProvider.java?rev=1841926&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProvider.java
(added)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProvider.java
Tue Sep 25 12:24:15 2018
@@ -0,0 +1,108 @@
+/*
+ * 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.index.lucene;
+
+import java.io.Closeable;
+import java.util.List;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.jackrabbit.oak.plugins.index.aggregate.AggregateIndex;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.score.ScorerProviderFactory;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.commit.Observer;
+import org.apache.jackrabbit.oak.spi.query.QueryIndex;
+import org.apache.jackrabbit.oak.spi.query.QueryIndexProvider;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * A provider for Lucene indexes.
+ *
+ * @see LuceneIndex
+ */
+public class LuceneIndexProvider implements QueryIndexProvider, Observer,
Closeable {
+
+ protected final IndexTracker tracker;
+
+ protected volatile QueryIndex.NodeAggregator aggregator = null;
+
+ ScorerProviderFactory scorerFactory;
+
+ IndexAugmentorFactory augmentorFactory;
+
+ public LuceneIndexProvider() {
+ this(new IndexTracker());
+ }
+
+ public LuceneIndexProvider(IndexCopier indexCopier) {
+ this(new IndexTracker(indexCopier));
+ }
+
+ public LuceneIndexProvider(IndexTracker tracker) {
+ this(tracker, ScorerProviderFactory.DEFAULT, null);
+ }
+
+ public LuceneIndexProvider(IndexTracker tracker, ScorerProviderFactory
scorerFactory, IndexAugmentorFactory augmentorFactory) {
+ this.tracker = tracker;
+ this.scorerFactory = scorerFactory;
+ this.augmentorFactory = augmentorFactory;
+ }
+
+ public void close() {
+ tracker.close();
+ }
+
+ //----------------------------------------------------------< Observer >--
+
+ @Override
+ public void contentChanged(@NotNull NodeState root, @NotNull CommitInfo
info) {
+ tracker.update(root);
+ }
+
+ //------------------------------------------------< QueryIndexProvider >--
+
+ @Override @NotNull
+ public List<QueryIndex> getQueryIndexes(NodeState nodeState) {
+ return ImmutableList.<QueryIndex> of(new
AggregateIndex(newLuceneIndex()), newLucenePropertyIndex());
+ }
+
+ protected LuceneIndex newLuceneIndex() {
+ return new LuceneIndex(tracker, aggregator);
+ }
+
+ protected LucenePropertyIndex newLucenePropertyIndex() {
+ return new LucenePropertyIndex(tracker, scorerFactory,
augmentorFactory);
+ }
+
+ /**
+ * sets the default node aggregator that will be used at query time
+ */
+ public void setAggregator(QueryIndex.NodeAggregator aggregator) {
+ this.aggregator = aggregator;
+ }
+
+ // ----- helper builder method
+
+ public LuceneIndexProvider with(QueryIndex.NodeAggregator analyzer) {
+ this.setAggregator(analyzer);
+ return this;
+ }
+
+ IndexTracker getTracker() {
+ return tracker;
+ }
+}
Propchange:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProvider.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java?rev=1841926&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java
(added)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java
Tue Sep 25 12:24:15 2018
@@ -0,0 +1,822 @@
+/*
+ * 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.index.lucene;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.management.NotCompliantMBeanException;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Property;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.ReferencePolicy;
+import org.apache.felix.scr.annotations.ReferencePolicyOption;
+import org.apache.jackrabbit.oak.api.jmx.CacheStatsMBean;
+import org.apache.jackrabbit.oak.api.jmx.CheckpointMBean;
+import org.apache.jackrabbit.oak.cache.CacheStats;
+import org.apache.jackrabbit.oak.commons.PropertiesUtil;
+import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard;
+import org.apache.jackrabbit.oak.plugins.document.spi.JournalPropertyService;
+import org.apache.jackrabbit.oak.plugins.index.AsyncIndexInfoService;
+import org.apache.jackrabbit.oak.plugins.index.IndexEditorProvider;
+import org.apache.jackrabbit.oak.plugins.index.IndexInfoProvider;
+import org.apache.jackrabbit.oak.plugins.index.IndexPathService;
+import
org.apache.jackrabbit.oak.plugins.index.fulltext.PreExtractedTextProvider;
+import org.apache.jackrabbit.oak.plugins.index.importer.IndexImporterProvider;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.directory.ActiveDeletedBlobCollectorFactory;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.directory.BufferedOakDirectory;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.directory.LuceneIndexImporter;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.DocumentQueue;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.ExternalObserverBuilder;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.LocalIndexObserver;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.LuceneJournalPropertyService;
+import org.apache.jackrabbit.oak.plugins.index.lucene.hybrid.NRTIndexFactory;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.property.PropertyIndexCleaner;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.reader.DefaultIndexReaderFactory;
+import
org.apache.jackrabbit.oak.plugins.index.lucene.score.ScorerProviderFactory;
+import org.apache.jackrabbit.oak.plugins.index.search.ExtractedTextCache;
+import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
+import org.apache.jackrabbit.oak.plugins.index.search.TextExtractionStatsMBean;
+import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
+import org.apache.jackrabbit.oak.spi.commit.BackgroundObserver;
+import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean;
+import org.apache.jackrabbit.oak.spi.commit.Observer;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.mount.MountInfoProvider;
+import org.apache.jackrabbit.oak.spi.query.QueryIndex;
+import org.apache.jackrabbit.oak.spi.query.QueryIndexProvider;
+import org.apache.jackrabbit.oak.spi.state.Clusterable;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+import org.apache.jackrabbit.oak.spi.whiteboard.Registration;
+import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard;
+import org.apache.jackrabbit.oak.stats.Clock;
+import org.apache.jackrabbit.oak.stats.StatisticsProvider;
+import org.apache.lucene.analysis.util.CharFilterFactory;
+import org.apache.lucene.analysis.util.TokenFilterFactory;
+import org.apache.lucene.analysis.util.TokenizerFactory;
+import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.util.InfoStream;
+import org.jetbrains.annotations.NotNull;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceRegistration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.util.Collections.emptyMap;
+import static org.apache.commons.io.FileUtils.ONE_MB;
+import static
org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexConstants.TYPE_LUCENE;
+import static
org.apache.jackrabbit.oak.spi.blob.osgi.SplitBlobStoreService.ONLY_STANDALONE_TARGET;
+import static
org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerMBean;
+import static
org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.scheduleWithFixedDelay;
+
+@SuppressWarnings("UnusedDeclaration")
+@Component(metatype = true, label = "Apache Jackrabbit Oak
LuceneIndexProvider")
+public class LuceneIndexProviderService {
+ public static final String REPOSITORY_HOME = "repository.home";
+
+ private LuceneIndexProvider indexProvider;
+
+ private final List<ServiceRegistration> regs = Lists.newArrayList();
+ private final List<Registration> oakRegs = Lists.newArrayList();
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.OPTIONAL_UNARY,
+ policyOption = ReferencePolicyOption.GREEDY,
+ policy = ReferencePolicy.DYNAMIC
+ )
+ private QueryIndex.NodeAggregator nodeAggregator;
+
+ private static final boolean PROP_DISABLED_DEFAULT = false;
+
+ @Property(
+ boolValue = PROP_DISABLED_DEFAULT,
+ label = "Disable this component",
+ description = "If true, this component is disabled."
+ )
+ private static final String PROP_DISABLED = "disabled";
+
+ @Property(
+ boolValue = false,
+ label = "Enable Debug Logging",
+ description = "Enables debug logging in Lucene. After enabling
this actual logging can be " +
+ "controlled via changing log level for category 'oak.lucene' to
debug")
+ private static final String PROP_DEBUG = "debug";
+
+ @Property(
+ boolValue = true,
+ label = "Enable CopyOnRead",
+ description = "Enable copying of Lucene index to local file system
to improve query performance",
+ propertyPrivate = true
+ )
+ private static final String PROP_COPY_ON_READ = "enableCopyOnReadSupport";
+
+ @Property(
+ label = "Local index storage path",
+ description = "Local file system path where Lucene indexes would
be copied when CopyOnRead is enabled. " +
+ "If not specified then indexes would be stored under
'index' dir under Repository Home"
+ )
+ private static final String PROP_LOCAL_INDEX_DIR = "localIndexDir";
+
+
+ private static final boolean PROP_COPY_ON_WRITE_DEFAULT = true;
+ @Property(
+ boolValue = PROP_COPY_ON_WRITE_DEFAULT,
+ label = "Enable CopyOnWrite",
+ description = "Enable copying of Lucene index to local file system
to improve index writer performance",
+ propertyPrivate = true
+ )
+ private static final String PROP_COPY_ON_WRITE =
"enableCopyOnWriteSupport";
+
+ @Property(
+ boolValue = true,
+ label = "Open index asynchronously",
+ description = "Enable opening of indexes in asynchronous mode"
+ )
+ private static final String PROP_ASYNC_INDEX_OPEN = "enableOpenIndexAsync";
+
+ private static final int PROP_THREAD_POOL_SIZE_DEFAULT = 5;
+ @Property(
+ intValue = PROP_THREAD_POOL_SIZE_DEFAULT,
+ label = "Thread pool size",
+ description = "Thread pool size used to perform various
asynchronous task in Oak Lucene"
+ )
+ private static final String PROP_THREAD_POOL_SIZE = "threadPoolSize";
+
+ private static final boolean PROP_PREFETCH_INDEX_FILES_DEFAULT = true;
+ @Property(
+ boolValue = PROP_PREFETCH_INDEX_FILES_DEFAULT,
+ label = "Prefetch Index Files",
+ description = "Prefetch the index files when CopyOnRead is
enabled. When enabled all new Lucene" +
+ " index files would be copied locally before the index is
made available to QueryEngine"
+ )
+ private static final String PROP_PREFETCH_INDEX_FILES =
"prefetchIndexFiles";
+
+ private static final int PROP_EXTRACTED_TEXT_CACHE_SIZE_DEFAULT = 20;
+ @Property(
+ intValue = PROP_EXTRACTED_TEXT_CACHE_SIZE_DEFAULT,
+ label = "Extracted text cache size (MB)",
+ description = "Cache size in MB for caching extracted text for
some time. When set to 0 then " +
+ "cache would be disabled"
+ )
+ private static final String PROP_EXTRACTED_TEXT_CACHE_SIZE =
"extractedTextCacheSizeInMB";
+
+ private static final int PROP_EXTRACTED_TEXT_CACHE_EXPIRY_DEFAULT = 300;
+ @Property(
+ intValue = PROP_EXTRACTED_TEXT_CACHE_EXPIRY_DEFAULT,
+ label = "Extracted text cache expiry (secs)",
+ description = "Time in seconds for which the extracted text would
be cached in memory"
+ )
+ private static final String PROP_EXTRACTED_TEXT_CACHE_EXPIRY =
"extractedTextCacheExpiryInSecs";
+
+ private static final boolean PROP_PRE_EXTRACTED_TEXT_ALWAYS_USE_DEFAULT =
false;
+ @Property(
+ boolValue = PROP_PRE_EXTRACTED_TEXT_ALWAYS_USE_DEFAULT,
+ label = "Always use pre-extracted text cache",
+ description = "By default pre extracted text cache would only be
used for reindex case. If this setting " +
+ "is enabled then it would also be used in normal
incremental indexing"
+ )
+ private static final String PROP_PRE_EXTRACTED_TEXT_ALWAYS_USE =
"alwaysUsePreExtractedCache";
+
+ private static final int PROP_BOOLEAN_CLAUSE_LIMIT_DEFAULT = 1024;
+ @Property(
+ intValue = PROP_BOOLEAN_CLAUSE_LIMIT_DEFAULT,
+ label = "Boolean Clause Limit",
+ description = "Limit for number of boolean clauses generated for
handling of OR query"
+ )
+ private static final String PROP_BOOLEAN_CLAUSE_LIMIT =
"booleanClauseLimit";
+
+ private static final boolean PROP_HYBRID_INDEXING_DEFAULT = true;
+ @Property(
+ boolValue = PROP_HYBRID_INDEXING_DEFAULT,
+ label = "Hybrid Indexing",
+ description = "When enabled Lucene NRT Indexing mode would be
enabled"
+ )
+ private static final String PROP_HYBRID_INDEXING = "enableHybridIndexing";
+
+ private static final int PROP_HYBRID_QUEUE_SIZE_DEFAULT = 10000;
+ @Property(
+ intValue = PROP_HYBRID_QUEUE_SIZE_DEFAULT,
+ label = "Queue size",
+ description = "Size of in memory queue used for storing Lucene
Documents which need to be " +
+ "added to local index"
+ )
+ private static final String PROP_HYBRID_QUEUE_SIZE = "hybridQueueSize";
+
+ private static final boolean PROP_DISABLE_DEFN_STORAGE_DEFAULT = false;
+ @Property(
+ boolValue = PROP_DISABLE_DEFN_STORAGE_DEFAULT,
+ label = "Disable index definition storage",
+ description = "By default index definitions would be stored at
time of reindexing to ensure that future " +
+ "modifications to it are not effective untill index is
reindex. Set this to true would disable " +
+ "this feature"
+ )
+ private static final String PROP_DISABLE_STORED_INDEX_DEFINITION =
"disableStoredIndexDefinition";
+
+ private static final boolean PROP_DELETED_BLOB_COLLECTION_DEFAULT_ENABLED
= true;
+ @Property(
+ boolValue = PROP_DELETED_BLOB_COLLECTION_DEFAULT_ENABLED,
+ label = "Enable actively removing deleted index blobs from blob
store",
+ description = "Index blobs are explicitly unique and don't require
mark-sweep type collection." +
+ "This is used to enable the feature. Cleanup implies
purging index blobs marked as deleted " +
+ "earlier during some indexing cycle."
+ )
+ private static final String
PROP_NAME_DELETED_BLOB_COLLECTION_DEFAULT_ENABLED =
"deletedBlobsCollectionEnabled";
+
+ private static final int PROP_INDEX_CLEANER_INTERVAL_DEFAULT = 10*60;
+ @Property(
+ intValue = PROP_INDEX_CLEANER_INTERVAL_DEFAULT,
+ label = "Property Index Cleaner Interval (seconds)",
+ description = "Cleaner interval time (in seconds) for synchronous
property indexes configured as " +
+ "part of lucene indexes"
+ )
+ private static final String PROP_INDEX_CLEANER_INTERVAL =
"propIndexCleanerIntervalInSecs";
+
+ private static final boolean
PROP_ENABLE_SINGLE_BLOB_PER_INDEX_FILE_DEFAULT = true;
+ @Property(
+ boolValue = PROP_ENABLE_SINGLE_BLOB_PER_INDEX_FILE_DEFAULT,
+ label = "With CoW enabled, should index files by written as single
blobs",
+ description = "Index files can be written as single blobs as
chunked into smaller blobs. Enable" +
+ " this to write single blob per index file. This would
reduce number of blobs in the data store."
+ )
+ private static final String PROP_NAME_ENABLE_SINGLE_BLOB_PER_INDEX_FILE =
"enableSingleBlobIndexFiles";
+
+ private final Clock clock = Clock.SIMPLE;
+
+ private Whiteboard whiteboard;
+
+ private BackgroundObserver backgroundObserver;
+
+ private BackgroundObserver externalIndexObserver;
+
+ @Reference
+ ScorerProviderFactory scorerFactory;
+
+ @Reference
+ private IndexAugmentorFactory augmentorFactory;
+
+ @Reference
+ private StatisticsProvider statisticsProvider;
+
+ @Reference(policy = ReferencePolicy.DYNAMIC,
+ cardinality = ReferenceCardinality.OPTIONAL_UNARY,
+ policyOption = ReferencePolicyOption.GREEDY
+ )
+ private volatile PreExtractedTextProvider extractedTextProvider;
+
+ @Reference
+ private MountInfoProvider mountInfoProvider;
+
+ @Reference
+ private NodeStore nodeStore;
+
+ @Reference
+ private IndexPathService indexPathService;
+
+ @Reference
+ private AsyncIndexInfoService asyncIndexInfoService;
+
+ @Reference(
+ cardinality = ReferenceCardinality.OPTIONAL_UNARY,
+ policy = ReferencePolicy.STATIC,
+ policyOption = ReferencePolicyOption.GREEDY,
+ target = ONLY_STANDALONE_TARGET
+ )
+ private GarbageCollectableBlobStore blobStore;
+
+ @Reference
+ private CheckpointMBean checkpointMBean;
+
+ private IndexCopier indexCopier;
+
+ private ActiveDeletedBlobCollectorFactory.ActiveDeletedBlobCollector
activeDeletedBlobCollector;
+
+ private File indexDir;
+
+ private ExecutorService executorService;
+
+ private int threadPoolSize;
+
+ private ExtractedTextCache extractedTextCache;
+
+ private boolean hybridIndex;
+
+ private NRTIndexFactory nrtIndexFactory;
+
+ private DocumentQueue documentQueue;
+
+ private LuceneIndexEditorProvider editorProvider;
+
+ private IndexTracker tracker;
+
+ private PropertyIndexCleaner cleaner;
+
+ @Activate
+ private void activate(BundleContext bundleContext, Map<String, ?> config)
+ throws NotCompliantMBeanException, IOException {
+ boolean disabled = PropertiesUtil.toBoolean(config.get(PROP_DISABLED),
PROP_DISABLED_DEFAULT);
+ hybridIndex =
PropertiesUtil.toBoolean(config.get(PROP_HYBRID_INDEXING),
PROP_DISABLED_DEFAULT);
+
+ if (disabled) {
+ log.info("Component disabled by configuration");
+ return;
+ }
+
+ configureIndexDefinitionStorage(config);
+ configureBooleanClauseLimit(config);
+ initializeFactoryClassLoaders(getClass().getClassLoader());
+ if
(System.getProperty(BufferedOakDirectory.ENABLE_WRITING_SINGLE_BLOB_INDEX_FILE_PARAM)
== null) {
+
BufferedOakDirectory.setEnableWritingSingleBlobIndexFile(PropertiesUtil.toBoolean(
+ config.get(PROP_NAME_ENABLE_SINGLE_BLOB_PER_INDEX_FILE),
+ PROP_ENABLE_SINGLE_BLOB_PER_INDEX_FILE_DEFAULT));
+ } else {
+ log.info("Not setting config for single blob for an index file as
it's set by command line!");
+ }
+
+ whiteboard = new OsgiWhiteboard(bundleContext);
+ threadPoolSize =
PropertiesUtil.toInteger(config.get(PROP_THREAD_POOL_SIZE),
PROP_THREAD_POOL_SIZE_DEFAULT);
+ initializeIndexDir(bundleContext, config);
+ initializeExtractedTextCache(bundleContext, config);
+ tracker = createTracker(bundleContext, config);
+ indexProvider = new LuceneIndexProvider(tracker, scorerFactory,
augmentorFactory);
+ initializeActiveBlobCollector(whiteboard, config);
+ initializeLogging(config);
+ initialize();
+
+
regs.add(bundleContext.registerService(QueryIndexProvider.class.getName(),
indexProvider, null));
+ registerObserver(bundleContext, config);
+ registerLocalIndexObserver(bundleContext, tracker, config);
+ registerIndexEditor(bundleContext, tracker, config);
+ registerIndexInfoProvider(bundleContext);
+ registerIndexImporterProvider(bundleContext);
+ registerPropertyIndexCleaner(config, bundleContext);
+
+ oakRegs.add(registerMBean(whiteboard,
+ LuceneIndexMBean.class,
+ new LuceneIndexMBeanImpl(indexProvider.getTracker(),
nodeStore, indexPathService, getIndexCheckDir(), cleaner),
+ LuceneIndexMBean.TYPE,
+ "Lucene Index statistics"));
+ registerGCMonitor(whiteboard, indexProvider.getTracker());
+ }
+
+ private File getIndexCheckDir() {
+ return new File(checkNotNull(indexDir), "indexCheckDir");
+ }
+
+ @Deactivate
+ private void deactivate() throws InterruptedException, IOException {
+ for (ServiceRegistration reg : regs) {
+ reg.unregister();
+ }
+
+ for (Registration reg : oakRegs){
+ reg.unregister();
+ }
+
+ if (backgroundObserver != null){
+ backgroundObserver.close();
+ }
+
+ if (externalIndexObserver != null){
+ externalIndexObserver.close();
+ }
+
+ if (indexProvider != null) {
+ indexProvider.close();
+ indexProvider = null;
+ }
+
+ if (documentQueue != null){
+ documentQueue.close();
+ }
+
+ if (nrtIndexFactory != null){
+ nrtIndexFactory.close();
+ }
+
+ //Close the copier first i.e. before executorService
+ if (indexCopier != null){
+ indexCopier.close();
+ }
+
+ if (executorService != null){
+ executorService.shutdown();
+ executorService.awaitTermination(1, TimeUnit.MINUTES);
+ }
+
+ if (extractedTextCache != null) {
+ extractedTextCache.close();
+ }
+
+ InfoStream.setDefault(InfoStream.NO_OUTPUT);
+ }
+
+ void initializeIndexDir(BundleContext bundleContext, Map<String, ?>
config) {
+ String indexDirPath =
PropertiesUtil.toString(config.get(PROP_LOCAL_INDEX_DIR), null);
+ if (Strings.isNullOrEmpty(indexDirPath)) {
+ String repoHome = bundleContext.getProperty(REPOSITORY_HOME);
+ if (repoHome != null){
+ indexDirPath = FilenameUtils.concat(repoHome, "index");
+ }
+ }
+
+ checkNotNull(indexDirPath, "Index directory cannot be determined as
neither index " +
+ "directory path [%s] nor repository home [%s] defined",
PROP_LOCAL_INDEX_DIR, REPOSITORY_HOME);
+
+ indexDir = new File(indexDirPath);
+ }
+
+ IndexCopier getIndexCopier() {
+ return indexCopier;
+ }
+
+ ExtractedTextCache getExtractedTextCache() {
+ return extractedTextCache;
+ }
+
+ private void initialize(){
+ if(indexProvider == null){
+ return;
+ }
+
+ if(nodeAggregator != null){
+ log.debug("Using NodeAggregator {}", nodeAggregator.getClass());
+ }
+
+ indexProvider.setAggregator(nodeAggregator);
+ }
+
+ private void initializeLogging(Map<String, ?> config) {
+ boolean debug = PropertiesUtil.toBoolean(config.get(PROP_DEBUG),
false);
+ if (debug) {
+ InfoStream.setDefault(LoggingInfoStream.INSTANCE);
+ log.info("Registered LoggingInfoStream with Lucene. Lucene logs
can be enabled " +
+ "now via category [{}]", LoggingInfoStream.PREFIX);
+ }
+ }
+
+ private void registerIndexEditor(BundleContext bundleContext, IndexTracker
tracker, Map<String, ?> config) throws IOException {
+ boolean enableCopyOnWrite =
PropertiesUtil.toBoolean(config.get(PROP_COPY_ON_WRITE),
PROP_COPY_ON_WRITE_DEFAULT);
+ if (enableCopyOnWrite){
+ initializeIndexCopier(bundleContext, config);
+ editorProvider = new LuceneIndexEditorProvider(indexCopier,
tracker, extractedTextCache,
+ augmentorFactory, mountInfoProvider,
activeDeletedBlobCollector);
+ log.info("Enabling CopyOnWrite support. Index files would be
copied under {}", indexDir.getAbsolutePath());
+ } else {
+ editorProvider = new LuceneIndexEditorProvider(null, tracker,
extractedTextCache, augmentorFactory,
+ mountInfoProvider, activeDeletedBlobCollector);
+ }
+ editorProvider.setBlobStore(blobStore);
+
+ if (hybridIndex){
+ editorProvider.setIndexingQueue(checkNotNull(documentQueue));
+ }
+
+ Dictionary<String, Object> props = new Hashtable<String, Object>();
+ props.put("type", TYPE_LUCENE);
+
regs.add(bundleContext.registerService(IndexEditorProvider.class.getName(),
editorProvider, props));
+ oakRegs.add(registerMBean(whiteboard,
+ TextExtractionStatsMBean.class,
+ editorProvider.getExtractedTextCache().getStatsMBean(),
+ TextExtractionStatsMBean.TYPE,
+ "TextExtraction statistics"));
+ }
+
+ private IndexTracker createTracker(BundleContext bundleContext,
Map<String, ?> config) throws IOException {
+ boolean enableCopyOnRead =
PropertiesUtil.toBoolean(config.get(PROP_COPY_ON_READ), true);
+ IndexTracker tracker;
+ if (enableCopyOnRead){
+ initializeIndexCopier(bundleContext, config);
+ log.info("Enabling CopyOnRead support. Index files would be copied
under {}", indexDir.getAbsolutePath());
+ if (hybridIndex) {
+ nrtIndexFactory = new NRTIndexFactory(indexCopier,
statisticsProvider);
+ }
+ tracker = new IndexTracker(new
DefaultIndexReaderFactory(mountInfoProvider, indexCopier), nrtIndexFactory);
+ } else {
+ tracker = new IndexTracker(new
DefaultIndexReaderFactory(mountInfoProvider, null));
+ }
+
+ tracker.setAsyncIndexInfoService(asyncIndexInfoService);
+ return tracker;
+ }
+
+ private void initializeIndexCopier(BundleContext bundleContext,
Map<String, ?> config) throws IOException {
+ if(indexCopier != null){
+ return;
+ }
+ boolean prefetchEnabled =
PropertiesUtil.toBoolean(config.get(PROP_PREFETCH_INDEX_FILES),
+ PROP_PREFETCH_INDEX_FILES_DEFAULT);
+
+ if (prefetchEnabled){
+ log.info("Prefetching of index files enabled. Index would be
opened after copying all new files locally");
+ }
+
+ indexCopier = new IndexCopier(getExecutorService(), indexDir,
prefetchEnabled);
+
+ oakRegs.add(registerMBean(whiteboard,
+ CopyOnReadStatsMBean.class,
+ indexCopier,
+ CopyOnReadStatsMBean.TYPE,
+ "IndexCopier support statistics"));
+
+ }
+
+ ExecutorService getExecutorService(){
+ if (executorService == null){
+ executorService = createExecutor();
+ }
+ return executorService;
+ }
+
+ private ExecutorService createExecutor() {
+ ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 60L,
TimeUnit.SECONDS,
+ new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
+ private final AtomicInteger counter = new AtomicInteger();
+ private final Thread.UncaughtExceptionHandler handler = new
Thread.UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ log.warn("Error occurred in asynchronous processing ", e);
+ }
+ };
+ @Override
+ public Thread newThread(@NotNull Runnable r) {
+ Thread thread = new Thread(r, createName());
+ thread.setDaemon(true);
+ thread.setPriority(Thread.MIN_PRIORITY);
+ thread.setUncaughtExceptionHandler(handler);
+ return thread;
+ }
+
+ private String createName() {
+ return "oak-lucene-" + counter.getAndIncrement();
+ }
+ });
+ executor.setKeepAliveTime(1, TimeUnit.MINUTES);
+ executor.allowCoreThreadTimeOut(true);
+ return executor;
+ }
+
+ private void registerObserver(BundleContext bundleContext, Map<String, ?>
config) {
+ boolean enableAsyncIndexOpen =
PropertiesUtil.toBoolean(config.get(PROP_ASYNC_INDEX_OPEN), true);
+ Observer observer = indexProvider;
+ if (enableAsyncIndexOpen) {
+ backgroundObserver = new BackgroundObserver(indexProvider,
getExecutorService(), 5);
+ observer = backgroundObserver;
+ oakRegs.add(registerMBean(whiteboard,
+ BackgroundObserverMBean.class,
+ backgroundObserver.getMBean(),
+ BackgroundObserverMBean.TYPE,
+ "LuceneIndexConfigObserver queue stats"));
+ log.info("Registering the LuceneIndexProvider as a
BackgroundObserver");
+ }
+ regs.add(bundleContext.registerService(Observer.class.getName(),
observer, null));
+ }
+
+ private void registerLocalIndexObserver(BundleContext bundleContext,
IndexTracker tracker, Map<String, ?> config) {
+ if (!hybridIndex){
+ log.info("Hybrid indexing feature disabled");
+ return;
+ }
+
+ int queueSize =
PropertiesUtil.toInteger(config.get(PROP_HYBRID_QUEUE_SIZE),
PROP_HYBRID_QUEUE_SIZE_DEFAULT);
+ documentQueue = new DocumentQueue(queueSize, tracker,
getExecutorService(), statisticsProvider);
+ LocalIndexObserver localIndexObserver = new
LocalIndexObserver(documentQueue, statisticsProvider);
+ regs.add(bundleContext.registerService(Observer.class.getName(),
localIndexObserver, null));
+
+ int observerQueueSize = 1000;
+ int builderMaxSize = 5000;
+
regs.add(bundleContext.registerService(JournalPropertyService.class.getName(),
+ new LuceneJournalPropertyService(builderMaxSize), null));
+ ExternalObserverBuilder builder = new
ExternalObserverBuilder(documentQueue, tracker, statisticsProvider,
+ getExecutorService(), observerQueueSize);
+ log.info("Configured JournalPropertyBuilder with max size {} and
backed by BackgroundObserver " +
+ "with queue size {}", builderMaxSize, observerQueueSize);
+
+ Observer observer = builder.build();
+ externalIndexObserver = builder.getBackgroundObserver();
+ regs.add(bundleContext.registerService(Observer.class.getName(),
observer, null));
+ oakRegs.add(registerMBean(whiteboard,
+ BackgroundObserverMBean.class,
+ externalIndexObserver.getMBean(),
+ BackgroundObserverMBean.TYPE,
+ "LuceneExternalIndexObserver queue stats"));
+ log.info("Hybrid indexing enabled for configured indexes with queue
size of {}", queueSize);
+ }
+
+ private void initializeFactoryClassLoaders(ClassLoader classLoader) {
+ ClassLoader originalClassLoader = Thread.currentThread()
+ .getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(classLoader);
+ //Access TokenizerFactory etc trigger a static initialization
+ //so switch the TCCL so that static initializer picks up the right
+ //classloader
+ initializeFactoryClassLoaders0(classLoader);
+ initializeClasses();
+ } catch (Throwable t) {
+ log.warn("Error occurred while initializing the Lucene " +
+ "Factories", t);
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ private void initializeFactoryClassLoaders0(ClassLoader classLoader) {
+ //Factories use the Threads context classloader to perform SPI classes
+ //lookup by default which would not work in OSGi world. So reload the
+ //factories by providing the bundle classloader
+ TokenizerFactory.reloadTokenizers(classLoader);
+ CharFilterFactory.reloadCharFilters(classLoader);
+ TokenFilterFactory.reloadTokenFilters(classLoader);
+ }
+
+ private void initializeClasses() {
+ // prevent LUCENE-6482
+ // (also done in IndexDefinition, just to be save)
+ OakCodec ensureLucene46CodecLoaded = new OakCodec();
+ // to ensure the JVM doesn't optimize away object creation
+ // (probably not really needed; just to be save)
+ log.debug("Lucene46Codec is loaded: {}", ensureLucene46CodecLoaded);
+ }
+
+ private void initializeExtractedTextCache(BundleContext bundleContext,
Map<String, ?> config) {
+ int cacheSizeInMB =
PropertiesUtil.toInteger(config.get(PROP_EXTRACTED_TEXT_CACHE_SIZE),
+ PROP_EXTRACTED_TEXT_CACHE_SIZE_DEFAULT);
+ int cacheExpiryInSecs =
PropertiesUtil.toInteger(config.get(PROP_EXTRACTED_TEXT_CACHE_EXPIRY),
+ PROP_EXTRACTED_TEXT_CACHE_EXPIRY_DEFAULT);
+ boolean alwaysUsePreExtractedCache =
PropertiesUtil.toBoolean(config.get(PROP_PRE_EXTRACTED_TEXT_ALWAYS_USE),
+ PROP_PRE_EXTRACTED_TEXT_ALWAYS_USE_DEFAULT);
+
+ extractedTextCache = new ExtractedTextCache(
+ cacheSizeInMB * ONE_MB,
+ cacheExpiryInSecs,
+ alwaysUsePreExtractedCache,
+ indexDir);
+ if (extractedTextProvider != null){
+ registerExtractedTextProvider(extractedTextProvider);
+ }
+ CacheStats stats = extractedTextCache.getCacheStats();
+ if (stats != null){
+ oakRegs.add(registerMBean(whiteboard,
+ CacheStatsMBean.class, stats,
+ CacheStatsMBean.TYPE, stats.getName()));
+ log.info("Extracted text caching enabled with maxSize {} MB,
expiry time {} secs",
+ cacheSizeInMB, cacheExpiryInSecs);
+ }
+ }
+
+ private void registerExtractedTextProvider(PreExtractedTextProvider
provider){
+ if (extractedTextCache != null){
+ if (provider != null){
+ String usage =
extractedTextCache.isAlwaysUsePreExtractedCache() ?
+ "always" : "only during reindexing phase";
+ log.info("Registering PreExtractedTextProvider {} with
extracted text cache. " +
+ "It would be used {}", provider, usage);
+ } else {
+ log.info("Unregistering PreExtractedTextProvider with
extracted text cache");
+ }
+ extractedTextCache.setExtractedTextProvider(provider);
+ }
+ }
+
+ private void configureBooleanClauseLimit(Map<String, ?> config) {
+ int booleanClauseLimit =
PropertiesUtil.toInteger(config.get(PROP_BOOLEAN_CLAUSE_LIMIT),
+ PROP_BOOLEAN_CLAUSE_LIMIT_DEFAULT);
+ if (booleanClauseLimit != BooleanQuery.getMaxClauseCount()){
+ BooleanQuery.setMaxClauseCount(booleanClauseLimit);
+ log.info("Changed the Max boolean clause limit to {}",
booleanClauseLimit);
+ }
+ }
+
+ private void configureIndexDefinitionStorage(Map<String, ?> config) {
+ boolean disableStorage =
PropertiesUtil.toBoolean(config.get(PROP_DISABLE_STORED_INDEX_DEFINITION),
+ PROP_DISABLE_DEFN_STORAGE_DEFAULT);
+ if (disableStorage){
+ log.info("Feature to ensure that index definition matches the
index state is disabled. Change in " +
+ "index definition would now affect query plans and might
lead to inconsistent results.");
+ IndexDefinition.setDisableStoredIndexDefinition(disableStorage);
+ }
+ }
+
+ private void registerGCMonitor(Whiteboard whiteboard,
+ final IndexTracker tracker) {
+ GCMonitor gcMonitor = new GCMonitor.Empty() {
+ @Override
+ public void compacted() {
+ tracker.refresh();
+ }
+ };
+ oakRegs.add(whiteboard.register(GCMonitor.class, gcMonitor,
emptyMap()));
+ }
+
+ private void registerIndexInfoProvider(BundleContext bundleContext) {
+ IndexInfoProvider infoProvider = new
LuceneIndexInfoProvider(nodeStore, asyncIndexInfoService, getIndexCheckDir());
+
regs.add(bundleContext.registerService(IndexInfoProvider.class.getName(),
infoProvider, null));
+ }
+
+ private void registerIndexImporterProvider(BundleContext bundleContext) {
+ LuceneIndexImporter importer = new LuceneIndexImporter(blobStore);
+
regs.add(bundleContext.registerService(IndexImporterProvider.class.getName(),
importer, null));
+ }
+
+ private void initializeActiveBlobCollector(Whiteboard whiteboard,
Map<String, ?> config) {
+ boolean activeDeletionEnabled = PropertiesUtil.toBoolean(
+ config.get(PROP_NAME_DELETED_BLOB_COLLECTION_DEFAULT_ENABLED),
+ PROP_DELETED_BLOB_COLLECTION_DEFAULT_ENABLED);
+ if (activeDeletionEnabled && blobStore!= null) {
+ File blobCollectorWorkingDir = new File(indexDir, "deleted-blobs");
+ activeDeletedBlobCollector =
ActiveDeletedBlobCollectorFactory.newInstance(blobCollectorWorkingDir,
+ getExecutorService());
+ ActiveDeletedBlobCollectorMBean bean =
+ new
ActiveDeletedBlobCollectorMBeanImpl(activeDeletedBlobCollector, whiteboard,
nodeStore,
+ indexPathService, asyncIndexInfoService,
blobStore, getExecutorService());
+
+ oakRegs.add(registerMBean(whiteboard,
ActiveDeletedBlobCollectorMBean.class, bean,
+ ActiveDeletedBlobCollectorMBean.TYPE, "Active lucene files
collection"));
+ log.info("Active blob collector initialized at working dir: {}",
blobCollectorWorkingDir);
+ } else {
+ activeDeletedBlobCollector =
ActiveDeletedBlobCollectorFactory.NOOP;
+ log.info("Active blob collector set to NOOP. enabled: {} seconds;
blobStore: {}",
+ activeDeletionEnabled, blobStore);
+ }
+ }
+
+ private void registerPropertyIndexCleaner(Map<String, ?> config,
BundleContext bundleContext) {
+ int cleanerInterval =
PropertiesUtil.toInteger(config.get(PROP_INDEX_CLEANER_INTERVAL),
+ PROP_INDEX_CLEANER_INTERVAL_DEFAULT);
+
+ if (cleanerInterval <= 0) {
+ log.info("Property index cleaner would not be registered");
+ return;
+ }
+
+ cleaner = new PropertyIndexCleaner(nodeStore, indexPathService,
asyncIndexInfoService, statisticsProvider);
+
+ //Proxy check for DocumentNodeStore
+ if (nodeStore instanceof Clusterable) {
+ cleaner.setRecursiveDelete(true);
+ log.info("PropertyIndexCleaner configured to perform recursive
delete");
+ }
+ oakRegs.add(scheduleWithFixedDelay(whiteboard, cleaner,
+ ImmutableMap.of("scheduler.name",
PropertyIndexCleaner.class.getName()),
+ cleanerInterval, true, true));
+ log.info("Property index cleaner configured to run every [{}]
seconds", cleanerInterval);
+ }
+
+
+ protected void bindNodeAggregator(QueryIndex.NodeAggregator aggregator) {
+ this.nodeAggregator = aggregator;
+ initialize();
+ }
+
+ protected void unbindNodeAggregator(QueryIndex.NodeAggregator aggregator) {
+ this.nodeAggregator = null;
+ initialize();
+ }
+
+ protected void bindExtractedTextProvider(PreExtractedTextProvider
preExtractedTextProvider){
+ this.extractedTextProvider = preExtractedTextProvider;
+ registerExtractedTextProvider(preExtractedTextProvider);
+ }
+
+ protected void unbindExtractedTextProvider(PreExtractedTextProvider
preExtractedTextProvider){
+ this.extractedTextProvider = null;
+ registerExtractedTextProvider(null);
+ }
+}
Propchange:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexProviderService.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexStatistics.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexStatistics.java?rev=1841926&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexStatistics.java
(added)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexStatistics.java
Tue Sep 25 12:24:15 2018
@@ -0,0 +1,126 @@
+/*
+ * 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.index.lucene;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.apache.jackrabbit.oak.plugins.index.search.IndexStatistics;
+import org.apache.lucene.index.Fields;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.MultiFields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.jackrabbit.oak.plugins.index.search.FieldNames.isPropertyField;
+
+/**
+ * This class would populate some statistics from a reader. We want to be
careful here such that
+ * we only collect statistics which don't incur reads from the index i.e. we
would only collect
+ * stats that lucene would already have read into memory when the reader was
opened.
+ */
+public class LuceneIndexStatistics implements IndexStatistics {
+ private static final Logger LOG =
LoggerFactory.getLogger(LuceneIndexStatistics.class);
+ private final int numDocs;
+ private final Map<String, Integer> numDocsForField;
+ private final boolean safelyInitialized;
+
+ // For ease of tests as there didn't seem an easy way to make an
IndexReader delegator
+ // that would fail calls to reader on-demand.
+ static boolean failReadingFields = false;
+ static boolean failReadingSyntheticallyFalliableField = false;
+
+ static final String SYNTHETICALLY_FALLIABLE_FIELD =
"synthetically-falliable-field";
+
+ /**
+ * @param reader {@link IndexReader} for which statistics need to be
collected.
+ */
+ LuceneIndexStatistics(IndexReader reader) {
+ numDocs = reader.numDocs();
+
+ Map<String, Integer> numDocsForField = Maps.newHashMap();
+
+ Fields fields = null;
+ try {
+ if (failReadingFields) {
+ throw new IOException("Synthetically fail to read fields");
+ }
+ fields = MultiFields.getFields(reader);
+ } catch (IOException e) {
+ LOG.warn("Couldn't open fields for reader ({}). Won't extract doc
count per field", reader);
+ numDocsForField = null;
+ }
+
+
+ if (fields != null) {
+ for(String f : fields) {
+ if (isPropertyField(f)) {
+ int docCntForField = -1;
+ try {
+ if (failReadingSyntheticallyFalliableField &&
SYNTHETICALLY_FALLIABLE_FIELD.equals(f)) {
+ throw new IOException("Synthetically fail to read
count for field jcr:title");
+ }
+ docCntForField = reader.getDocCount(f);
+ } catch (IOException e) {
+ LOG.warn("Couldn't read doc count for field {} via
reader ({}).");
+ }
+ numDocsForField.put(f, docCntForField);
+ }
+ }
+ }
+
+ if (numDocsForField != null) {
+ this.numDocsForField =
Collections.unmodifiableMap(numDocsForField);
+ this.safelyInitialized = true;
+ } else {
+ this.numDocsForField = Collections.emptyMap();
+ this.safelyInitialized = false;
+ }
+ }
+
+ /**
+ * @return number of documents in the index
+ */
+ public int numDocs() {
+ return numDocs;
+ }
+
+ /**
+ * @param field Index field for which number of indexed documents are to
be return
+ * @return number of indexed documents (without subtracting potentially
deleted ones)
+ * for the given {@code field}.<br>
+ * -1: if index codec doesn't store doc-count-for-field
statistics, OR <br>
+ * reader threw an
exception while reading fields, OR <br>
+ * exception thrown while
reading count for the field, OR <br>
+ * doc-count is asked for
a non-property field.
+ */
+ public int getDocCountFor(String field) {
+ if (!safelyInitialized) {
+ return -1;
+ }
+
+ int docCntForField = isPropertyField(field) ? 0 : -1;
+ if (numDocsForField.containsKey(field)) {
+ docCntForField = numDocsForField.get(field);
+ }
+
+ return docCntForField;
+ }
+
+}
Propchange:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexStatistics.java
------------------------------------------------------------------------------
svn:eol-style = native
Added:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexWriterFactory.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexWriterFactory.java?rev=1841926&view=auto
==============================================================================
---
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexWriterFactory.java
(added)
+++
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexWriterFactory.java
Tue Sep 25 12:24:15 2018
@@ -0,0 +1,30 @@
+/*
+ * 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.index.lucene;
+
+import org.apache.jackrabbit.oak.plugins.index.lucene.writer.LuceneIndexWriter;
+import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriterFactory;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+
+public interface LuceneIndexWriterFactory extends FulltextIndexWriterFactory {
+ @Override
+ LuceneIndexWriter newInstance(IndexDefinition definition, NodeBuilder
definitionBuilder, boolean reindex);
+}
Propchange:
jackrabbit/oak/trunk/oak-lucene/src/main/java/org/apache/jackrabbit/oak/plugins/index/lucene/LuceneIndexWriterFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
