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

import com.splicemachine.constants.HBaseConstants;
import com.splicemachine.derby.hbase.SpliceOperationCoprocessor;
import com.splicemachine.derby.iapi.sql.execute.SpliceNoPutResultSet;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperation;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperationContext;
import com.splicemachine.derby.iapi.storage.RowProvider;
import com.splicemachine.derby.impl.storage.ClientScanProvider;
import com.splicemachine.derby.impl.storage.MergeSortRegionAwareRowProvider;
import com.splicemachine.derby.impl.store.access.SpliceAccessManager;
import com.splicemachine.derby.impl.store.access.hbase.HBaseRowLocation;
import com.splicemachine.derby.utils.*;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.derby.iapi.error.StandardException;
import org.apache.derby.iapi.services.io.FormatableArrayHolder;
import org.apache.derby.iapi.services.io.FormatableIntHolder;
import org.apache.derby.iapi.services.loader.GeneratedMethod;
import org.apache.derby.iapi.sql.Activation;
import org.apache.derby.iapi.sql.execute.ExecRow;
import org.apache.derby.iapi.sql.execute.NoPutResultSet;
import org.apache.derby.iapi.store.access.Qualifier;
import org.apache.derby.iapi.types.DataValueDescriptor;
import org.apache.derby.iapi.types.SQLInteger;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class MergeSortJoinOperation extends JoinOperation {
	private static Logger LOG = Logger.getLogger(MergeSortJoinOperation.class);
	protected String emptyRowFunMethodName;
	protected boolean wasRightOuterJoin;
	protected Qualifier[][] qualifierProbe;
	protected int leftHashKeyItem;
	protected int[] leftHashKeys;
	protected int rightHashKeyItem;
	protected int[] rightHashKeys;
	protected ExecRow rightTemplate;
	protected static List<NodeType> nodeTypes; 
	protected Scan reduceScan;
	protected enum JoinSide {RIGHT,LEFT};
	protected JoinSide joinSide;
	protected RowProvider clientProvider;
	protected MergeSortRegionAwareRowProvider serverProvider;
	protected SQLInteger rowType;
	protected byte[] priorHash;
	protected List<ExecRow> rights;
	protected byte[] rightHash;
	protected Iterator<ExecRow> rightIterator;
	
	static {
		nodeTypes = new ArrayList<NodeType>();
		nodeTypes.add(NodeType.REDUCE);
		nodeTypes.add(NodeType.SCAN);
		nodeTypes.add(NodeType.SINK);
	}
	
	public MergeSortJoinOperation() {
		super();
	}
	
	public MergeSortJoinOperation(NoPutResultSet leftResultSet,
			   int leftNumCols,
			   NoPutResultSet rightResultSet,
			   int rightNumCols,
			   int leftHashKeyItem,
			   int rightHashKeyItem,
			   Activation activation,
			   GeneratedMethod restriction,
			   int resultSetNumber,
			   boolean oneRowRightSide,
			   boolean notExistsRightSide,
			   double optimizerEstimatedRowCount,
			   double optimizerEstimatedCost,
			   String userSuppliedOptimizerOverrides) throws StandardException {		
				super(leftResultSet, leftNumCols, rightResultSet, rightNumCols,
						activation, restriction, resultSetNumber,oneRowRightSide, notExistsRightSide,
						optimizerEstimatedRowCount, optimizerEstimatedCost,userSuppliedOptimizerOverrides);
				SpliceLogUtils.trace(LOG, "instantiate");
				this.leftHashKeyItem = leftHashKeyItem;
				this.rightHashKeyItem = rightHashKeyItem;
				this.joinSide = JoinSide.LEFT;
                init(SpliceOperationContext.newContext(activation));
	}
	
	@Override
	public void readExternal(ObjectInput in) throws IOException,ClassNotFoundException {
		SpliceLogUtils.trace(LOG, "readExternal");
		super.readExternal(in);
		leftHashKeyItem = in.readInt();
		rightHashKeyItem = in.readInt();
		joinSide = JoinSide.values()[in.readInt()];
	}

	@Override
	public void writeExternal(ObjectOutput out) throws IOException {
		SpliceLogUtils.trace(LOG, "writeExternal");
		super.writeExternal(out);
		out.writeInt(leftHashKeyItem);
		out.writeInt(rightHashKeyItem);
		out.writeInt(joinSide.ordinal());
	}

	@Override
	public ExecRow getNextRowCore() throws StandardException {
		SpliceLogUtils.trace(LOG, "getNextRowCore");
		if (rightIterator!= null && rightIterator.hasNext()) {
			currentRow = JoinUtils.getMergedRow(leftRow, rightIterator.next(), wasRightOuterJoin, this.leftNumCols, this.rightNumCols, mergedRow);
			this.setCurrentRow(currentRow);
			currentRowLocation = new HBaseRowLocation(SpliceUtils.getUniqueKey());
			SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
			return currentRow;
		}
		if (!serverProvider.hasNext()) {
			SpliceLogUtils.trace(LOG, "serverProvider exhausted");
			return null;
		}
		JoinSideExecRow joinRow = serverProvider.nextJoinRow();
		SpliceLogUtils.trace(LOG, "joinRow=%s",joinRow);
		if (joinRow == null) {
			SpliceLogUtils.trace(LOG, "serverProvider returned null rows");
			this.setCurrentRow(null);
			return null;
		}
		
		if (joinRow.getJoinSide().ordinal() == JoinSide.RIGHT.ordinal()) { // Right Side
			rightHash = joinRow.getHash();
			if (joinRow.sameHash(priorHash)) {
				SpliceLogUtils.trace(LOG, "adding additional right=%s",joinRow);
				rights.add(joinRow.getRow());
			} else {
				resetRightSide();
				SpliceLogUtils.trace(LOG, "adding initial right=%s",joinRow);
				rights.add(joinRow.getRow());
				priorHash = joinRow.getHash();
			}
			return getNextRowCore();
		} else { // Left Side
			leftRow = joinRow.getRow();
			if (joinRow.sameHash(priorHash)) {
				if (joinRow.sameHash(rightHash)) {
					SpliceLogUtils.trace(LOG, "initializing iterator with rights for left=%s",joinRow);
					rightIterator = rights.iterator();
					currentRow = JoinUtils.getMergedRow(leftRow, rightIterator.next(), wasRightOuterJoin, this.leftNumCols, this.rightNumCols, mergedRow);
					this.setCurrentRow(currentRow);
					currentRowLocation = new HBaseRowLocation(SpliceUtils.getUniqueKey());					
					SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
					return currentRow;
				} else {
					SpliceLogUtils.trace(LOG, "right hash miss left=%s",joinRow);
					resetRightSide();	
					priorHash = joinRow.getHash();
					return getNextRowCore();
				}
			} else {
				SpliceLogUtils.trace(LOG, "simple left with no right=%s",joinRow);
				resetRightSide();
				priorHash = joinRow.getHash();
				SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
				return getNextRowCore();			
			}			
		}
	}

	@Override
	public RowProvider getReduceRowProvider(SpliceOperation top,ExecRow template){
        if(clientProvider==null){
            SpliceUtils.setInstructions(reduceScan,activation,top);
            clientProvider = new ClientScanProvider(SpliceOperationCoprocessor.TEMP_TABLE,reduceScan,template,null);
        }
        return clientProvider;
	}

	private int[] generateHashKeys(int hashKeyItem) {
		FormatableArrayHolder fah = (FormatableArrayHolder)(activation.getPreparedStatement().getSavedObject(hashKeyItem));
		FormatableIntHolder[] fihArray = (FormatableIntHolder[]) fah.getArray(FormatableIntHolder.class);
		int[] keyColumns = new int[fihArray.length];
		for (int index = 0; index < fihArray.length; index++) {
			keyColumns[index] = fihArray[index].getInt();
		}
		return keyColumns;
	}
	
	@Override
	public void init(SpliceOperationContext context){
		SpliceLogUtils.trace(LOG, "init");
		super.init(context);
		try {
			leftHashKeys = generateHashKeys(leftHashKeyItem);
			rightHashKeys = generateHashKeys(rightHashKeyItem);
			mergedRow = activation.getExecutionFactory().getValueRow(leftNumCols + rightNumCols);
			rightTemplate = activation.getExecutionFactory().getValueRow(rightNumCols);
			byte[] start = DerbyBytesUtil.generateBeginKeyForTemp(sequence[0]);
			byte[] finish = DerbyBytesUtil.generateEndKeyForTemp(sequence[0]);
			rowType = (SQLInteger) activation.getDataValueFactory().getNullInteger(null);
			Hasher leftHasher = new Hasher(leftRow.getRowArray(),leftHashKeys,null,sequence[0]); 
			Hasher rightHasher = new Hasher(rightRow.getRowArray(),rightHashKeys,null,sequence[0]); 
			if(regionScanner==null){
				reduceScan = Scans.newScan(start,finish,transactionID);
			}else{
				serverProvider = new MergeSortRegionAwareRowProvider(context.getRegion(),SpliceOperationCoprocessor.TEMP_TABLE,HBaseConstants.DEFAULT_FAMILY_BYTES,
						start,finish,leftHasher,leftRow,rightHasher,rightRow,null,rowType);		
				serverProvider.open();
			}
			} catch (IOException e) {
				SpliceLogUtils.logAndThrowRuntime(LOG,"Unable to create reduce scan", e);
			} catch (StandardException e) {
				SpliceLogUtils.logAndThrowRuntime(LOG,"Unable to create reduce scan", e);
			}
	}
	<commit_msg>Rolled in MergeSort Fix for index out of bounds issues<commit_after>package com.splicemachine.derby.impl.sql.execute.operations;

import com.splicemachine.constants.HBaseConstants;
import com.splicemachine.derby.hbase.SpliceOperationCoprocessor;
import com.splicemachine.derby.iapi.sql.execute.SpliceNoPutResultSet;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperation;
import com.splicemachine.derby.iapi.sql.execute.SpliceOperationContext;
import com.splicemachine.derby.iapi.storage.RowProvider;
import com.splicemachine.derby.impl.storage.ClientScanProvider;
import com.splicemachine.derby.impl.storage.MergeSortRegionAwareRowProvider;
import com.splicemachine.derby.impl.store.access.SpliceAccessManager;
import com.splicemachine.derby.impl.store.access.hbase.HBaseRowLocation;
import com.splicemachine.derby.utils.*;
import com.splicemachine.utils.SpliceLogUtils;
import org.apache.derby.iapi.error.StandardException;
import org.apache.derby.iapi.services.io.FormatableArrayHolder;
import org.apache.derby.iapi.services.io.FormatableIntHolder;
import org.apache.derby.iapi.services.loader.GeneratedMethod;
import org.apache.derby.iapi.sql.Activation;
import org.apache.derby.iapi.sql.execute.ExecRow;
import org.apache.derby.iapi.sql.execute.NoPutResultSet;
import org.apache.derby.iapi.store.access.Qualifier;
import org.apache.derby.iapi.types.DataValueDescriptor;
import org.apache.derby.iapi.types.SQLInteger;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class MergeSortJoinOperation extends JoinOperation {
	private static Logger LOG = Logger.getLogger(MergeSortJoinOperation.class);
	protected String emptyRowFunMethodName;
	protected boolean wasRightOuterJoin;
	protected Qualifier[][] qualifierProbe;
	protected int leftHashKeyItem;
	protected int[] leftHashKeys;
	protected int rightHashKeyItem;
	protected int[] rightHashKeys;
	protected ExecRow rightTemplate;
	protected static List<NodeType> nodeTypes; 
	protected Scan reduceScan;
	protected enum JoinSide {RIGHT,LEFT};
	protected JoinSide joinSide;
	protected RowProvider clientProvider;
	protected MergeSortRegionAwareRowProvider serverProvider;
	protected SQLInteger rowType;
	protected byte[] priorHash;
	protected List<ExecRow> rights;
	protected byte[] rightHash;
	protected Iterator<ExecRow> rightIterator;
	
	static {
		nodeTypes = new ArrayList<NodeType>();
		nodeTypes.add(NodeType.REDUCE);
		nodeTypes.add(NodeType.SCAN);
		nodeTypes.add(NodeType.SINK);
	}
	
	public MergeSortJoinOperation() {
		super();
	}
	
	public MergeSortJoinOperation(NoPutResultSet leftResultSet,
			   int leftNumCols,
			   NoPutResultSet rightResultSet,
			   int rightNumCols,
			   int leftHashKeyItem,
			   int rightHashKeyItem,
			   Activation activation,
			   GeneratedMethod restriction,
			   int resultSetNumber,
			   boolean oneRowRightSide,
			   boolean notExistsRightSide,
			   double optimizerEstimatedRowCount,
			   double optimizerEstimatedCost,
			   String userSuppliedOptimizerOverrides) throws StandardException {		
				super(leftResultSet, leftNumCols, rightResultSet, rightNumCols,
						activation, restriction, resultSetNumber,oneRowRightSide, notExistsRightSide,
						optimizerEstimatedRowCount, optimizerEstimatedCost,userSuppliedOptimizerOverrides);
				SpliceLogUtils.trace(LOG, "instantiate");
				this.leftHashKeyItem = leftHashKeyItem;
				this.rightHashKeyItem = rightHashKeyItem;
				this.joinSide = JoinSide.LEFT;
                init(SpliceOperationContext.newContext(activation));
	}
	
	@Override
	public void readExternal(ObjectInput in) throws IOException,ClassNotFoundException {
		SpliceLogUtils.trace(LOG, "readExternal");
		super.readExternal(in);
		leftHashKeyItem = in.readInt();
		rightHashKeyItem = in.readInt();
		joinSide = JoinSide.values()[in.readInt()];
	}

	@Override
	public void writeExternal(ObjectOutput out) throws IOException {
		SpliceLogUtils.trace(LOG, "writeExternal");
		super.writeExternal(out);
		out.writeInt(leftHashKeyItem);
		out.writeInt(rightHashKeyItem);
		out.writeInt(joinSide.ordinal());
	}

	@Override
	public ExecRow getNextRowCore() throws StandardException {
		SpliceLogUtils.trace(LOG, "getNextRowCore");
		if (rightIterator!= null && rightIterator.hasNext()) {
			currentRow = JoinUtils.getMergedRow(leftRow, rightIterator.next(), wasRightOuterJoin, this.leftNumCols, this.rightNumCols, mergedRow);
			this.setCurrentRow(currentRow);
			currentRowLocation = new HBaseRowLocation(SpliceUtils.getUniqueKey());
			SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
			return currentRow;
		}
		if (!serverProvider.hasNext()) {
			SpliceLogUtils.trace(LOG, "serverProvider exhausted");
			return null;
		}
		JoinSideExecRow joinRow = serverProvider.nextJoinRow();
		SpliceLogUtils.trace(LOG, "joinRow=%s",joinRow);
		if (joinRow == null) {
			SpliceLogUtils.trace(LOG, "serverProvider returned null rows");
			this.setCurrentRow(null);
			return null;
		}
		
		if (joinRow.getJoinSide().ordinal() == JoinSide.RIGHT.ordinal()) { // Right Side
			rightHash = joinRow.getHash();
			if (joinRow.sameHash(priorHash)) {
				SpliceLogUtils.trace(LOG, "adding additional right=%s",joinRow);
				rights.add(joinRow.getRow());
			} else {
				resetRightSide();
				SpliceLogUtils.trace(LOG, "adding initial right=%s",joinRow);
				rights.add(joinRow.getRow());
				priorHash = joinRow.getHash();
			}
			return getNextRowCore();
		} else { // Left Side
			leftRow = joinRow.getRow();
			if (joinRow.sameHash(priorHash)) {
				if (joinRow.sameHash(rightHash)) {
					SpliceLogUtils.trace(LOG, "initializing iterator with rights for left=%s",joinRow);
					rightIterator = rights.iterator();
					currentRow = JoinUtils.getMergedRow(leftRow, rightIterator.next(), wasRightOuterJoin, this.leftNumCols, this.rightNumCols, mergedRow);
					this.setCurrentRow(currentRow);
					currentRowLocation = new HBaseRowLocation(SpliceUtils.getUniqueKey());					
					SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
					return currentRow;
				} else {
					SpliceLogUtils.trace(LOG, "right hash miss left=%s",joinRow);
					resetRightSide();	
					priorHash = joinRow.getHash();
					return getNextRowCore();
				}
			} else {
				SpliceLogUtils.trace(LOG, "simple left with no right=%s",joinRow);
				resetRightSide();
				priorHash = joinRow.getHash();
				SpliceLogUtils.trace(LOG, "current row returned %s",currentRow);
				return getNextRowCore();			
			}			
		}
	}

	@Override
	public RowProvider getReduceRowProvider(SpliceOperation top,ExecRow template){
        if(clientProvider==null){
            SpliceUtils.setInstructions(reduceScan,activation,top);
            clientProvider = new ClientScanProvider(SpliceOperationCoprocessor.TEMP_TABLE,reduceScan,template,null);
        }
        return clientProvider;
	}

	private int[] generateHashKeys(int hashKeyItem, SpliceBaseOperation resultSet) {
		FormatableArrayHolder fah = (FormatableArrayHolder)(activation.getPreparedStatement().getSavedObject(hashKeyItem));
		FormatableIntHolder[] fihArray = (FormatableIntHolder[]) fah.getArray(FormatableIntHolder.class);
		int[] keyColumns = new int[fihArray.length];
		for (int index = 0; index < fihArray.length; index++) {
			keyColumns[index] = FormatableBitSetUtils.currentRowPositionFromBaseRow(resultSet.getRootAccessedCols(),fihArray[index].getInt());
		}
		return keyColumns;
	}
	
	@Override
	public void init(SpliceOperationContext context){
		SpliceLogUtils.trace(LOG, "init");
		super.init(context);
		try {
			leftHashKeys = generateHashKeys(leftHashKeyItem, (SpliceBaseOperation) this.leftResultSet);
			rightHashKeys = generateHashKeys(rightHashKeyItem, (SpliceBaseOperation) this.rightResultSet);
			mergedRow = activation.getExecutionFactory().getValueRow(leftNumCols + rightNumCols);
			rightTemplate = activation.getExecutionFactory().getValueRow(rightNumCols);
			byte[] start = DerbyBytesUtil.generateBeginKeyForTemp(sequence[0]);
			byte[] finish = DerbyBytesUtil.generateEndKeyForTemp(sequence[0]);
			rowType = (SQLInteger) activation.getDataValueFactory().getNullInteger(null);
			Hasher leftHasher = new Hasher(leftRow.getRowArray(),leftHashKeys,null,sequence[0]); 
			Hasher rightHasher = new Hasher(rightRow.getRowArray(),rightHashKeys,null,sequence[0]); 
			if(regionScanner==null){
				reduceScan = Scans.newScan(start,finish,transactionID);
			}else{
				serverProvider = new MergeSortRegionAwareRowProvider(context.getRegion(),SpliceOperationCoprocessor.TEMP_TABLE,HBaseConstants.DEFAULT_FAMILY_BYTES,
						start,finish,leftHasher,leftRow,rightHasher,rightRow,null,rowType);		
				serverProvider.open();
			}
			} catch (IOException e) {
				SpliceLogUtils.logAndThrowRuntime(LOG,"Unable to create reduce scan", e);
			} catch (StandardException e) {
				SpliceLogUtils.logAndThrowRuntime(LOG,"Unable to create reduce scan", e);
			}
	}
	
{'fraction_non_alphanumeric': 0.10561349012646994, 'fraction_numerical': 0.0006656312402928778, 'mean_word_length': 5.462007168458781, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, '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': '4561273', 'n_tokens_mistral': 6204, 'n_tokens_neox': 5503, 'n_words': 949}