<commit_before>package mil.nga.giat.geowave.datastore.hbase.metadata;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.UUID;

import org.apache.commons.lang.NotImplementedException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import mil.nga.giat.geowave.core.index.ByteArrayId;
import mil.nga.giat.geowave.core.store.CloseableIterator;
import mil.nga.giat.geowave.core.store.adapter.statistics.DataStatistics;
import mil.nga.giat.geowave.core.store.adapter.statistics.DataStatisticsStore;
import mil.nga.giat.geowave.datastore.hbase.operations.BasicHBaseOperations;
import mil.nga.giat.geowave.datastore.hbase.util.HBaseUtils;

public class HBaseDataStatisticsStore extends
		AbstractHBasePersistence<DataStatistics<?>> implements
		DataStatisticsStore
{

	private static final String STATISTICS_CF = "STATS";
	private final static Logger LOGGER = Logger.getLogger(HBaseDataStatisticsStore.class);

	public HBaseDataStatisticsStore(
			final BasicHBaseOperations operations ) {
		super(
				operations);
	}

	@Override
	public void setStatistics(
			final DataStatistics<?> statistics ) {
		removeStatistics(
				statistics.getDataAdapterId(),
				statistics.getStatisticsId());
		addObject(statistics);

	}

	@Override
	public void incorporateStatistics(
			final DataStatistics<?> statistics ) {
		addObject(statistics);

	}

	@Override
	public CloseableIterator<DataStatistics<?>> getDataStatistics(
			final ByteArrayId adapterId,
			final String... authorizations ) {
		return getAllObjectsWithSecondaryId(
				adapterId,
				authorizations);
	}

	@Override
	public CloseableIterator<DataStatistics<?>> getAllDataStatistics(
			final String... authorizations ) {
		return getObjects(authorizations);
	}

	@Override
	public DataStatistics<?> getDataStatistics(
			final ByteArrayId adapterId,
			final ByteArrayId statisticsId,
			final String... authorizations ) {
		return getObject(
				statisticsId,
				adapterId,
				authorizations);
	}

	@Override
	public boolean removeStatistics(
			final ByteArrayId adapterId,
			final ByteArrayId statisticsId,
			final String... authorizations ) {
		if (statisticsId == null) {
			LOGGER.log(Level.ERROR, "No statistics id specified for removeStatistics, ignoring request!");
			return false;
		}
		return deleteObject(
				statisticsId,
				adapterId,
				authorizations);
	}

	@Override
	protected ByteArrayId getPrimaryId(
			final DataStatistics<?> persistedObject ) {
		return persistedObject.getStatisticsId();
	}

	@Override
	protected String getPersistenceTypeName() {
		return STATISTICS_CF;
	}

	@Override
	protected ByteArrayId getSecondaryId(
			final DataStatistics<?> persistedObject ) {
		return persistedObject.getDataAdapterId();
	}

	@Override
	protected DataStatistics<?> entryToValue(
			final Cell entry ) {
		final DataStatistics<?> stats = super.entryToValue(entry);
		if (stats != null) {
			stats.setDataAdapterId(new ByteArrayId(
					CellUtil.cloneQualifier(entry)));
		}
		return stats;
	}

	@Override
	public void removeAllStatistics(
			final ByteArrayId adapterId,
			final String... authorizations ) {
		deleteObjects(
			null,
			adapterId,
			authorizations);

	}

	@Override
	public void transformVisibility(
			final ByteArrayId adapterId,
			final String transformingRegex,
			final String replacement,
			final String... authorizations ) {
		// TODO #406 Need to fix
		LOGGER.error("This method transformVisibility is not yet coded. Need to fix it");

	}

	/**
	 * This function is used to change the scan object created in the superclass to enable 
	 * prefixing.
	 */
	@Override
	protected Scan applyScannerSettings(Scan scanner, ByteArrayId primaryId, ByteArrayId secondaryId) {
		Scan scan = super.applyScannerSettings(scanner, primaryId, secondaryId);
		if (primaryId != null) {
			// So this will set the stop row to just after all the possible suffixes to this primaryId.
			scan.setStopRow(HBaseUtils.getNextPrefix(scan.getStopRow()));
		}
		return scan;
	}
	
	/**
	 * This function converts results and merges data statistic elements together that have
	 * the same id.
	 */
	@Override
	protected Iterator<DataStatistics<?>> getNativeIteratorWrapper(Iterator<Result> resultIterator) {
		return new StatisticsNativeIteratorWrapper(resultIterator);
	}

	/**
	 * A special version of NativeIteratorWrapper (defined in the parent) which 
	 * will combine records that have the same dataid & statsId 
	 */
	private class StatisticsNativeIteratorWrapper implements Iterator<DataStatistics<?>> {
		final private Iterator<Result> it;
		private DataStatistics<?> nextVal = null;

		public StatisticsNativeIteratorWrapper(Iterator<Result> resultIterator) {
			this.it = resultIterator;
		}

		@Override
		public boolean hasNext() {
			return nextVal != null || it.hasNext();
		}

		@Override
		public DataStatistics<?> next() {
			DataStatistics<?> currentStatistics = nextVal;
			nextVal = null;
			while (it.hasNext()) {
				Cell cell = it.next().listCells().get(0);
				
				// This entryToValue function has the side effect of adding the object to the cache.  
				// We need to make sure to add the merged version of the stat at the end of this
				// function, before it is returned.
				DataStatistics<?> statEntry = entryToValue(cell);
				
				if (currentStatistics == null) {
					currentStatistics = statEntry;
				}
				else {
					if (statEntry.getStatisticsId().equals(currentStatistics.getStatisticsId()) 
							&& statEntry.getDataAdapterId().equals(currentStatistics.getDataAdapterId())) {
						currentStatistics.merge(statEntry);
					}
					else {
						nextVal = statEntry;
						break;
					}
				}
			}
			
			// Add this entry to cache (see comment above)
			addObjectToCache(currentStatistics);
			return currentStatistics;
		}

		@Override
		public void remove() {			
			throw new NotImplementedException("Transforming iterator cannot use remove()");
		}

	}	
	
	/**
	 * This function will append a UUID to the record that's inserted into the database.
	 * @param object
	 * @return
	 */
	@Override
	protected ByteArrayId getRowId(DataStatistics<?> object) {
		byte[] parentRecord = super.getRowId(object).getBytes();
		ByteBuffer parentBuffer = ByteBuffer.allocate(parentRecord.length + 16);
		parentBuffer.put(parentRecord);
		UUID uuid = UUID.randomUUID();
		parentBuffer.putLong(uuid.getMostSignificantBits());
		parentBuffer.putLong(uuid.getLeastSignificantBits());
		return new ByteArrayId(parentBuffer.array());
	}
}<commit_msg>GEOWAVE-546 - Fix issue with row ordering<commit_after>package mil.nga.giat.geowave.datastore.hbase.metadata;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.UUID;

