trunk/oak-search/src/main/java/org/apache/jackrabbit/oak/plugins/index/search/spi/editor/FulltextIndexEditorContext.java - jackrabbit-oak - Git at Google

 /*
  * 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.search.spi.editor;

 import java.io.IOException;
 import java.util.Calendar;

 import org.apache.jackrabbit.oak.api.CommitFailedException;
 import org.apache.jackrabbit.oak.api.Type;
 import org.apache.jackrabbit.oak.commons.PerfLogger;
 import org.apache.jackrabbit.oak.plugins.index.IndexConstants;
 import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback;
 import org.apache.jackrabbit.oak.plugins.index.IndexingContext;
 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.PropertyUpdateCallback;
 import org.apache.jackrabbit.oak.plugins.index.search.ReindexOperations;
 import org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
 import org.apache.jackrabbit.oak.plugins.index.search.util.NodeStateCloner;
 import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
 import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
 import org.apache.jackrabbit.oak.spi.state.NodeState;
 import org.apache.jackrabbit.oak.spi.state.NodeStateUtils;
 import org.apache.jackrabbit.oak.stats.Clock;
 import org.apache.jackrabbit.util.ISO8601;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import static com.google.common.base.Preconditions.checkNotNull;
 import static org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants.PROP_REFRESH_DEFN;
 import static org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition.INDEX_DEFINITION_NODE;

 /**
  *
  */
 public abstract class FulltextIndexEditorContext<D> {

   private static final Logger log = LoggerFactory
       .getLogger(FulltextIndexEditorContext.class);

   private static final PerfLogger PERF_LOGGER =
       new PerfLogger(LoggerFactory.getLogger(FulltextIndexEditorContext.class.getName() + ".perf"));

   protected IndexDefinition definition;

   protected final NodeBuilder definitionBuilder;

   private final FulltextIndexWriterFactory<D> indexWriterFactory;

   private FulltextIndexWriter<D> writer = null;

   private long indexedNodes;

   private final IndexUpdateCallback updateCallback;

   private boolean reindex;

   private final ExtractedTextCache extractedTextCache;

   private final NodeState root;

   private final IndexingContext indexingContext;

   private final boolean asyncIndexing;

   //Intentionally static, so that it can be set without passing around clock objects
   //Set for testing ONLY
   private static Clock clock = Clock.SIMPLE;

   private final boolean indexDefnRewritten;

   private FulltextBinaryTextExtractor textExtractor;

   private PropertyUpdateCallback propertyUpdateCallback;

   protected FulltextIndexEditorContext(NodeState root, NodeBuilder definition,
                                        @Nullable IndexDefinition indexDefinition,
                                        IndexUpdateCallback updateCallback,
                                        FulltextIndexWriterFactory indexWriterFactory,
                                        ExtractedTextCache extractedTextCache,
                                        IndexingContext indexingContext, boolean asyncIndexing) {
     this.root = root;
     this.indexingContext = checkNotNull(indexingContext);
     this.definitionBuilder = definition;
     this.indexWriterFactory = indexWriterFactory;
     this.definition = indexDefinition != null ? indexDefinition :
         createIndexDefinition(root, definition, indexingContext, asyncIndexing);
     this.indexedNodes = 0;
     this.updateCallback = updateCallback;
     this.extractedTextCache = extractedTextCache;
     this.asyncIndexing = asyncIndexing;
     if (this.definition.isOfOldFormat()){
       indexDefnRewritten = true;
       IndexDefinition.updateDefinition(definition, indexingContext.getIndexPath());
     } else {
       indexDefnRewritten = false;
     }
   }

   public abstract IndexDefinition.Builder newDefinitionBuilder();

   public abstract DocumentMaker<D> newDocumentMaker(IndexDefinition.IndexingRule rule, String path);

   protected FulltextBinaryTextExtractor createBinaryTextExtractor(ExtractedTextCache extractedTextCache,
                                                                   IndexDefinition definition, boolean reindex) {
     return new FulltextBinaryTextExtractor(extractedTextCache, definition, reindex);
   }

   public FulltextIndexWriter<D> getWriter() throws IOException {
     if (writer == null) {
       //Lazy initialization so as to ensure that definition is based
       //on latest NodeBuilder state specially in case of reindexing
       writer = indexWriterFactory.newInstance(definition, definitionBuilder, reindex);
     }
     return writer;
   }

   public IndexingContext getIndexingContext() {
     return indexingContext;
   }

   @Nullable
   public PropertyUpdateCallback getPropertyUpdateCallback() {
     return propertyUpdateCallback;
   }

   public void setPropertyUpdateCallback(PropertyUpdateCallback propertyUpdateCallback) {
     this.propertyUpdateCallback = propertyUpdateCallback;
   }

   /**
    * close writer if it's not null
    */
   public void closeWriter() throws IOException {
     Calendar currentTime = getCalendar();
     final long start = PERF_LOGGER.start();
     boolean indexUpdated = getWriter().close(currentTime.getTimeInMillis());

     if (indexUpdated) {
       PERF_LOGGER.end(start, -1, "Closed writer for directory {}", definition);
       //OAK-2029 Record the last updated status so
       //as to make IndexTracker detect changes when index
       //is stored in file system
       NodeBuilder status = definitionBuilder.child(IndexDefinition.STATUS_NODE);
       status.setProperty(IndexDefinition.STATUS_LAST_UPDATED, getUpdatedTime(currentTime), Type.DATE);
       status.setProperty("indexedNodes", indexedNodes);

       PERF_LOGGER.end(start, -1, "Overall Closed IndexWriter for directory {}", definition);

       if (textExtractor != null){
         textExtractor.done(reindex);
       }
     }
   }

   private String getUpdatedTime(Calendar currentTime) {
     CommitInfo info = getIndexingContext().getCommitInfo();
     String checkpointTime = (String) info.getInfo().get(IndexConstants.CHECKPOINT_CREATION_TIME);
     if (checkpointTime != null) {
       return checkpointTime;
     }
     return ISO8601.format(currentTime);
   }

   /** Only set for testing */
   protected static void setClock(Clock c) {
     checkNotNull(c);
     clock = c;
   }

   static private Calendar getCalendar() {
     Calendar ret = Calendar.getInstance();
     ret.setTime(clock.getDate());
     return ret;
   }

   public void enableReindexMode(){
     reindex = true;
     ReindexOperations reindexOps =
             new ReindexOperations(root, definitionBuilder, definition.getIndexPath(), newDefinitionBuilder());
     definition = reindexOps.apply(indexDefnRewritten);
   }

   public long incIndexedNodes() {
     indexedNodes++;
     return indexedNodes;
   }

   public boolean isAsyncIndexing() {
     return asyncIndexing;
   }

   public long getIndexedNodes() {
     return indexedNodes;
   }

   public void indexUpdate() throws CommitFailedException {
     updateCallback.indexUpdate();
   }

   public IndexDefinition getDefinition() {
     return definition;
   }

   protected FulltextBinaryTextExtractor getTextExtractor(){
     if (textExtractor == null && isAsyncIndexing()){
       //Create lazily to ensure that if its reindex case then update definition is picked
       textExtractor = createBinaryTextExtractor(extractedTextCache, definition, reindex);
     }
     return textExtractor;
   }

   public boolean isReindex() {
     return reindex;
   }

   public static String configureUniqueId(NodeBuilder definition) {
     NodeBuilder status = definition.child(IndexDefinition.STATUS_NODE);
     String uid = status.getString(IndexDefinition.PROP_UID);
     if (uid == null) {
       try {
         uid = String.valueOf(Clock.SIMPLE.getTimeIncreasing());
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         uid = String.valueOf(Clock.SIMPLE.getTime());
       }
       status.setProperty(IndexDefinition.PROP_UID, uid);
     }
     return uid;
   }

   private IndexDefinition createIndexDefinition(NodeState root, NodeBuilder definition, IndexingContext
       indexingContext, boolean asyncIndexing) {
     NodeState defnState = definition.getBaseState();
     if (asyncIndexing && !IndexDefinition.isDisableStoredIndexDefinition()){
       if (definition.getBoolean(PROP_REFRESH_DEFN)){
         definition.removeProperty(PROP_REFRESH_DEFN);
         NodeState clonedState = NodeStateCloner.cloneVisibleState(defnState);
         definition.setChildNode(INDEX_DEFINITION_NODE, clonedState);
         log.info("Refreshed the index definition for [{}]", indexingContext.getIndexPath());
         if (log.isDebugEnabled()){
           log.debug("Updated index definition is {}", NodeStateUtils.toString(clonedState));
         }
       } else if (!definition.hasChildNode(INDEX_DEFINITION_NODE)){
         definition.setChildNode(INDEX_DEFINITION_NODE, NodeStateCloner.cloneVisibleState(defnState));
         log.info("Stored the cloned index definition for [{}]. Changes in index definition would now only be " +
             "effective post reindexing", indexingContext.getIndexPath());
       }
     }
     return newDefinitionBuilder()
             .root(root)
             .defn(defnState)
             .indexPath(indexingContext.getIndexPath())
             .build();
   }
 }
	/*
	* 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.search.spi.editor;

	import java.io.IOException;
	import java.util.Calendar;

	import org.apache.jackrabbit.oak.api.CommitFailedException;
	import org.apache.jackrabbit.oak.api.Type;
	import org.apache.jackrabbit.oak.commons.PerfLogger;
	import org.apache.jackrabbit.oak.plugins.index.IndexConstants;
	import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback;
	import org.apache.jackrabbit.oak.plugins.index.IndexingContext;
	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.PropertyUpdateCallback;
	import org.apache.jackrabbit.oak.plugins.index.search.ReindexOperations;
	import org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
	import org.apache.jackrabbit.oak.plugins.index.search.util.NodeStateCloner;
	import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
	import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
	import org.apache.jackrabbit.oak.spi.state.NodeState;
	import org.apache.jackrabbit.oak.spi.state.NodeStateUtils;
	import org.apache.jackrabbit.oak.stats.Clock;
	import org.apache.jackrabbit.util.ISO8601;
	import org.jetbrains.annotations.Nullable;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import static com.google.common.base.Preconditions.checkNotNull;
	import static org.apache.jackrabbit.oak.plugins.index.search.FulltextIndexConstants.PROP_REFRESH_DEFN;
	import static org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition.INDEX_DEFINITION_NODE;

	/**
	*
	*/
	public abstract class FulltextIndexEditorContext<D> {

	private static final Logger log = LoggerFactory
	.getLogger(FulltextIndexEditorContext.class);

	private static final PerfLogger PERF_LOGGER =
	new PerfLogger(LoggerFactory.getLogger(FulltextIndexEditorContext.class.getName() + ".perf"));

	protected IndexDefinition definition;

	protected final NodeBuilder definitionBuilder;

	private final FulltextIndexWriterFactory<D> indexWriterFactory;

	private FulltextIndexWriter<D> writer = null;

	private long indexedNodes;

	private final IndexUpdateCallback updateCallback;

	private boolean reindex;

	private final ExtractedTextCache extractedTextCache;

	private final NodeState root;

	private final IndexingContext indexingContext;

	private final boolean asyncIndexing;

	//Intentionally static, so that it can be set without passing around clock objects
	//Set for testing ONLY
	private static Clock clock = Clock.SIMPLE;

	private final boolean indexDefnRewritten;

	private FulltextBinaryTextExtractor textExtractor;

	private PropertyUpdateCallback propertyUpdateCallback;

	protected FulltextIndexEditorContext(NodeState root, NodeBuilder definition,
	@Nullable IndexDefinition indexDefinition,
	IndexUpdateCallback updateCallback,
	FulltextIndexWriterFactory indexWriterFactory,
	ExtractedTextCache extractedTextCache,
	IndexingContext indexingContext, boolean asyncIndexing) {
	this.root = root;
	this.indexingContext = checkNotNull(indexingContext);
	this.definitionBuilder = definition;
	this.indexWriterFactory = indexWriterFactory;
	this.definition = indexDefinition != null ? indexDefinition :
	createIndexDefinition(root, definition, indexingContext, asyncIndexing);
	this.indexedNodes = 0;
	this.updateCallback = updateCallback;
	this.extractedTextCache = extractedTextCache;
	this.asyncIndexing = asyncIndexing;
	if (this.definition.isOfOldFormat()){
	indexDefnRewritten = true;
	IndexDefinition.updateDefinition(definition, indexingContext.getIndexPath());
	} else {
	indexDefnRewritten = false;
	}
	}

	public abstract IndexDefinition.Builder newDefinitionBuilder();

	public abstract DocumentMaker<D> newDocumentMaker(IndexDefinition.IndexingRule rule, String path);

	protected FulltextBinaryTextExtractor createBinaryTextExtractor(ExtractedTextCache extractedTextCache,
	IndexDefinition definition, boolean reindex) {
	return new FulltextBinaryTextExtractor(extractedTextCache, definition, reindex);
	}

	public FulltextIndexWriter<D> getWriter() throws IOException {
	if (writer == null) {
	//Lazy initialization so as to ensure that definition is based
	//on latest NodeBuilder state specially in case of reindexing
	writer = indexWriterFactory.newInstance(definition, definitionBuilder, reindex);
	}
	return writer;
	}

	public IndexingContext getIndexingContext() {
	return indexingContext;
	}

	@Nullable
	public PropertyUpdateCallback getPropertyUpdateCallback() {
	return propertyUpdateCallback;
	}

	public void setPropertyUpdateCallback(PropertyUpdateCallback propertyUpdateCallback) {
	this.propertyUpdateCallback = propertyUpdateCallback;
	}

	/**
	* close writer if it's not null
	*/
	public void closeWriter() throws IOException {
	Calendar currentTime = getCalendar();
	final long start = PERF_LOGGER.start();
	boolean indexUpdated = getWriter().close(currentTime.getTimeInMillis());

	if (indexUpdated) {
	PERF_LOGGER.end(start, -1, "Closed writer for directory {}", definition);
	//OAK-2029 Record the last updated status so
	//as to make IndexTracker detect changes when index
	//is stored in file system
	NodeBuilder status = definitionBuilder.child(IndexDefinition.STATUS_NODE);
	status.setProperty(IndexDefinition.STATUS_LAST_UPDATED, getUpdatedTime(currentTime), Type.DATE);
	status.setProperty("indexedNodes", indexedNodes);

	PERF_LOGGER.end(start, -1, "Overall Closed IndexWriter for directory {}", definition);

	if (textExtractor != null){
	textExtractor.done(reindex);
	}
	}
	}

	private String getUpdatedTime(Calendar currentTime) {
	CommitInfo info = getIndexingContext().getCommitInfo();
	String checkpointTime = (String) info.getInfo().get(IndexConstants.CHECKPOINT_CREATION_TIME);
	if (checkpointTime != null) {
	return checkpointTime;
	}
	return ISO8601.format(currentTime);
	}

	/** Only set for testing */
	protected static void setClock(Clock c) {
	checkNotNull(c);
	clock = c;
	}

	static private Calendar getCalendar() {
	Calendar ret = Calendar.getInstance();
	ret.setTime(clock.getDate());
	return ret;
	}

	public void enableReindexMode(){
	reindex = true;
	ReindexOperations reindexOps =
	new ReindexOperations(root, definitionBuilder, definition.getIndexPath(), newDefinitionBuilder());
	definition = reindexOps.apply(indexDefnRewritten);
	}

	public long incIndexedNodes() {
	indexedNodes++;
	return indexedNodes;
	}

	public boolean isAsyncIndexing() {
	return asyncIndexing;
	}

	public long getIndexedNodes() {
	return indexedNodes;
	}

	public void indexUpdate() throws CommitFailedException {
	updateCallback.indexUpdate();
	}

	public IndexDefinition getDefinition() {
	return definition;
	}

	protected FulltextBinaryTextExtractor getTextExtractor(){
	if (textExtractor == null && isAsyncIndexing()){
	//Create lazily to ensure that if its reindex case then update definition is picked
	textExtractor = createBinaryTextExtractor(extractedTextCache, definition, reindex);
	}
	return textExtractor;
	}

	public boolean isReindex() {
	return reindex;
	}

	public static String configureUniqueId(NodeBuilder definition) {
	NodeBuilder status = definition.child(IndexDefinition.STATUS_NODE);
	String uid = status.getString(IndexDefinition.PROP_UID);
	if (uid == null) {
	try {
	uid = String.valueOf(Clock.SIMPLE.getTimeIncreasing());
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	uid = String.valueOf(Clock.SIMPLE.getTime());
	}
	status.setProperty(IndexDefinition.PROP_UID, uid);
	}
	return uid;
	}

	private IndexDefinition createIndexDefinition(NodeState root, NodeBuilder definition, IndexingContext
	indexingContext, boolean asyncIndexing) {
	NodeState defnState = definition.getBaseState();
	if (asyncIndexing && !IndexDefinition.isDisableStoredIndexDefinition()){
	if (definition.getBoolean(PROP_REFRESH_DEFN)){
	definition.removeProperty(PROP_REFRESH_DEFN);
	NodeState clonedState = NodeStateCloner.cloneVisibleState(defnState);
	definition.setChildNode(INDEX_DEFINITION_NODE, clonedState);
	log.info("Refreshed the index definition for [{}]", indexingContext.getIndexPath());
	if (log.isDebugEnabled()){
	log.debug("Updated index definition is {}", NodeStateUtils.toString(clonedState));
	}
	} else if (!definition.hasChildNode(INDEX_DEFINITION_NODE)){
	definition.setChildNode(INDEX_DEFINITION_NODE, NodeStateCloner.cloneVisibleState(defnState));
	log.info("Stored the cloned index definition for [{}]. Changes in index definition would now only be " +
	"effective post reindexing", indexingContext.getIndexPath());
	}
	}
	return newDefinitionBuilder()
	.root(root)
	.defn(defnState)
	.indexPath(indexingContext.getIndexPath())
	.build();
	}
	}