<commit_before>package water.rapids;

// Since we have a single key field in H2O (different to data.table), bmerge() becomes a lot simpler (no
// need for recursion through join columns) with a downside of transfer-cost should we not need all the key.

import water.*;
import water.fvec.Chunk;
import water.fvec.Frame;
import water.fvec.NewChunk;
import water.fvec.Vec;
import static water.rapids.SingleThreadRadixOrder.getSortedOXHeaderKey;
import water.util.ArrayUtils;
import water.util.Log;

import java.util.Arrays;

public class BinaryMerge extends DTask<BinaryMerge> {
  long _numRowsInResult=0;  // returned to caller, so not transient
  int _chunkSizes[];      // TODO:  only _chunkSizes.length is needed by caller, so return that length only
  double _timings[];

  transient long _retFirst[/*n2GB*/][];  // The row number of the first right table's index key that matches
  transient long _retLen[/*n2GB*/][];    // How many rows does it match to?
  transient byte _leftKey[/*n2GB*/][/*i mod 2GB * _keySize*/];
  transient byte _rightKey[][];
  transient long _leftOrder[/*n2GB*/][/*i mod 2GB * _keySize*/];
  transient long _rightOrder[][];
  transient boolean _oneToManyMatch = false;   // does any left row match to more than 1 right row?  If not, can allocate and loop more efficiently, and mark the resulting key'd frame with a 'unique' index.
                                               //      TODO: implement
  int _leftFieldSizes[], _rightFieldSizes[];  // the widths of each column in the key
  transient int _leftKeyNCol, _rightKeyNCol;  // the number of columns in the key i.e. length of _leftFieldSizes and _rightFieldSizes
  transient int _leftKeySize, _rightKeySize;   // the total width in bytes of the key, sum of field sizes
  transient int _numJoinCols;
  transient long _leftN, _rightN;
  transient long _leftBatchSize, _rightBatchSize;
  Frame _leftFrame, _rightFrame;

  transient long _perNodeNumRightRowsToFetch[];
  transient long _perNodeNumLeftRowsToFetch[];
  int _leftMSB, _rightMSB;

  boolean _allLeft, _allRight;

  BinaryMerge(Frame leftFrame, Frame rightFrame, int leftMSB, int rightMSB, int leftFieldSizes[], int rightFieldSizes[], boolean allLeft) {   // In X[Y], 'left'=i and 'right'=x
    _leftFrame = leftFrame;
    _rightFrame = rightFrame;
    _leftMSB = leftMSB;
    _rightMSB = rightMSB;
    _leftFieldSizes = leftFieldSizes;
    _rightFieldSizes = rightFieldSizes;
    _allLeft = allLeft;
    _allRight = false;  // TODO: pass through
    // TODO: set 2 Frame and 2 int[] to NULL at the end of compute2 to save some traffic back, but should be small and insignificant
  }

