private List<Tuple2<Long,BigInteger>> processTupleFromPartitionDataBolt(Tuple tuple){
  matrixElements.clear();
  int rowIndex=tuple.getIntegerByField(StormConstants.HASH_FIELD);
  if (!colIndexByRow.containsKey(rowIndex)) {
    colIndexByRow.put(rowIndex,0);
    hitsByRow.put(rowIndex,0);
  }
  if (splitPartitions) {
    dataArray.add((BigInteger)tuple.getValueByField(StormConstants.PARTIONED_DATA_FIELD));
  }
 else {
    dataArray=(ArrayList<BigInteger>)tuple.getValueByField(StormConstants.PARTIONED_DATA_FIELD);
  }
  logger.debug("Retrieving {} elements in EncRowCalcBolt.",dataArray.size());
  try {
    int colIndex=colIndexByRow.get(rowIndex);
    int numRecords=hitsByRow.get(rowIndex);
    if (limitHitsPerSelector && numRecords < maxHitsPerSelector) {
      logger.debug("computing matrix elements.");
      matrixElements=ComputeEncryptedRow.computeEncRow(dataArray,query,rowIndex,colIndex);
      colIndexByRow.put(rowIndex,colIndex + matrixElements.size());
      hitsByRow.put(rowIndex,numRecords + 1);
    }
 else     if (limitHitsPerSelector) {
      logger.info("maxHits: rowIndex = " + rowIndex + " elementCounter = "+ numRecords);
    }
  }
 catch (  IOException e) {
    logger.warn("Caught IOException while encrypting row. ",e);
  }
  dataArray.clear();
  return matrixElements;
}

private List < Tuple2 < Long , BigInteger > > processTupleFromPartitionDataBolt ( Tuple tuple ) { matrixElements . clear ( ) ; int rowIndex = tuple . getIntegerByField ( StormConstants . HASH_FIELD ) ; if ( ! colIndexByRow . containsKey ( rowIndex ) ) { colIndexByRow . put ( rowIndex , 0 ) ; hitsByRow . put ( rowIndex , 0 ) ; } if ( splitPartitions ) { dataArray . add ( ( BigInteger ) tuple . getValueByField ( StormConstants . PARTIONED_DATA_FIELD ) ) ; } else { dataArray = ( ArrayList < BigInteger > ) tuple . getValueByField ( StormConstants . PARTIONED_DATA_FIELD ) ; } logger . debug ( "Retrieving {} elements in EncRowCalcBolt." , dataArray . size ( ) ) ; try { int colIndex = colIndexByRow . get ( rowIndex ) ; int numRecords = hitsByRow . get ( rowIndex ) ; if ( limitHitsPerSelector && numRecords < maxHitsPerSelector ) { logger . debug ( "computing matrix elements." ) ; matrixElements = ComputeEncryptedRow . computeEncRow ( dataArray , query , rowIndex , colIndex ) ; colIndexByRow . put ( rowIndex , colIndex + matrixElements . size ( ) ) ; hitsByRow . put ( rowIndex , numRecords + 1 ) ; } else if ( limitHitsPerSelector ) { logger . info ( "maxHits: rowIndex = " + rowIndex + " elementCounter = " + numRecords ) ; } } catch ( IOException e ) { logger . warn ( "Caught IOException while encrypting row. " , e ) ; } dataArray . clear ( ) ; return matrixElements ; }
extracts ( hash , data partitions ) from tuple . encrypts the data partitions . returns all of the pairs of ( col index , col value ) . also advances the colindexbyrow and hitsbyrow appropriately .
