<commit_before>package com.splicemachine.derby.impl.sql.execute.operations.scanner;

import com.google.common.collect.Lists;
import com.splicemachine.SpliceKryoRegistry;
import com.splicemachine.constants.SIConstants;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.derby.iapi.sql.execute.SpliceRuntimeContext;
import com.splicemachine.derby.impl.store.ExecRowAccumulator;
import com.splicemachine.derby.impl.store.access.hbase.HBaseRowLocation;
import com.splicemachine.derby.utils.StandardIterator;
import com.splicemachine.derby.utils.marshall.dvd.TypeProvider;
import com.splicemachine.derby.utils.marshall.dvd.VersionedSerializers;
import com.splicemachine.encoding.MultiFieldDecoder;
import com.splicemachine.si.api.HTransactorFactory;
import com.splicemachine.si.api.SIFilter;
import com.splicemachine.si.data.hbase.HRowAccumulator;
import com.splicemachine.si.impl.*;
import com.splicemachine.stats.Counter;
import com.splicemachine.stats.MetricFactory;
import com.splicemachine.stats.TimeView;
import com.splicemachine.stats.Timer;
import com.splicemachine.storage.EntryAccumulator;
import com.splicemachine.storage.EntryDecoder;
import com.splicemachine.storage.EntryPredicateFilter;
import com.splicemachine.storage.Indexed;
import com.splicemachine.utils.ByteSlice;
import com.splicemachine.utils.Provider;
import com.splicemachine.utils.Providers;
import org.apache.derby.iapi.error.StandardException;
import org.apache.derby.iapi.services.io.FormatableBitSet;
import org.apache.derby.iapi.services.io.StoredFormatIds;
import org.apache.derby.iapi.sql.execute.ExecRow;
import org.apache.derby.iapi.types.RowLocation;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.regionserver.RegionScanner;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;

/**
 * TableScanner which applies SI to generate a row
 * @author <NAME>
 * Date: 4/4/14
 */
public class SITableScanner implements StandardIterator<ExecRow>{
		private final Timer timer;
		private final Counter filterCounter;

		private RegionScanner regionScanner;
		private final Scan scan;
		private final ExecRow template;
		private final String tableVersion;
		private final int[] rowDecodingMap;


		private SIFilter siFilter;
		private EntryPredicateFilter predicateFilter;
		private List<KeyValue> keyValues;

		private RowLocation currentRowLocation;
		private final boolean[] keyColumnSortOrder;
		private String indexName;
		private ByteSlice slice = new ByteSlice();

		private boolean isKeyed = true;
		private KeyIndex primaryKeyIndex;
		private MultiFieldDecoder keyDecoder;
		private final Provider<MultiFieldDecoder> keyDecoderProvider;
		private ExecRowAccumulator keyAccumulator;
		private int[] keyDecodingMap;
		private FormatableBitSet accessedKeys;

		private final SIFilterFactory filterFactory;


		SITableScanner(RegionScanner scanner,
													ExecRow template,
													MetricFactory metricFactory,
													Scan scan,
													int[] rowDecodingMap,
													String transactionID,
													int[] allPkColumns,
													boolean[] keyColumnSortOrder,
													int[] keyColumnTypes,
													int[] keyDecodingMap,
													FormatableBitSet accessedPks,
													String indexName,
													String tableVersion) {
			this(scanner,
							template,
							metricFactory,
							scan,
							rowDecodingMap,
							transactionID,
							allPkColumns,
							keyColumnSortOrder,
							keyColumnTypes,
							keyDecodingMap,
							accessedPks,
							indexName,
							tableVersion,null);
		}

		SITableScanner(RegionScanner scanner,
													final ExecRow template,
													MetricFactory metricFactory,
													Scan scan,
													final int[] rowDecodingMap,
													final String transactionID,
													int[] keyColumnEncodingOrder,
													boolean[] keyColumnSortOrder,
													int[] keyColumnTypes,
													int[] keyDecodingMap,
													FormatableBitSet accessedPks,
													String indexName,
													final String tableVersion,
													SIFilterFactory filterFactory) {
				this.scan = scan;
				this.template = template;
				this.rowDecodingMap = rowDecodingMap;
				this.keyColumnSortOrder = keyColumnSortOrder;
				this.indexName = indexName;
				this.timer = metricFactory.newTimer();
				this.filterCounter = metricFactory.newCounter();
				this.regionScanner = scanner;
				this.keyDecodingMap = keyDecodingMap;
				this.accessedKeys = accessedPks;
				this.keyDecoderProvider = getKeyDecoder(accessedPks, keyColumnEncodingOrder,
								keyColumnTypes, VersionedSerializers.typesForVersion(tableVersion));
				this.tableVersion = tableVersion;
				if(filterFactory==null){
						this.filterFactory = new SIFilterFactory() {
								@Override
								public SIFilter newFilter(EntryPredicateFilter predicateFilter,
																					EntryDecoder rowEntryDecoder,
																					EntryAccumulator accumulator,
																					boolean isCountStar) throws IOException {
										TransactionId transactionId= new TransactionId(transactionID);
										IFilterState iFilterState = HTransactorFactory.getTransactionReadController().newFilterState(null, transactionId);

										HRowAccumulator hRowAccumulator = new HRowAccumulator(predicateFilter, getRowEntryDecoder(), accumulator, isCountStar);
										//noinspection unchecked
										return new FilterStatePacked((FilterState)iFilterState, hRowAccumulator){
												@Override
												public Filter.ReturnCode doAccumulate(KeyValue dataKeyValue) throws IOException {
														if (!accumulator.isFinished() && accumulator.isOfInterest(dataKeyValue)) {
																if (!accumulator.accumulate(dataKeyValue)) {
																		return Filter.ReturnCode.NEXT_ROW;
																}
																return Filter.ReturnCode.INCLUDE;
														}else return Filter.ReturnCode.INCLUDE;
												}
										};
								}
						};
				}
				else
						this.filterFactory = filterFactory;
		}