  @Override
  protected void compute2() {
    _timings = new double[12];
    long t0 = System.nanoTime();
    SingleThreadRadixOrder.OXHeader leftSortedOXHeader = DKV.getGet(getSortedOXHeaderKey(/*left=*/true, _leftMSB));
    if (leftSortedOXHeader == null) {
      if (_allRight) throw H2O.unimpl();  // TODO pass through _allRight and implement
      tryComplete(); return;
    }
    SingleThreadRadixOrder.OXHeader rightSortedOXHeader = DKV.getGet(getSortedOXHeaderKey(/*left=*/false, _rightMSB));
    if (rightSortedOXHeader == null) {
      if (_allLeft == false) { tryComplete(); return; }
      rightSortedOXHeader = new SingleThreadRadixOrder.OXHeader(0, 0, 0);  // enables general case code to run below without needing new special case code
    }
    // both left and right MSB have some data to match
    _leftBatchSize = leftSortedOXHeader._batchSize;
    _rightBatchSize = rightSortedOXHeader._batchSize;
    _perNodeNumRightRowsToFetch = new long[H2O.CLOUD.size()];
    _perNodeNumLeftRowsToFetch = new long[H2O.CLOUD.size()];

    // get left batches
    _leftKey = new byte[leftSortedOXHeader._nBatch][];
    _leftOrder = new long[leftSortedOXHeader._nBatch][];
    _retFirst = new long[leftSortedOXHeader._nBatch][];
    _retLen = new long[leftSortedOXHeader._nBatch][];
    for (int b=0; b<leftSortedOXHeader._nBatch; ++b) {
      SplitByMSBLocal.OXbatch oxLeft = DKV.getGet(SplitByMSBLocal.getSortedOXbatchKey(/*left=*/true, _leftMSB, b));
      _leftKey[b] = oxLeft._x;
      _leftOrder[b] = oxLeft._o;
      _retFirst[b] = new long[oxLeft._o.length];
      _retLen[b] = new long[oxLeft._o.length];
    }
    _leftN = leftSortedOXHeader._numRows;

    // get right batches
    _rightKey = new byte[rightSortedOXHeader._nBatch][];
    _rightOrder = new long[rightSortedOXHeader._nBatch][];
    for (int b=0; b<rightSortedOXHeader._nBatch; ++b) {
      SplitByMSBLocal.OXbatch oxRight = DKV.getGet(SplitByMSBLocal.getSortedOXbatchKey(/*left=*/false, _rightMSB, b));
      _rightKey[b] = oxRight._x;
      _rightOrder[b] = oxRight._o;
    }
    _rightN = rightSortedOXHeader._numRows;

    _leftKeyNCol = _leftFieldSizes.length;
    _rightKeyNCol = _rightFieldSizes.length;
    _leftKeySize = ArrayUtils.sum(_leftFieldSizes);
    _rightKeySize = ArrayUtils.sum(_rightFieldSizes);
    _numJoinCols = Math.min(_leftKeyNCol, _rightKeyNCol);

    _timings[0] = (System.nanoTime() - t0) / 1e9;

    if ((_leftN != 0 || _allRight) && (_rightN != 0 || _allLeft)) {

      t0 = System.nanoTime();
      bmerge_r(-1, _leftN, -1, _rightN);
      _timings[1] = (System.nanoTime() - t0) / 1e9;

      if (_numRowsInResult > 0) createChunksInDKV();
    }

    //null out members before returning to calling node
    tryComplete();
  }


  private int keycmp(byte x[][], long xi, byte y[][], long yi) {   // TO DO - faster way closer to CPU like batches of long compare, maybe.
    byte xByte=0, yByte=0;
    xi *= _leftKeySize;
    yi *= _rightKeySize;   // x[] and y[] are len keys.
    // TO DO: rationalize x and y being chunked into 2GB pieces.  Take x[0][] and y[0][] outside loop / function
    // TO DO: switch to use keycmp_sameShape() for common case of all(leftFieldSizes == rightFieldSizes), although, skipping to current column will
    //        help save repeating redundant work and saving the outer for() loop and one if() may not be worth it.
    int i=0, xlen=0, ylen=0, diff=0;
    while (i<_numJoinCols && xlen==0) {    // TO DO: pass i in to start at a later key column, when known
      xlen = _leftFieldSizes[i];
      ylen = _rightFieldSizes[i];
      if (xlen!=ylen) {
        while (xlen>ylen && x[0][(int)xi]==0) { xi++; xlen--; }
        while (ylen>xlen && y[0][(int)yi]==0) { yi++; ylen--; }
        if (xlen!=ylen) return (xlen - ylen);
      }
      while (xlen>0 && (xByte=x[0][(int)xi])==(yByte=y[0][(int)yi])) { xi++; yi++; xlen--; }
      i++;
    }
    return (xByte & 0xFF) - (yByte & 0xFF);
    // Same return value as strcmp in C. <0 => xi<yi
  }

