<commit_before>package controllers.impl;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import play.Play;
import play.PlayPlugin;

import com.alvazan.orm.api.base.NoSqlEntityManager;
import com.alvazan.orm.api.z3api.NoSqlTypedSession;
import com.alvazan.orm.api.z5api.NoSqlSession;
import com.alvazan.orm.api.z8spi.action.Column;
import com.alvazan.orm.api.z8spi.conv.StandardConverters;
import com.alvazan.orm.api.z8spi.conv.StorageTypeEnum;
import com.alvazan.orm.api.z8spi.iter.AbstractCursor;
import com.alvazan.orm.api.z8spi.iter.Cursor;
import com.alvazan.orm.api.z8spi.meta.DboColumnIdMeta;
import com.alvazan.orm.api.z8spi.meta.DboColumnMeta;
import com.alvazan.orm.api.z8spi.meta.DboTableMeta;
import com.alvazan.orm.api.z8spi.meta.TypedColumn;
import com.alvazan.orm.api.z8spi.meta.TypedRow;
import com.alvazan.play.NoSql;

import controllers.api.ApiPostDataPointsImpl;
import controllers.modules2.RawTimeSeriesProcessor;
import controllers.modules2.framework.ReadResult;
import controllers.modules2.framework.TSRelational;
import controllers.modules2.framework.VisitorInfo;
import controllers.modules2.framework.procs.RowMeta;

public class RawTimeSeriesImpl  extends PlayPlugin {

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

	public static final TSRelational END_OF_DATA_TOKEN = new TSRelational();
	
	protected DboTableMeta meta;
	protected DboColumnIdMeta idcolMeta;

	protected byte[] endBytes;
	protected byte[] startBytes;
	protected Long currentPartitionId;
	protected Long start;
	protected Long end;
	protected AbstractCursor<Column> cursor;
	protected Long partitionSize;
	private List<DboColumnMeta> colMeta = new ArrayList<DboColumnMeta>();
	protected int currentIndex;
	protected List<Long> existingPartitions = new ArrayList<Long>();
	protected NoSqlEntityManager mgr;
	protected BlockingQueue<TSRelational> itemsQueue;
	protected int buffersize = 2000;

	private String timeColumn;
	private String valueColumn;
	
	private final static Charset charset = StandardCharsets.UTF_8;

	public void init(DboTableMeta meta, Long start, Long end, RowMeta rowMeta, NoSqlEntityManager mgr) {
		this.mgr = mgr;
		this.meta = meta;
		idcolMeta = meta.getIdColumnMeta();

		
		this.timeColumn = rowMeta.getTimeColumn();
		this.valueColumn = rowMeta.getValueColumn();
		
		this.start = start;
		this.end = end;
		loadPartitions(meta);
		partitionSize = meta.getTimeSeriesPartionSize();
		currentIndex = partition(partitionSize);

		if(start != null)
			this.startBytes = meta.getIdColumnMeta().convertToStorage2(new BigInteger(start+""));
		if(end != null)
			this.endBytes = meta.getIdColumnMeta().convertToStorage2(new BigInteger(end+""));

		for (DboColumnMeta thismeta:meta.getAllColumns())
			colMeta.add(thismeta);
		if (log.isInfoEnabled())
			log.info("Setting up for reading partitions, partId="+currentIndex+" partitions="+existingPartitions+" start="+start);
		
		//jsc fire off read thread
		String buffersizeString = Play.configuration.getProperty("databus.preread.buffer.size", "2000");
		buffersize = Integer.parseInt(buffersizeString);
		itemsQueue = new LinkedBlockingQueue<TSRelational>(buffersize);
		new Thread(new AgressivePreReader()).start();
	}
	
	public void deleteRange() {
		int flushSize = 1000;
		NoSqlTypedSession s = mgr.getTypedSession();
		
		TSRelational ts = nextResult();
		long startTime = System.currentTimeMillis();
		long count = 0;
		while (ts != null) {
			//slow way:  get row, delete column.
			TypedRow partition = getPartitionForTime(ts.getTime());

			partition.removeColumn(meta.getIdColumnMeta().convertToStorage2(ApiPostDataPointsImpl.stringToObject(meta.getIdColumnMeta().getStorageType(), ""+ts.getTime())));
			//partition.removeColumn(""+ts.getTime());
			s.put(meta.getColumnFamily(), partition);
			if (count%flushSize == 0) {
				log.info("flushing, count is "+count);
				s.flush();
				mgr.flush();
			}
			
			//fast way:  figure out if this row (partition) will have all of it's columns deleted in this operation, if so, just delete the entire row
			count++;
			ts = nextResult();
		}
		s.flush();
		mgr.flush();
		if (log.isDebugEnabled()) {
			long endTime = System.currentTimeMillis();
			if (count == 0) log.debug("no items to delete in the range "+start+" to "+end);
			else log.debug("deleted "+count+" items, total time "+(endTime-startTime)+"ms, "+((endTime-startTime)/count)+"ms per item deleted");
		}
		
	}