		@Override
		public void open() throws StandardException, IOException {

		}

		@Override
		public ExecRow next(SpliceRuntimeContext spliceRuntimeContext) throws StandardException, IOException {
				timer.startTiming();

				SIFilter filter = getSIFilter();
				if(keyValues==null)
						keyValues = Lists.newArrayListWithExpectedSize(2);
				boolean hasRow;
				do{
						keyValues.clear();
						template.resetRowArray();
						hasRow = regionScanner.next(keyValues);
						if(keyValues.size()<=0){
								timer.stopTiming();
								currentRowLocation = null;
								return null;
						}else{
								if(template.nColumns()>0){
										if(!filterRowKey(keyValues.get(0))||!filterRow(filter)){
												//filter the row first, then filter the row key
												filterCounter.increment();
												continue;
										}
								}else if(!filterRow(filter)){
										//still need to filter rows to deal with transactional issues
										filterCounter.increment();
										continue;
								}
								setRowLocation(keyValues.get(0));
								timer.tick(1);
								return template;
						}
				}while(hasRow);

				timer.stopTiming();
				currentRowLocation = null;<commit_msg>No Reset of Row Array<commit_after>package com.splicemachine.derby.impl.sql.execute.operations.scanner;

import com.google.common.collect.Lists;
import com.splicemachine.SpliceKryoRegistry;
import com.splicemachine.constants.SIConstants;
import com.splicemachine.constants.SpliceConstants;
import com.splicemachine.derby.iapi.sql.execute.SpliceRuntimeContext;
import com.splicemachine.derby.impl.store.ExecRowAccumulator;
import com.splicemachine.derby.impl.store.access.hbase.HBaseRowLocation;
import com.splicemachine.derby.utils.StandardIterator;
import com.splicemachine.derby.utils.marshall.dvd.TypeProvider;
import com.splicemachine.derby.utils.marshall.dvd.VersionedSerializers;
import com.splicemachine.encoding.MultiFieldDecoder;
import com.splicemachine.si.api.HTransactorFactory;
import com.splicemachine.si.api.SIFilter;
import com.splicemachine.si.data.hbase.HRowAccumulator;
import com.splicemachine.si.impl.*;
import com.splicemachine.stats.Counter;
import com.splicemachine.stats.MetricFactory;
import com.splicemachine.stats.TimeView;
import com.splicemachine.stats.Timer;
import com.splicemachine.storage.EntryAccumulator;
import com.splicemachine.storage.EntryDecoder;
import com.splicemachine.storage.EntryPredicateFilter;
import com.splicemachine.storage.Indexed;
import com.splicemachine.utils.ByteSlice;
import com.splicemachine.utils.Provider;
import com.splicemachine.utils.Providers;
import org.apache.derby.iapi.error.StandardException;
import org.apache.derby.iapi.services.io.FormatableBitSet;
import org.apache.derby.iapi.services.io.StoredFormatIds;
import org.apache.derby.iapi.sql.execute.ExecRow;
import org.apache.derby.iapi.types.RowLocation;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.regionserver.RegionScanner;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;

/**
 * TableScanner which applies SI to generate a row
 * @author <NAME>
 * Date: 4/4/14
 */
public class SITableScanner implements StandardIterator<ExecRow>{
		private final Timer timer;
		private final Counter filterCounter;

		private RegionScanner regionScanner;
		private final Scan scan;
		private final ExecRow template;
		private final String tableVersion;
		private final int[] rowDecodingMap;


		private SIFilter siFilter;
		private EntryPredicateFilter predicateFilter;
		private List<KeyValue> keyValues;

		private RowLocation currentRowLocation;
		private final boolean[] keyColumnSortOrder;
		private String indexName;
		private ByteSlice slice = new ByteSlice();

		private boolean isKeyed = true;
		private KeyIndex primaryKeyIndex;
		private MultiFieldDecoder keyDecoder;
		private final Provider<MultiFieldDecoder> keyDecoderProvider;
		private ExecRowAccumulator keyAccumulator;
		private int[] keyDecodingMap;
		private FormatableBitSet accessedKeys;

		private final SIFilterFactory filterFactory;


		SITableScanner(RegionScanner scanner,
													ExecRow template,
													MetricFactory metricFactory,
													Scan scan,
													int[] rowDecodingMap,
													String transactionID,
													int[] allPkColumns,
													boolean[] keyColumnSortOrder,
													int[] keyColumnTypes,
													int[] keyDecodingMap,
													FormatableBitSet accessedPks,
													String indexName,
													String tableVersion) {
			this(scanner,
							template,
							metricFactory,
							scan,
							rowDecodingMap,
							transactionID,
							allPkColumns,
							keyColumnSortOrder,
							keyColumnTypes,
							keyDecodingMap,
							accessedPks,
							indexName,
							tableVersion,null);
		}

		SITableScanner(RegionScanner scanner,
													final ExecRow template,
													MetricFactory metricFactory,
													Scan scan,
													final int[] rowDecodingMap,
													final String transactionID,
													int[] keyColumnEncodingOrder,
													boolean[] keyColumnSortOrder,
													int[] keyColumnTypes,
													int[] keyDecodingMap,
													FormatableBitSet accessedPks,
													String indexName,
													final String tableVersion,
													SIFilterFactory filterFactory) {
				this.scan = scan;
				this.template = template;
				this.rowDecodingMap = rowDecodingMap;
				this.keyColumnSortOrder = keyColumnSortOrder;
				this.indexName = indexName;
				this.timer = metricFactory.newTimer();
				this.filterCounter = metricFactory.newCounter();
				this.regionScanner = scanner;
				this.keyDecodingMap = keyDecodingMap;
				this.accessedKeys = accessedPks;
				this.keyDecoderProvider = getKeyDecoder(accessedPks, keyColumnEncodingOrder,
								keyColumnTypes, VersionedSerializers.typesForVersion(tableVersion));
				this.tableVersion = tableVersion;
				if(filterFactory==null){
						this.filterFactory = new SIFilterFactory() {
								@Override
								public SIFilter newFilter(EntryPredicateFilter predicateFilter,
																					EntryDecoder rowEntryDecoder,
																					EntryAccumulator accumulator,
																					boolean isCountStar) throws IOException {
										TransactionId transactionId= new TransactionId(transactionID);
										IFilterState iFilterState = HTransactorFactory.getTransactionReadController().newFilterState(null, transactionId);

										HRowAccumulator hRowAccumulator = new HRowAccumulator(predicateFilter, getRowEntryDecoder(), accumulator, isCountStar);
										//noinspection unchecked
										return new FilterStatePacked((FilterState)iFilterState, hRowAccumulator){
												@Override
												public Filter.ReturnCode doAccumulate(KeyValue dataKeyValue) throws IOException {
														if (!accumulator.isFinished() && accumulator.isOfInterest(dataKeyValue)) {
																if (!accumulator.accumulate(dataKeyValue)) {
																		return Filter.ReturnCode.NEXT_ROW;
																}
																return Filter.ReturnCode.INCLUDE;
														}else return Filter.ReturnCode.INCLUDE;
												}
										};
								}
						};
				}
				else
						this.filterFactory = filterFactory;
		}

		@Override
		public void open() throws StandardException, IOException {

		}

		@Override
		public ExecRow next(SpliceRuntimeContext spliceRuntimeContext) throws StandardException, IOException {
				timer.startTiming();

				SIFilter filter = getSIFilter();
				if(keyValues==null)
						keyValues = Lists.newArrayListWithExpectedSize(2);
				boolean hasRow;
				do{
						keyValues.clear();
//						template.resetRowArray();
						hasRow = regionScanner.next(keyValues);
						if(keyValues.size()<=0){
								timer.stopTiming();
								currentRowLocation = null;
								return null;
						}else{
								if(template.nColumns()>0){
										if(!filterRowKey(keyValues.get(0))||!filterRow(filter)){
												//filter the row first, then filter the row key
												filterCounter.increment();
												continue;
										}
								}else if(!filterRow(filter)){
										//still need to filter rows to deal with transactional issues
										filterCounter.increment();
										continue;
								}
								setRowLocation(keyValues.get(0));
								timer.tick(1);
								return template;
						}
				}while(hasRow);

				timer.stopTiming();
				currentRowLocation = null;
{'fraction_non_alphanumeric': 0.08083832335329341, 'fraction_numerical': 0.0013609145345672292, 'mean_word_length': 3.1812233285917495, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '4639136', 'n_tokens_mistral': 5846, 'n_tokens_neox': 4127, 'n_words': 807}