<commit_before>/*
 * Copyright 1999-2010 <NAME> (l.garulli--at--orientechnologies.com)
 *
 * Licensed 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 com.orientechnologies.orient.core.index;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import com.orientechnologies.common.collection.ONavigableMap;
import com.orientechnologies.common.concur.resource.OSharedResourceAbstract;
import com.orientechnologies.common.listener.OProgressListener;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.common.profiler.OProfiler;
import com.orientechnologies.common.profiler.OProfiler.OProfilerHookValue;
import com.orientechnologies.orient.core.OMemoryWatchDog.Listener;
import com.orientechnologies.orient.core.Orient;
import com.orientechnologies.orient.core.annotation.ODocumentInstance;
import com.orientechnologies.orient.core.db.ODatabase;
import com.orientechnologies.orient.core.db.ODatabaseListener;
import com.orientechnologies.orient.core.db.ODatabaseRecordThreadLocal;
import com.orientechnologies.orient.core.db.record.ODatabaseRecord;
import com.orientechnologies.orient.core.db.record.OIdentifiable;
import com.orientechnologies.orient.core.db.record.ORecordElement;
import com.orientechnologies.orient.core.db.record.ORecordLazySet;
import com.orientechnologies.orient.core.id.ORID;
import com.orientechnologies.orient.core.intent.OIntentMassiveInsert;
import com.orientechnologies.orient.core.metadata.schema.OType;
import com.orientechnologies.orient.core.record.ORecord;
import com.orientechnologies.orient.core.record.impl.ODocument;
import com.orientechnologies.orient.core.record.impl.ORecordBytes;
import com.orientechnologies.orient.core.serialization.serializer.OStringSerializerHelper;
import com.orientechnologies.orient.core.serialization.serializer.record.string.ORecordSerializerStringAbstract;
import com.orientechnologies.orient.core.serialization.serializer.stream.OStreamSerializerListRID;
import com.orientechnologies.orient.core.serialization.serializer.stream.OStreamSerializerLiteral;
import com.orientechnologies.orient.core.tx.OTransactionIndexChanges.OPERATION;
import com.orientechnologies.orient.core.type.tree.OMVRBTreeDatabaseLazySave;

/**
 * Handles indexing when records change.
 * 
 * @author <NAME>
 * 
 */
public abstract class OIndexMVRBTreeAbstract extends OSharedResourceAbstract implements OIndexInternal, ODatabaseListener {
	protected static final String																		CONFIG_MAP_RID	= "mapRid";
	protected static final String																		CONFIG_CLUSTERS	= "clusters";
	protected String																								name;
	protected String																								type;
	protected OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>	map;
	protected Set<String>																						clustersToIndex	= new LinkedHashSet<String>();
	protected OIndexCallback																				callback;
	protected boolean																								automatic;

	@ODocumentInstance
	protected ODocument																							configuration;
	private Listener																								watchDog;
	private volatile int																						optimization		= 0;

	public OIndexMVRBTreeAbstract(final String iType) {
		type = iType;

		watchDog = new Listener() {
			public void memoryUsageLow(final TYPE iType, final long usedMemory, final long maxMemory) {
				if (iType == TYPE.JVM)
					optimization = 1;
			}

			public void memoryUsageCritical(final TYPE iType, final long usedMemory, final long maxMemory) {
				if (iType == TYPE.JVM)
					optimization = 2;
			}
		};
	}

	public void flush() {
		lazySave();
	}