	private void loadPartitions(DboTableMeta meta2) {
		DboTableMeta tableMeta = mgr.find(DboTableMeta.class, "partitions");
		NoSqlSession session = mgr.getSession();
		byte[] rowKey = StandardConverters.convertToBytes(meta2.getColumnFamily());
		Cursor<Column> results = session.columnSlice(tableMeta, rowKey, null, null, buffersize, BigInteger.class);
		
		while(results.next()) {
			Column col = results.getCurrent();
			BigInteger time = StandardConverters.convertFromBytes(BigInteger.class, col.getName());
			existingPartitions.add(time.longValue());
		}
		
		Collections.sort(existingPartitions);
	}
	
	private TypedRow getPartitionForTime(long time) {
		long previousKey = existingPartitions.get(0);
		for (int i = 1; i < existingPartitions.size(); i++) {
			long nextKey = existingPartitions.get(i);
			if (previousKey <= time && nextKey > time)
				return getPartition(previousKey);
			previousKey=nextKey;
		}
		return null;
	}
	
	public TypedRow getPartition(long partitionId) {
		NoSqlTypedSession typedSession = mgr.getTypedSession();
		TypedRow row = typedSession.createTypedRow(meta.getColumnFamily());
		BigInteger rowKey = new BigInteger(""+partitionId);
		row.setRowKey(rowKey);
		return row;
	}

