<commit_before>package com.splicemachine.mrio.api;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.log4j.Logger;
import com.splicemachine.utils.SpliceLogUtils;
/*
 * 
 * Split Scanner for multiple region scanners
 * 
 */
public class SplitRegionScanner implements SpliceRegionScanner {
    protected static final Logger LOG = Logger.getLogger(SplitRegionScanner.class);
	protected List<RegionScanner> regionScanners = new ArrayList<RegionScanner>(2);	
	protected RegionScanner currentScanner;
	protected FileSystem fileSystem;
	protected HRegion region;
	
	public SplitRegionScanner(Scan scan, HTable table) throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "init");
		ClientSideRegionScanner clientSideRegionScanner = 
				new ClientSideRegionScanner(table.getConfiguration(),FSUtils.getCurrentFileSystem(table.getConfiguration()), FSUtils.getRootDir(table.getConfiguration()),
						table.getTableDescriptor(),table.getRegionLocation(scan.getStartRow()).getRegionInfo(),
						scan,null);		
		region = clientSideRegionScanner.region;
		registerRegionScanner(clientSideRegionScanner);
	}
	
	public void registerRegionScanner(RegionScanner regionScanner) {
		if (currentScanner != null)
			regionScanners.add(regionScanner);
		else
			currentScanner = regionScanner;
	}
	
	public boolean next(List<KeyValue> results) throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "next with results=%s",results);
		boolean next = currentScanner.nextRaw(results,null);
		if (!next && !regionScanners.isEmpty()) {
			currentScanner = regionScanners.remove(0);
			return next(results);
		}
		return next;
	}


	@Override
	public void close() throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "close");
		currentScanner.close();
		for (RegionScanner rs: regionScanners) {
			rs.close();
		}
	}

	@Override
	public HRegionInfo getRegionInfo() {
		return currentScanner.getRegionInfo();
	}

	@Override
	public boolean isFilterDone() {
		return currentScanner.isFilterDone();
	}

	@Override
	public boolean reseek(byte[] row) throws IOException {
		throw new RuntimeException("Reseek not supported");
	}

	@Override
	public long getMvccReadPoint() {
		return currentScanner.getMvccReadPoint();
	}
	@Override
	public HRegion getRegion() {
		return region;
	}

	@Override
	public boolean next(List<KeyValue> results, String arg1) throws IOException {
		return next(results);
	}

	@Override
	public boolean next(List<KeyValue> results, int arg1) throws IOException {
		return next(results);
	}

	@Override
	public boolean next(List<KeyValue> results, int arg1, String arg2)
			throws IOException {
		return next(results);
	}

	@Override
	public boolean nextRaw(List<KeyValue> results, String arg1) throws IOException {
		return next(results);
	}

	@Override
	public boolean nextRaw(List<KeyValue> results, int arg1, String arg2)
			throws IOException {
		return next(results);
	}
	
}<commit_msg>Add Code<commit_after>package com.splicemachine.mrio.api;

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

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.log4j.Logger;

import com.splicemachine.utils.SpliceLogUtils;
/*
 * 
 * Split Scanner for multiple region scanners
 * 
 */
public class SplitRegionScanner implements SpliceRegionScanner {
    protected static final Logger LOG = Logger.getLogger(SplitRegionScanner.class);
	protected List<RegionScanner> regionScanners = new ArrayList<RegionScanner>(2);	
	protected RegionScanner currentScanner;
	protected FileSystem fileSystem;
	protected HRegion region;
	protected int scannerPosition = 1;
	protected int scannerCount = 0;
	protected Scan scan;
	protected HTable htable;
	protected List<KeyValue> holderResults = new ArrayList<KeyValue>();
	protected boolean holderReturn;
	