	/**
	 * Creates the index.
	 * 
	 * @param iDatabase
	 *          Current Database instance
	 * @param iProperty
	 *          Owner property
	 * @param iClusterIndexName
	 *          Cluster name where to place the TreeMap
	 * @param iProgressListener
	 *          Listener to get called on progress
	 */
	public OIndexInternal create(final String iName, final ODatabaseRecord iDatabase, final String iClusterIndexName,
			final int[] iClusterIdsToIndex, final OProgressListener iProgressListener, final boolean iAutomatic) {
		acquireExclusiveLock();
		try {

			name = iName;
			configuration = new ODocument(iDatabase);
			automatic = iAutomatic;

			if (iClusterIdsToIndex != null)
				for (int id : iClusterIdsToIndex)
					clustersToIndex.add(iDatabase.getClusterNameById(id));

			map = new OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>(iDatabase, iClusterIndexName,
					OStreamSerializerLiteral.INSTANCE, OStreamSerializerListRID.INSTANCE);

			installHooks(iDatabase);

			if (automatic)
				rebuild(iProgressListener);
			updateConfiguration();
			return this;

		} finally {
			releaseExclusiveLock();
		}
	}

	public OIndexInternal loadFromConfiguration(final ODocument iConfig) {
		acquireExclusiveLock();
		try {

			final ORID rid = (ORID) iConfig.field(CONFIG_MAP_RID, ORID.class);
			if (rid == null)
				return null;

			configuration = iConfig;
			name = configuration.field(OIndexInternal.CONFIG_NAME);
			automatic = (Boolean) (configuration.field(OIndexInternal.CONFIG_AUTOMATIC) != null ? configuration
					.field(OIndexInternal.CONFIG_AUTOMATIC) : true);
			clustersToIndex.clear();

			final Collection<? extends String> clusters = configuration.field(CONFIG_CLUSTERS);
			if (clusters != null)
				clustersToIndex.addAll(clusters);

			map = new OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>(getDatabase(), rid);
			map.load();

			installHooks(iConfig.getDatabase());

			return this;

		} finally {
			releaseExclusiveLock();
		}
	}

	public Set<OIdentifiable> get(final Object iKey) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			final ORecordLazySet values = (ORecordLazySet) map.get(iKey);
			if (values != null)
				values.setDatabase(ODatabaseRecordThreadLocal.INSTANCE.get());

			if (values == null)
				return ORecordLazySet.EMPTY_SET;