	protected int partition(long partitionSize) {
		if(existingPartitions.size() > 0 && start == null)
			return 0;
		
		for(int i = 0; i < existingPartitions.size();i++) {
			long partId = existingPartitions.get(i);
			if(start < partId+partitionSize)
				return i;
		}
<commit_msg>add deleterange api and delete table<commit_after>package controllers.impl;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import play.Play;
import play.PlayPlugin;

import com.alvazan.orm.api.base.NoSqlEntityManager;
import com.alvazan.orm.api.z3api.NoSqlTypedSession;
import com.alvazan.orm.api.z5api.NoSqlSession;
import com.alvazan.orm.api.z8spi.action.Column;
import com.alvazan.orm.api.z8spi.conv.StandardConverters;
import com.alvazan.orm.api.z8spi.conv.StorageTypeEnum;
import com.alvazan.orm.api.z8spi.iter.AbstractCursor;
import com.alvazan.orm.api.z8spi.iter.Cursor;
import com.alvazan.orm.api.z8spi.meta.DboColumnIdMeta;
import com.alvazan.orm.api.z8spi.meta.DboColumnMeta;
import com.alvazan.orm.api.z8spi.meta.DboTableMeta;
import com.alvazan.orm.api.z8spi.meta.TypedColumn;
import com.alvazan.orm.api.z8spi.meta.TypedRow;
import com.alvazan.play.NoSql;

import controllers.api.ApiPostDataPointsImpl;
import controllers.modules2.RawTimeSeriesProcessor;
import controllers.modules2.framework.ReadResult;
import controllers.modules2.framework.TSRelational;
import controllers.modules2.framework.VisitorInfo;
import controllers.modules2.framework.procs.RowMeta;

public class RawTimeSeriesImpl  extends PlayPlugin {

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

	public static final TSRelational END_OF_DATA_TOKEN = new TSRelational();
	
	protected DboTableMeta meta;
	protected DboColumnIdMeta idcolMeta;

	protected byte[] endBytes;
	protected byte[] startBytes;
	protected Long currentPartitionId;
	protected Long start;
	protected Long end;
	protected AbstractCursor<Column> cursor;
	protected Long partitionSize;
	private List<DboColumnMeta> colMeta = new ArrayList<DboColumnMeta>();
	protected int currentIndex;
	protected List<Long> existingPartitions = new ArrayList<Long>();
	protected NoSqlEntityManager mgr;
	protected BlockingQueue<TSRelational> itemsQueue;
	protected int buffersize = 2000;

	private String timeColumn;
	private String valueColumn;
	
	private final static Charset charset = StandardCharsets.UTF_8;

	public void init(DboTableMeta meta, Long start, Long end, RowMeta rowMeta, NoSqlEntityManager mgr) {
		this.mgr = mgr;
		this.meta = meta;
		idcolMeta = meta.getIdColumnMeta();

		
		this.timeColumn = rowMeta.getTimeColumn();
		this.valueColumn = rowMeta.getValueColumn();
		
		this.start = start;
		this.end = end;
		loadPartitions(meta);
		partitionSize = meta.getTimeSeriesPartionSize();
		currentIndex = partition(partitionSize);

		if(start != null)
			this.startBytes = meta.getIdColumnMeta().convertToStorage2(new BigInteger(start+""));
		if(end != null)
			this.endBytes = meta.getIdColumnMeta().convertToStorage2(new BigInteger(end+""));

		for (DboColumnMeta thismeta:meta.getAllColumns())
			colMeta.add(thismeta);
		if (log.isInfoEnabled())
			log.info("Setting up for reading partitions, partId="+currentIndex+" partitions="+existingPartitions+" start="+start);
		
		//jsc fire off read thread
		String buffersizeString = Play.configuration.getProperty("databus.preread.buffer.size", "2000");
		buffersize = Integer.parseInt(buffersizeString);
		itemsQueue = new LinkedBlockingQueue<TSRelational>(buffersize);
		new Thread(new AgressivePreReader()).start();
	}
	
	public void deleteRange() {
		int flushSize = 1000;
		NoSqlTypedSession s = mgr.getTypedSession();
		
		TSRelational ts = nextResult();
		long startTime = System.currentTimeMillis();
		long count = 0;
		while (ts != null) {
			//slow way:  get row, delete column.
			TypedRow partition = getPartitionForTime(ts.getTime());

			partition.removeColumn(meta.getIdColumnMeta().convertToStorage2(ApiPostDataPointsImpl.stringToObject(meta.getIdColumnMeta().getStorageType(), ""+ts.getTime())));
			//partition.removeColumn(""+ts.getTime());
			s.put(meta.getColumnFamily(), partition);
			if (count%flushSize == 0) {
				log.info("flushing, count is "+count);
				s.flush();
				mgr.flush();
			}
			
			//fast way:  figure out if this row (partition) will have all of it's columns deleted in this operation, if so, just delete the entire row
			count++;
			ts = nextResult();
		}
		s.flush();
		mgr.flush();
		if (log.isDebugEnabled()) {
			long endTime = System.currentTimeMillis();
			if (count == 0) log.debug("no items to delete in the range "+start+" to "+end);
			else log.debug("deleted "+count+" items, total time "+(endTime-startTime)+"ms, "+((endTime-startTime)/count)+"ms per item deleted");
		}
		
	}
	
	public void deleteAll() {
		for (long partitionid:existingPartitions) {
			TypedRow partition = getPartition(partitionid);
			mgr.getTypedSession().remove(meta.getColumnFamily(), partition);
		}
		mgr.flush();
		
	}

	private void loadPartitions(DboTableMeta meta2) {
		DboTableMeta tableMeta = mgr.find(DboTableMeta.class, "partitions");
		NoSqlSession session = mgr.getSession();
		byte[] rowKey = StandardConverters.convertToBytes(meta2.getColumnFamily());
		Cursor<Column> results = session.columnSlice(tableMeta, rowKey, null, null, buffersize, BigInteger.class);
		
		while(results.next()) {
			Column col = results.getCurrent();
			BigInteger time = StandardConverters.convertFromBytes(BigInteger.class, col.getName());
			existingPartitions.add(time.longValue());
		}
		
		Collections.sort(existingPartitions);
	}
	
	private TypedRow getPartitionForTime(long time) {
		long previousKey = existingPartitions.get(0);
		for (int i = 1; i < existingPartitions.size(); i++) {
			long nextKey = existingPartitions.get(i);
			if (previousKey <= time && nextKey > time)
				return getPartition(previousKey);
			previousKey=nextKey;
		}
		//special case, if time is in the very last partition we need to figure it out:
		if (previousKey <= time && (previousKey + meta.getTimeSeriesPartionSize()) >time)
			return getPartition(previousKey);
		return null;
	}
	
	public TypedRow getPartition(long partitionId) {
		NoSqlTypedSession typedSession = mgr.getTypedSession();
		TypedRow row = typedSession.createTypedRow(meta.getColumnFamily());
		BigInteger rowKey = new BigInteger(""+partitionId);
		row.setRowKey(rowKey);
		return row;
	}

	protected int partition(long partitionSize) {
		if(existingPartitions.size() > 0 && start == null)
			return 0;
		
		for(int i = 0; i < existingPartitions.size();i++) {
			long partId = existingPartitions.get(i);
			if(start < partId+partitionSize)
				return i;
		}

{'fraction_non_alphanumeric': 0.11757603199758508, 'fraction_numerical': 0.006791940230925968, 'mean_word_length': 5.218676677616143, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 24, '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': '2007466', 'n_tokens_mistral': 4527, 'n_tokens_neox': 4181, 'n_words': 907}