	public SplitRegionScanner(Scan scan, HTable table, List<HRegionLocation> locations) throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "init");
		this.scan = scan;
		boolean hasAdditionalScanners = true;
		while (hasAdditionalScanners) {
			try {
				for (int i = 0; i< locations.size(); i++) {
					Scan newScan = new Scan(scan);
				    byte[] startRow = scan.getStartRow();
				    byte[] stopRow = scan.getStopRow();
				    byte[] regionStartKey = locations.get(i).getRegionInfo().getStartKey();
				    byte[] regionStopKey = locations.get(i).getRegionInfo().getEndKey();
				    // determine if the given start an stop key fall into the region
				    if ((startRow.length == 0 || regionStopKey.length == 0 ||
				          Bytes.compareTo(startRow, regionStopKey) < 0) &&
				          (stopRow.length == 0 ||
				           Bytes.compareTo(stopRow, regionStartKey) > 0)) { 
				    	  byte[] splitStart = startRow.length == 0 ||
				    			  Bytes.compareTo(regionStartKey, startRow) >= 0 ?
				    					  regionStartKey : startRow;
				    	  byte[] splitStop = (stopRow.length == 0 ||
				    			  Bytes.compareTo(regionStopKey, stopRow) <= 0) &&
				    			  regionStopKey.length > 0 ? regionStopKey : stopRow;
				    	  newScan.setStartRow(splitStart);
				    	  newScan.setStopRow(splitStop);
				    	  SpliceLogUtils.trace(LOG, "adding Split Region Scanner for startKey=%s, endKey=%s",splitStart,splitStop);
				  		  ClientSideRegionScanner clientSideRegionScanner = 
				  				  new ClientSideRegionScanner(table.getConfiguration(),FSUtils.getCurrentFileSystem(table.getConfiguration()), FSUtils.getRootDir(table.getConfiguration()),
										table.getTableDescriptor(),table.getRegionLocation(newScan.getStartRow()).getRegionInfo(),
										newScan,null);
				  				region = clientSideRegionScanner.region;
				  				registerRegionScanner(clientSideRegionScanner);			    			  
				      }
				 }
				 hasAdditionalScanners = false;
			} catch (SplitScannerDNRIOException ioe) {
				if (LOG.isDebugEnabled())
					SpliceLogUtils.debug(LOG, "exception logged creating split region scanner %s",StringUtils.stringifyException(ioe));
				hasAdditionalScanners = true;
				try {Thread.sleep(200);} catch (Exception e) {}; // Pause for 200 ms...
				htable.clearRegionCache();
				locations = htable.getRegionsInRange(scan.getStartRow(), scan.getStopRow());
				close();
			}
		}
			
	}
	
	public void registerRegionScanner(RegionScanner regionScanner) {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "registerRegionScanner %s",regionScanner);
		if (currentScanner == null)
			currentScanner = regionScanner;
		regionScanners.add(regionScanner);
	}
	
	@Override
	public boolean next(List<KeyValue> results) throws IOException {
		if (holderReturn) {
			holderReturn = false;
			results.addAll(holderResults);
			return true;
		}
		boolean next = currentScanner.nextRaw(results,null);
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "next with results=%s and row count {%d}",results, scannerCount);
		scannerCount++;
		if (!next && scannerPosition<regionScanners.size()) {
			if (LOG.isTraceEnabled())
				SpliceLogUtils.trace(LOG, "scanner [%d] exhausted after {%d} records",scannerPosition,scannerCount);
			currentScanner = regionScanners.get(scannerPosition);
			scannerPosition++;
			scannerCount = 0;
			holderResults.clear();
			holderReturn = next(holderResults);
			return holderReturn;
		}

		return next;
	}

	@Override
	public boolean next(List<KeyValue> result, int limit) throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "next with results=%s and limit=%d",result,limit);
		return next(result);
	}

	@Override
	public void close() throws IOException {
		if (LOG.isTraceEnabled())
			SpliceLogUtils.trace(LOG, "close");
		if (currentScanner != null)
			currentScanner.close();
		for (RegionScanner rs: regionScanners) {
			rs.close();
		}
	}

	@Override
	public HRegionInfo getRegionInfo() {
		return currentScanner.getRegionInfo();
	}

	@Override
	public boolean isFilterDone() {
		return currentScanner.isFilterDone();
	}

	@Override
	public boolean reseek(byte[] row) throws IOException {
		throw new RuntimeException("Reseek not supported");
	}


	@Override
	public long getMvccReadPoint() {
		return currentScanner.getMvccReadPoint();
	}

	@Override
	public HRegion getRegion() {
		return region;
	}

	@Override
	public boolean nextRaw(List<KeyValue> result, String arg1) throws IOException {
		// TODO Auto-generated method stub
		return next(result);
	}

	@Override
	public boolean nextRaw(List<KeyValue> result, int arg1, String arg2)
			throws IOException {
		// TODO Auto-generated method stub
		return next(result);
	}

	@Override
	public boolean next(List<KeyValue> result, String arg1) throws IOException {
		// TODO Auto-generated method stub
		return next(result);
	}

	@Override
	public boolean next(List<KeyValue> result, int arg1, String arg2)
			throws IOException {
		// TODO Auto-generated method stub
		return next(result);
	}
}
{'fraction_non_alphanumeric': 0.10793943114385103, 'fraction_numerical': 0.003887865766318805, 'mean_word_length': 4.476190476190476, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 25, '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': '2774639', 'n_tokens_mistral': 3335, 'n_tokens_neox': 2931, 'n_words': 660}