  private void bmerge_r(long lLowIn, long lUppIn, long rLowIn, long rUppIn) {
    // TO DO: parallel each of the 256 bins
    long lLow = lLowIn, lUpp = lUppIn, rLow = rLowIn, rUpp = rUppIn;
    long mid, tmpLow, tmpUpp;
    long lr = lLow + (lUpp - lLow) / 2;   // i.e. (lLow+lUpp)/2 but being robust to one day in the future someone somewhere overflowing long; e.g. 32 exabytes of 1-column ints
    while (rLow < rUpp - 1) {
      mid = rLow + (rUpp - rLow) / 2;
      int cmp = keycmp(_leftKey, lr, _rightKey, mid);  // -1, 0 or 1, like strcmp
      if (cmp < 0) {
        rUpp = mid;
      } else if (cmp > 0) {
        rLow = mid;<commit_msg>Made keycmp batch aware, gets further with 1E10 now.<commit_after>package water.rapids;

// Since we have a single key field in H2O (different to data.table), bmerge() becomes a lot simpler (no
// need for recursion through join columns) with a downside of transfer-cost should we not need all the key.

import water.*;
import water.fvec.Chunk;
import water.fvec.Frame;
import water.fvec.NewChunk;
import water.fvec.Vec;
import static water.rapids.SingleThreadRadixOrder.getSortedOXHeaderKey;
import water.util.ArrayUtils;
import water.util.Log;

import java.util.Arrays;

public class BinaryMerge extends DTask<BinaryMerge> {
  long _numRowsInResult=0;  // returned to caller, so not transient
  int _chunkSizes[];      // TODO:  only _chunkSizes.length is needed by caller, so return that length only
  double _timings[];

  transient long _retFirst[/*n2GB*/][];  // The row number of the first right table's index key that matches
  transient long _retLen[/*n2GB*/][];    // How many rows does it match to?
  transient byte _leftKey[/*n2GB*/][/*i mod 2GB * _keySize*/];
  transient byte _rightKey[][];
  transient long _leftOrder[/*n2GB*/][/*i mod 2GB * _keySize*/];
  transient long _rightOrder[][];
  transient boolean _oneToManyMatch = false;   // does any left row match to more than 1 right row?  If not, can allocate and loop more efficiently, and mark the resulting key'd frame with a 'unique' index.
                                               //      TODO: implement
  int _leftFieldSizes[], _rightFieldSizes[];  // the widths of each column in the key
  transient int _leftKeyNCol, _rightKeyNCol;  // the number of columns in the key i.e. length of _leftFieldSizes and _rightFieldSizes
  transient int _leftKeySize, _rightKeySize;   // the total width in bytes of the key, sum of field sizes
  transient int _numJoinCols;
  transient long _leftN, _rightN;
  transient long _leftBatchSize, _rightBatchSize;
  Frame _leftFrame, _rightFrame;

  transient long _perNodeNumRightRowsToFetch[];
  transient long _perNodeNumLeftRowsToFetch[];
  int _leftMSB, _rightMSB;

  boolean _allLeft, _allRight;

  BinaryMerge(Frame leftFrame, Frame rightFrame, int leftMSB, int rightMSB, int leftFieldSizes[], int rightFieldSizes[], boolean allLeft) {   // In X[Y], 'left'=i and 'right'=x
    _leftFrame = leftFrame;
    _rightFrame = rightFrame;
    _leftMSB = leftMSB;
    _rightMSB = rightMSB;
    _leftFieldSizes = leftFieldSizes;
    _rightFieldSizes = rightFieldSizes;
    _allLeft = allLeft;
    _allRight = false;  // TODO: pass through
    // TODO: set 2 Frame and 2 int[] to NULL at the end of compute2 to save some traffic back, but should be small and insignificant
  }

  @Override
  protected void compute2() {
    _timings = new double[12];
    long t0 = System.nanoTime();
    SingleThreadRadixOrder.OXHeader leftSortedOXHeader = DKV.getGet(getSortedOXHeaderKey(/*left=*/true, _leftMSB));
    if (leftSortedOXHeader == null) {
      if (_allRight) throw H2O.unimpl();  // TODO pass through _allRight and implement
      tryComplete(); return;
    }
    SingleThreadRadixOrder.OXHeader rightSortedOXHeader = DKV.getGet(getSortedOXHeaderKey(/*left=*/false, _rightMSB));
    if (rightSortedOXHeader == null) {
      if (_allLeft == false) { tryComplete(); return; }
      rightSortedOXHeader = new SingleThreadRadixOrder.OXHeader(0, 0, 0);  // enables general case code to run below without needing new special case code
    }
    // both left and right MSB have some data to match
    _leftBatchSize = leftSortedOXHeader._batchSize;
    _rightBatchSize = rightSortedOXHeader._batchSize;
    _perNodeNumRightRowsToFetch = new long[H2O.CLOUD.size()];
    _perNodeNumLeftRowsToFetch = new long[H2O.CLOUD.size()];

    // get left batches
    _leftKey = new byte[leftSortedOXHeader._nBatch][];
    _leftOrder = new long[leftSortedOXHeader._nBatch][];
    _retFirst = new long[leftSortedOXHeader._nBatch][];
    _retLen = new long[leftSortedOXHeader._nBatch][];
    for (int b=0; b<leftSortedOXHeader._nBatch; ++b) {
      SplitByMSBLocal.OXbatch oxLeft = DKV.getGet(SplitByMSBLocal.getSortedOXbatchKey(/*left=*/true, _leftMSB, b));
      _leftKey[b] = oxLeft._x;
      _leftOrder[b] = oxLeft._o;
      _retFirst[b] = new long[oxLeft._o.length];
      _retLen[b] = new long[oxLeft._o.length];
    }
    _leftN = leftSortedOXHeader._numRows;

    // get right batches
    _rightKey = new byte[rightSortedOXHeader._nBatch][];
    _rightOrder = new long[rightSortedOXHeader._nBatch][];
    for (int b=0; b<rightSortedOXHeader._nBatch; ++b) {
      SplitByMSBLocal.OXbatch oxRight = DKV.getGet(SplitByMSBLocal.getSortedOXbatchKey(/*left=*/false, _rightMSB, b));
      _rightKey[b] = oxRight._x;
      _rightOrder[b] = oxRight._o;
    }
    _rightN = rightSortedOXHeader._numRows;

    _leftKeyNCol = _leftFieldSizes.length;
    _rightKeyNCol = _rightFieldSizes.length;
    _leftKeySize = ArrayUtils.sum(_leftFieldSizes);
    _rightKeySize = ArrayUtils.sum(_rightFieldSizes);
    _numJoinCols = Math.min(_leftKeyNCol, _rightKeyNCol);

    _timings[0] = (System.nanoTime() - t0) / 1e9;

    if ((_leftN != 0 || _allRight) && (_rightN != 0 || _allLeft)) {

      t0 = System.nanoTime();
      bmerge_r(-1, _leftN, -1, _rightN);
      _timings[1] = (System.nanoTime() - t0) / 1e9;

      if (_numRowsInResult > 0) createChunksInDKV();
    }

    //null out members before returning to calling node
    tryComplete();
  }


  private int keycmp(byte x[][], long xi, byte y[][], long yi) {   // TO DO - faster way closer to CPU like batches of long compare, maybe.
    byte xbatch[] = x[(int)(xi / _leftBatchSize)];
    byte ybatch[] = y[(int)(yi / _rightBatchSize)];
    int xoff = (int)(xi % _leftBatchSize) * _leftKeySize;
    int yoff = (int)(yi % _rightBatchSize) * _rightKeySize;
    byte xByte=0, yByte=0;
    // TODO: switch to use keycmp_sameShape() for common case of all(leftFieldSizes == rightFieldSizes), although, skipping to current column will
    //        help save repeating redundant work and saving the outer for() loop and one if() may not be worth it.
    int i=0, xlen=0, ylen=0;

    while (i<_numJoinCols && xlen==0) {    // TO DO: pass i in to start at a later key column, when known
      xlen = _leftFieldSizes[i];
      ylen = _rightFieldSizes[i];
      if (xlen!=ylen) {
        while (xlen>ylen && xbatch[xoff]==0) { xoff++; xlen--; }
        while (ylen>xlen && ybatch[yoff]==0) { yoff++; ylen--; }
        if (xlen!=ylen) return (xlen - ylen);
      }
      while (xlen>0 && (xByte=xbatch[xoff])==(yByte=ybatch[yoff])) { xoff++; yoff++; xlen--; }
      i++;
    }
    return (xByte & 0xFF) - (yByte & 0xFF);
    // Same return value as strcmp in C. <0 => xi<yi
  }

  private void bmerge_r(long lLowIn, long lUppIn, long rLowIn, long rUppIn) {
    // TO DO: parallel each of the 256 bins
    long lLow = lLowIn, lUpp = lUppIn, rLow = rLowIn, rUpp = rUppIn;
    long mid, tmpLow, tmpUpp;
    long lr = lLow + (lUpp - lLow) / 2;   // i.e. (lLow+lUpp)/2 but being robust to one day in the future someone somewhere overflowing long; e.g. 32 exabytes of 1-column ints
    while (rLow < rUpp - 1) {
      mid = rLow + (rUpp - rLow) / 2;
      int cmp = keycmp(_leftKey, lr, _rightKey, mid);  // -1, 0 or 1, like strcmp
      if (cmp < 0) {
        rUpp = mid;
      } else if (cmp > 0) {
        rLow = mid;
{'fraction_non_alphanumeric': 0.115866602752859, 'fraction_numerical': 0.009655550229404918, 'mean_word_length': 3.3438429506246283, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6075515', 'n_tokens_mistral': 5100, 'n_tokens_neox': 4969, 'n_words': 1451}