			return values;

		} finally {
			releaseExclusiveLock();
		}
	}

	public boolean contains(final Object iKey) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			return map.containsKey(iKey);

		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with key between the range passed as parameter. Range bounds are included.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @see #getValuesBetween(Object, Object, boolean)
	 * @return
	 */
	public Set<OIdentifiable> getValuesBetween(final Object iRangeFrom, final Object iRangeTo) {
		return getValuesBetween(iRangeFrom, iRangeTo, true);
	}

	/**
	 * Returns a set of records with keys greater than passed parameter.
	 * 
	 * @param fromKey
	 *          Starting key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with keys greater than passed parameter.
	 */
	public Collection<OIdentifiable> getValuesMajor(Object fromKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.tailMap(fromKey, isInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents that contains fields ("key", "rid") where "key" - index key, "rid" - record id of records with keys
	 * greater than passed parameter.
	 * 
	 * @param fromKey
	 *          Starting key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with key greater than passed parameter.
	 */
	public Collection<ODocument> getEntriesMajor(Object fromKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.tailMap(fromKey, isInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with keys less than passed parameter.
	 * 
	 * @param toKey
	 *          Ending key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with keys less than passed parameter.
	 */
	public Collection<OIdentifiable> getValuesMinor(Object toKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.headMap(toKey, isInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents that contains fields ("key", "rid") where "key" - index key, "rid" - record id of records with keys
	 * less than passed parameter.
	 * 
	 * @param toKey
	 *          Ending key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with key greater than passed parameter.
	 */
	public Collection<ODocument> getEntriesMinor(Object toKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.headMap(toKey, isInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with key between the range passed as parameter.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @param iInclusive
	 *          Include from/to bounds
	 * @see #getValuesBetween(Object, Object)
	 * @return
	 */
	public Set<OIdentifiable> getValuesBetween(final Object iRangeFrom, final Object iRangeTo, final boolean iInclusive) {
		if (iRangeFrom.getClass() != iRangeTo.getClass())
			throw new IllegalArgumentException("Range from-to parameters are of different types");

		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.subMap(iRangeFrom, iInclusive, iRangeTo, iInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;

		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents with key between the range passed as parameter. Range bounds are included.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @see #getEntriesBetween(Object, Object, boolean)
	 * @return
	 */
	public Set<ODocument> getEntriesBetween(final Object iRangeFrom, final Object iRangeTo) {
		return getEntriesBetween(iRangeFrom, iRangeTo, true);
	}

	/**
	 * Returns a set of documents with key between the range passed as parameter.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @param iInclusive
	 *          Include from/to bounds
	 * @see #getEntriesBetween(Object, Object)
	 * @return
	 */
	public Set<ODocument> getEntriesBetween(final Object iRangeFrom, final Object iRangeTo, final boolean iInclusive) {
		if (iRangeFrom.getClass() != iRangeTo.getClass())
			throw new IllegalArgumentException("Range from-to parameters are of different types");

		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.subMap(iRangeFrom, iInclusive, iRangeTo, iInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;

		} finally {
			releaseExclusiveLock();
		}
	}

	public ORID getIdentity() {
		return map.getRecord().getIdentity();
	}

	public OIndexInternal rebuild() {
		return rebuild(null);
	}

	/**
	 * Populates the index with all the existent records. Uses the massive insert intent to speed up and keep the consumed memory low.
	 */
	public OIndexInternal rebuild(final OProgressListener iProgressListener) {
		clear();

		final boolean intentInstalled = getDatabase().declareIntent(new OIntentMassiveInsert());

		checkForOptimization();
		acquireExclusiveLock();
		try {

			int documentIndexed = 0;
			int documentNum = 0;
			long documentTotal = 0;

			for (String cluster : clustersToIndex)
				documentTotal += getDatabase().countClusterElements(cluster);

			if (iProgressListener != null)
				iProgressListener.onBegin(this, documentTotal);

			for (String clusterName : clustersToIndex)
				for (ORecord<?> record : getDatabase().browseCluster(clusterName)) {
					if (record instanceof ODocument) {
						final ODocument doc = (ODocument) record;
						final Object fieldValue = callback.getDocumentValueToIndex(doc);

						if (fieldValue != null) {
							put(fieldValue, doc);
							++documentIndexed;
						}
					}
					documentNum++;

					if (iProgressListener != null)
						iProgressListener.onProgress(this, documentNum, documentNum * 100f / documentTotal);
				}

			lazySave();

			if (iProgressListener != null)
				iProgressListener.onCompletition(this, true);

		} catch (Exception e) {
			if (iProgressListener != null)
				iProgressListener.onCompletition(this, false);

			clear();

			throw new OIndexException("Error on rebuilding the index for clusters: " + clustersToIndex, e);

		} finally {
			if (intentInstalled)
				getDatabase().declareIntent(null);
			
			releaseExclusiveLock();
		}

		return this;
	}

	public boolean remove(final Object iKey, final OIdentifiable iValue) {
		return remove(iKey);
	}

	public boolean remove(final Object key) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			return map.remove(key) != null;

		} finally {
			releaseExclusiveLock();<commit_msg>Index: replaced rebuild of index always<commit_after>/*
 * Copyright 1999-2010 <NAME> (l.garulli--at--orientechnologies.com)
 *
 * Licensed 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 com.orientechnologies.orient.core.index;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;

import com.orientechnologies.common.collection.ONavigableMap;
import com.orientechnologies.common.concur.resource.OSharedResourceAbstract;
import com.orientechnologies.common.listener.OProgressListener;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.common.profiler.OProfiler;
import com.orientechnologies.common.profiler.OProfiler.OProfilerHookValue;
import com.orientechnologies.orient.core.OMemoryWatchDog.Listener;
import com.orientechnologies.orient.core.Orient;
import com.orientechnologies.orient.core.annotation.ODocumentInstance;
import com.orientechnologies.orient.core.db.ODatabase;
import com.orientechnologies.orient.core.db.ODatabaseListener;
import com.orientechnologies.orient.core.db.ODatabaseRecordThreadLocal;
import com.orientechnologies.orient.core.db.record.ODatabaseRecord;
import com.orientechnologies.orient.core.db.record.OIdentifiable;
import com.orientechnologies.orient.core.db.record.ORecordElement;
import com.orientechnologies.orient.core.db.record.ORecordLazySet;
import com.orientechnologies.orient.core.id.ORID;
import com.orientechnologies.orient.core.intent.OIntentMassiveInsert;
import com.orientechnologies.orient.core.metadata.schema.OType;
import com.orientechnologies.orient.core.record.ORecord;
import com.orientechnologies.orient.core.record.impl.ODocument;
import com.orientechnologies.orient.core.record.impl.ORecordBytes;
import com.orientechnologies.orient.core.serialization.serializer.OStringSerializerHelper;
import com.orientechnologies.orient.core.serialization.serializer.record.string.ORecordSerializerStringAbstract;
import com.orientechnologies.orient.core.serialization.serializer.stream.OStreamSerializerListRID;
import com.orientechnologies.orient.core.serialization.serializer.stream.OStreamSerializerLiteral;
import com.orientechnologies.orient.core.tx.OTransactionIndexChanges.OPERATION;
import com.orientechnologies.orient.core.type.tree.OMVRBTreeDatabaseLazySave;

/**
 * Handles indexing when records change.
 * 
 * @author <NAME>
 * 
 */
public abstract class OIndexMVRBTreeAbstract extends OSharedResourceAbstract implements OIndexInternal, ODatabaseListener {
	protected static final String																		CONFIG_MAP_RID	= "mapRid";
	protected static final String																		CONFIG_CLUSTERS	= "clusters";
	protected String																								name;
	protected String																								type;
	protected OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>	map;
	protected Set<String>																						clustersToIndex	= new LinkedHashSet<String>();
	protected OIndexCallback																				callback;
	protected boolean																								automatic;

	@ODocumentInstance
	protected ODocument																							configuration;
	private Listener																								watchDog;
	private volatile int																						optimization		= 0;

	public OIndexMVRBTreeAbstract(final String iType) {
		type = iType;

		watchDog = new Listener() {
			public void memoryUsageLow(final TYPE iType, final long usedMemory, final long maxMemory) {
				if (iType == TYPE.JVM)
					optimization = 1;
			}

			public void memoryUsageCritical(final TYPE iType, final long usedMemory, final long maxMemory) {
				if (iType == TYPE.JVM)
					optimization = 2;
			}
		};
	}

	public void flush() {
		lazySave();
	}

	/**
	 * Creates the index.
	 * 
	 * @param iDatabase
	 *          Current Database instance
	 * @param iProperty
	 *          Owner property
	 * @param iClusterIndexName
	 *          Cluster name where to place the TreeMap
	 * @param iProgressListener
	 *          Listener to get called on progress
	 */
	public OIndexInternal create(final String iName, final ODatabaseRecord iDatabase, final String iClusterIndexName,
			final int[] iClusterIdsToIndex, final OProgressListener iProgressListener, final boolean iAutomatic) {
		acquireExclusiveLock();
		try {

			name = iName;
			configuration = new ODocument(iDatabase);
			automatic = iAutomatic;

			if (iClusterIdsToIndex != null)
				for (int id : iClusterIdsToIndex)
					clustersToIndex.add(iDatabase.getClusterNameById(id));

			map = new OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>(iDatabase, iClusterIndexName,
					OStreamSerializerLiteral.INSTANCE, OStreamSerializerListRID.INSTANCE);

			installHooks(iDatabase);

			rebuild(iProgressListener);
			updateConfiguration();
			return this;

		} finally {
			releaseExclusiveLock();
		}
	}

	public OIndexInternal loadFromConfiguration(final ODocument iConfig) {
		acquireExclusiveLock();
		try {

			final ORID rid = (ORID) iConfig.field(CONFIG_MAP_RID, ORID.class);
			if (rid == null)
				return null;

			configuration = iConfig;
			name = configuration.field(OIndexInternal.CONFIG_NAME);
			automatic = (Boolean) (configuration.field(OIndexInternal.CONFIG_AUTOMATIC) != null ? configuration
					.field(OIndexInternal.CONFIG_AUTOMATIC) : true);
			clustersToIndex.clear();

			final Collection<? extends String> clusters = configuration.field(CONFIG_CLUSTERS);
			if (clusters != null)
				clustersToIndex.addAll(clusters);

			map = new OMVRBTreeDatabaseLazySave<Object, Set<OIdentifiable>>(getDatabase(), rid);
			map.load();

			installHooks(iConfig.getDatabase());

			return this;

		} finally {
			releaseExclusiveLock();
		}
	}

	public Set<OIdentifiable> get(final Object iKey) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			final ORecordLazySet values = (ORecordLazySet) map.get(iKey);
			if (values != null)
				values.setDatabase(ODatabaseRecordThreadLocal.INSTANCE.get());

			if (values == null)
				return ORecordLazySet.EMPTY_SET;

			return values;

		} finally {
			releaseExclusiveLock();
		}
	}

	public boolean contains(final Object iKey) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			return map.containsKey(iKey);

		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with key between the range passed as parameter. Range bounds are included.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @see #getValuesBetween(Object, Object, boolean)
	 * @return
	 */
	public Set<OIdentifiable> getValuesBetween(final Object iRangeFrom, final Object iRangeTo) {
		return getValuesBetween(iRangeFrom, iRangeTo, true);
	}

	/**
	 * Returns a set of records with keys greater than passed parameter.
	 * 
	 * @param fromKey
	 *          Starting key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with keys greater than passed parameter.
	 */
	public Collection<OIdentifiable> getValuesMajor(Object fromKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.tailMap(fromKey, isInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents that contains fields ("key", "rid") where "key" - index key, "rid" - record id of records with keys
	 * greater than passed parameter.
	 * 
	 * @param fromKey
	 *          Starting key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with key greater than passed parameter.
	 */
	public Collection<ODocument> getEntriesMajor(Object fromKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.tailMap(fromKey, isInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with keys less than passed parameter.
	 * 
	 * @param toKey
	 *          Ending key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with keys less than passed parameter.
	 */
	public Collection<OIdentifiable> getValuesMinor(Object toKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.headMap(toKey, isInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents that contains fields ("key", "rid") where "key" - index key, "rid" - record id of records with keys
	 * less than passed parameter.
	 * 
	 * @param toKey
	 *          Ending key.
	 * @param isInclusive
	 *          Indicates whether record with passed key will be included.
	 * 
	 * @return set of records with key greater than passed parameter.
	 */
	public Collection<ODocument> getEntriesMinor(Object toKey, boolean isInclusive) {
		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.headMap(toKey, isInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;
		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of records with key between the range passed as parameter.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @param iInclusive
	 *          Include from/to bounds
	 * @see #getValuesBetween(Object, Object)
	 * @return
	 */
	public Set<OIdentifiable> getValuesBetween(final Object iRangeFrom, final Object iRangeTo, final boolean iInclusive) {
		if (iRangeFrom.getClass() != iRangeTo.getClass())
			throw new IllegalArgumentException("Range from-to parameters are of different types");

		checkForOptimization();
		acquireExclusiveLock();

		try {
			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.subMap(iRangeFrom, iInclusive, iRangeTo, iInclusive);
			if (subSet == null)
				return ORecordLazySet.EMPTY_SET;

			final Set<OIdentifiable> result = new ORecordLazySet(configuration.getDatabase());
			for (Set<OIdentifiable> v : subSet.values()) {
				result.addAll(v);
			}

			return result;

		} finally {
			releaseExclusiveLock();
		}
	}

	/**
	 * Returns a set of documents with key between the range passed as parameter. Range bounds are included.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @see #getEntriesBetween(Object, Object, boolean)
	 * @return
	 */
	public Set<ODocument> getEntriesBetween(final Object iRangeFrom, final Object iRangeTo) {
		return getEntriesBetween(iRangeFrom, iRangeTo, true);
	}

	/**
	 * Returns a set of documents with key between the range passed as parameter.
	 * 
	 * @param iRangeFrom
	 *          Starting range
	 * @param iRangeTo
	 *          Ending range
	 * @param iInclusive
	 *          Include from/to bounds
	 * @see #getEntriesBetween(Object, Object)
	 * @return
	 */
	public Set<ODocument> getEntriesBetween(final Object iRangeFrom, final Object iRangeTo, final boolean iInclusive) {
		if (iRangeFrom.getClass() != iRangeTo.getClass())
			throw new IllegalArgumentException("Range from-to parameters are of different types");

		checkForOptimization();
		acquireExclusiveLock();

		try {
			final Set<ODocument> result = new HashSet<ODocument>();

			final ONavigableMap<Object, Set<OIdentifiable>> subSet = map.subMap(iRangeFrom, iInclusive, iRangeTo, iInclusive);
			if (subSet != null) {
				for (Entry<Object, Set<OIdentifiable>> v : subSet.entrySet()) {
					for (OIdentifiable id : v.getValue()) {
						final ODocument document = new ODocument();
						document.field("key", v.getKey());
						document.field("rid", id.getIdentity());
						document.unsetDirty();
						result.add(document);
					}
				}
			}

			return result;

		} finally {
			releaseExclusiveLock();
		}
	}

	public ORID getIdentity() {
		return map.getRecord().getIdentity();
	}

	public OIndexInternal rebuild() {
		return rebuild(null);
	}

	/**
	 * Populates the index with all the existent records. Uses the massive insert intent to speed up and keep the consumed memory low.
	 */
	public OIndexInternal rebuild(final OProgressListener iProgressListener) {
		clear();

		final boolean intentInstalled = getDatabase().declareIntent(new OIntentMassiveInsert());

		checkForOptimization();
		acquireExclusiveLock();
		try {

			int documentIndexed = 0;
			int documentNum = 0;
			long documentTotal = 0;

			for (String cluster : clustersToIndex)
				documentTotal += getDatabase().countClusterElements(cluster);

			if (iProgressListener != null)
				iProgressListener.onBegin(this, documentTotal);

			for (String clusterName : clustersToIndex)
				for (ORecord<?> record : getDatabase().browseCluster(clusterName)) {
					if (record instanceof ODocument) {
						final ODocument doc = (ODocument) record;
						final Object fieldValue = callback.getDocumentValueToIndex(doc);

						if (fieldValue != null) {
							put(fieldValue, doc);
							++documentIndexed;
						}
					}
					documentNum++;

					if (iProgressListener != null)
						iProgressListener.onProgress(this, documentNum, documentNum * 100f / documentTotal);
				}

			lazySave();

			if (iProgressListener != null)
				iProgressListener.onCompletition(this, true);

		} catch (Exception e) {
			if (iProgressListener != null)
				iProgressListener.onCompletition(this, false);

			clear();

			throw new OIndexException("Error on rebuilding the index for clusters: " + clustersToIndex, e);

		} finally {
			if (intentInstalled)
				getDatabase().declareIntent(null);

			releaseExclusiveLock();
		}

		return this;
	}

	public boolean remove(final Object iKey, final OIdentifiable iValue) {
		return remove(iKey);
	}

	public boolean remove(final Object key) {
		checkForOptimization();
		acquireExclusiveLock();
		try {

			return map.remove(key) != null;

		} finally {
			releaseExclusiveLock();
{'fraction_non_alphanumeric': 0.09887378885113879, 'fraction_numerical': 0.0013212533031332577, 'mean_word_length': 3.732618728599077, 'pattern_counts': {'":': 0, '<': 105, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '423379', 'n_tokens_mistral': 10681, 'n_tokens_neox': 9171, 'n_words': 2380}