import org.apache.commons.lang.NotImplementedException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import mil.nga.giat.geowave.core.index.ByteArrayId;
import mil.nga.giat.geowave.core.store.CloseableIterator;
import mil.nga.giat.geowave.core.store.adapter.statistics.DataStatistics;
import mil.nga.giat.geowave.core.store.adapter.statistics.DataStatisticsStore;
import mil.nga.giat.geowave.datastore.hbase.operations.BasicHBaseOperations;
import mil.nga.giat.geowave.datastore.hbase.util.HBaseUtils;

public class HBaseDataStatisticsStore extends
		AbstractHBasePersistence<DataStatistics<?>> implements
		DataStatisticsStore
{

	private static final String STATISTICS_CF = "STATS";
	private final static Logger LOGGER = Logger.getLogger(HBaseDataStatisticsStore.class);

	public HBaseDataStatisticsStore(
			final BasicHBaseOperations operations ) {
		super(
				operations);
	}

	@Override
	public void setStatistics(
			final DataStatistics<?> statistics ) {
		removeStatistics(
				statistics.getDataAdapterId(),
				statistics.getStatisticsId());
		addObject(statistics);

	}

	@Override
	public void incorporateStatistics(
			final DataStatistics<?> statistics ) {
		addObject(statistics);

	}

	@Override
	public CloseableIterator<DataStatistics<?>> getDataStatistics(
			final ByteArrayId adapterId,
			final String... authorizations ) {
		return getAllObjectsWithSecondaryId(
				adapterId,
				authorizations);
	}

	@Override
	public CloseableIterator<DataStatistics<?>> getAllDataStatistics(
			final String... authorizations ) {
		return getObjects(authorizations);
	}

	@Override
	public DataStatistics<?> getDataStatistics(
			final ByteArrayId adapterId,
			final ByteArrayId statisticsId,
			final String... authorizations ) {
		return getObject(
				statisticsId,
				adapterId,
				authorizations);
	}

	@Override
	public boolean removeStatistics(
			final ByteArrayId adapterId,
			final ByteArrayId statisticsId,
			final String... authorizations ) {
		if (statisticsId == null) {
			LOGGER.log(Level.ERROR, "No statistics id specified for removeStatistics, ignoring request!");
			return false;
		}
		return deleteObject(
				statisticsId,
				adapterId,
				authorizations);
	}

	@Override
	protected ByteArrayId getPrimaryId(
			final DataStatistics<?> persistedObject ) {
		return persistedObject.getStatisticsId();
	}

	@Override
	protected String getPersistenceTypeName() {
		return STATISTICS_CF;
	}

	@Override
	protected ByteArrayId getSecondaryId(
			final DataStatistics<?> persistedObject ) {
		return persistedObject.getDataAdapterId();
	}

	@Override
	protected DataStatistics<?> entryToValue(
			final Cell entry ) {
		final DataStatistics<?> stats = super.entryToValue(entry);
		if (stats != null) {
			stats.setDataAdapterId(new ByteArrayId(
					CellUtil.cloneQualifier(entry)));
		}
		return stats;
	}

	@Override
	public void removeAllStatistics(
			final ByteArrayId adapterId,
			final String... authorizations ) {
		deleteObjects(
			null,
			adapterId,
			authorizations);

	}

	@Override
	public void transformVisibility(
			final ByteArrayId adapterId,
			final String transformingRegex,
			final String replacement,
			final String... authorizations ) {
		// TODO #406 Need to fix
		LOGGER.error("This method transformVisibility is not yet coded. Need to fix it");

	}

	/**
	 * This function is used to change the scan object created in the superclass to enable 
	 * prefixing.
	 */
	@Override
	protected Scan applyScannerSettings(Scan scanner, ByteArrayId primaryId, ByteArrayId secondaryId) {
		Scan scan = super.applyScannerSettings(scanner, primaryId, secondaryId);
		if (primaryId != null) {
			// So this will set the stop row to just after all the possible suffixes to this primaryId.
			scan.setStopRow(HBaseUtils.getNextPrefix(scan.getStopRow()));
		}
		return scan;
	}
	
	/**
	 * This function converts results and merges data statistic elements together that have
	 * the same id.
	 */
	@Override
	protected Iterator<DataStatistics<?>> getNativeIteratorWrapper(Iterator<Result> resultIterator) {
		return new StatisticsNativeIteratorWrapper(resultIterator);
	}

	/**
	 * A special version of NativeIteratorWrapper (defined in the parent) which 
	 * will combine records that have the same dataid & statsId 
	 */
	private class StatisticsNativeIteratorWrapper implements Iterator<DataStatistics<?>> {
		final private Iterator<Result> it;
		private DataStatistics<?> nextVal = null;

		public StatisticsNativeIteratorWrapper(Iterator<Result> resultIterator) {
			this.it = resultIterator;
		}

		@Override
		public boolean hasNext() {
			return nextVal != null || it.hasNext();
		}

		@Override
		public DataStatistics<?> next() {
			DataStatistics<?> currentStatistics = nextVal;
			nextVal = null;
			while (it.hasNext()) {
				Cell cell = it.next().listCells().get(0);
				
				// This entryToValue function has the side effect of adding the object to the cache.  
				// We need to make sure to add the merged version of the stat at the end of this
				// function, before it is returned.
				DataStatistics<?> statEntry = entryToValue(cell);
				
				if (currentStatistics == null) {
					currentStatistics = statEntry;
				}
				else {
					if (statEntry.getStatisticsId().equals(currentStatistics.getStatisticsId()) 
							&& statEntry.getDataAdapterId().equals(currentStatistics.getDataAdapterId())) {
						currentStatistics.merge(statEntry);
					}
					else {
						nextVal = statEntry;
						break;
					}
				}
			}
			
			// Add this entry to cache (see comment above)
			addObjectToCache(currentStatistics);
			return currentStatistics;
		}

		@Override
		public void remove() {			
			throw new NotImplementedException("Transforming iterator cannot use remove()");
		}

	}	
	
	/**
	 * This function will append a UUID to the record that's inserted into the database.
	 * @param object
	 * @return
	 */
	@Override
	protected ByteArrayId getRowId(DataStatistics<?> object) {
		byte[] parentRecord = super.getRowId(object).getBytes();
		ByteBuffer parentBuffer = ByteBuffer.allocate(parentRecord.length + 1 + 16);
		parentBuffer.put(parentRecord);
		UUID uuid = UUID.randomUUID();
		parentBuffer.put(new byte[] { 0 });
		parentBuffer.putLong(uuid.getMostSignificantBits());
		parentBuffer.putLong(uuid.getLeastSignificantBits());
		return new ByteArrayId(parentBuffer.array());
	}
}
{'fraction_non_alphanumeric': 0.09475806451612903, 'fraction_numerical': 0.0015681003584229391, 'mean_word_length': 4.27906976744186, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6208730', 'n_tokens_mistral': 4396, 'n_tokens_neox': 3820, 'n_words': 991}