output: #fixed code
                                  @Override
      protected RegionScanner doPostScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws IOException, SQLException {
          RegionCoprocessorEnvironment env = c.getEnvironment();
          Region region = env.getRegion();
          long ts = scan.getTimeRange().getMax();
          boolean localIndexScan = ScanUtil.isLocalIndex(scan);
          if (ScanUtil.isAnalyzeTable(scan)) {
              byte[] gp_width_bytes =
                      scan.getAttribute(BaseScannerRegionObserver.GUIDEPOST_WIDTH_BYTES);
              byte[] gp_per_region_bytes =
                      scan.getAttribute(BaseScannerRegionObserver.GUIDEPOST_PER_REGION);
              // Let this throw, as this scan is being done for the sole purpose of collecting stats
              StatisticsCollector statsCollector = StatisticsCollectorFactory.createStatisticsCollector(
                      env, region.getRegionInfo().getTable().getNameAsString(), ts,
                      gp_width_bytes, gp_per_region_bytes);
              return collectStats(s, statsCollector, region, scan, env.getConfiguration());
          } else if (ScanUtil.isIndexRebuild(scan)) { return rebuildIndices(s, region, scan, env.getConfiguration()); }
          int offsetToBe = 0;
          if (localIndexScan) {
              /*
               * For local indexes, we need to set an offset on row key expressions to skip
               * the region start key.
               */
              offsetToBe = region.getRegionInfo().getStartKey().length != 0 ? region.getRegionInfo().getStartKey().length :
                  region.getRegionInfo().getEndKey().length;
              ScanUtil.setRowKeyOffset(scan, offsetToBe);
          }
          final int offset = offsetToBe;
          
          PTable projectedTable = null;
          PTable writeToTable = null;
          byte[][] values = null;
          byte[] descRowKeyTableBytes = scan.getAttribute(UPGRADE_DESC_ROW_KEY);
          boolean isDescRowKeyOrderUpgrade = descRowKeyTableBytes != null;
          if (isDescRowKeyOrderUpgrade) {
              logger.debug("Upgrading row key for " + region.getRegionInfo().getTable().getNameAsString());
              projectedTable = deserializeTable(descRowKeyTableBytes);
              try {
                  writeToTable = PTableImpl.makePTable(projectedTable, true);
              } catch (SQLException e) {
                  ServerUtil.throwIOException("Upgrade failed", e); // Impossible
              }
              values = new byte[projectedTable.getPKColumns().size()][];
          }
          byte[] localIndexBytes = scan.getAttribute(LOCAL_INDEX_BUILD);
          List<IndexMaintainer> indexMaintainers = localIndexBytes == null ? null : IndexMaintainer.deserialize(localIndexBytes);
          List<Mutation> indexMutations = localIndexBytes == null ? Collections.<Mutation>emptyList() : Lists.<Mutation>newArrayListWithExpectedSize(1024);
          
          RegionScanner theScanner = s;
          
          byte[] indexUUID = scan.getAttribute(PhoenixIndexCodec.INDEX_UUID);
          byte[] txState = scan.getAttribute(BaseScannerRegionObserver.TX_STATE);
          List<Expression> selectExpressions = null;
          byte[] upsertSelectTable = scan.getAttribute(BaseScannerRegionObserver.UPSERT_SELECT_TABLE);
          boolean isUpsert = false;
          boolean isDelete = false;
          byte[] deleteCQ = null;
          byte[] deleteCF = null;
          byte[] emptyCF = null;
          HTable targetHTable = null;
          boolean areMutationInSameRegion = true;
          ImmutableBytesWritable ptr = new ImmutableBytesWritable();
          if (upsertSelectTable != null) {
              isUpsert = true;
              projectedTable = deserializeTable(upsertSelectTable);
              targetHTable = new HTable(env.getConfiguration(), projectedTable.getPhysicalName().getBytes());
              selectExpressions = deserializeExpressions(scan.getAttribute(BaseScannerRegionObserver.UPSERT_SELECT_EXPRS));
              values = new byte[projectedTable.getPKColumns().size()][];
              areMutationInSameRegion = Bytes.compareTo(targetHTable.getTableName(),
                      region.getTableDesc().getTableName().getName()) == 0
                      && !isPkPositionChanging(new TableRef(projectedTable), selectExpressions);
              
          } else {
              byte[] isDeleteAgg = scan.getAttribute(BaseScannerRegionObserver.DELETE_AGG);
              isDelete = isDeleteAgg != null && Bytes.compareTo(PDataType.TRUE_BYTES, isDeleteAgg) == 0;
              if (!isDelete) {
                  deleteCF = scan.getAttribute(BaseScannerRegionObserver.DELETE_CF);
                  deleteCQ = scan.getAttribute(BaseScannerRegionObserver.DELETE_CQ);
              }
              emptyCF = scan.getAttribute(BaseScannerRegionObserver.EMPTY_CF);
          }
          TupleProjector tupleProjector = null;
          byte[][] viewConstants = null;
          ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan);
          final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan);
          final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
          if ((localIndexScan && !isDelete && !isDescRowKeyOrderUpgrade) || (j == null && p != null)) {
              if (dataColumns != null) {
                  tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns);
                  viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan);
              }
              ImmutableBytesWritable tempPtr = new ImmutableBytesWritable();
              theScanner =
                      getWrappedScanner(c, theScanner, offset, scan, dataColumns, tupleProjector, 
                          region, indexMaintainers == null ? null : indexMaintainers.get(0), viewConstants, p, tempPtr);
          } 
          
          if (j != null)  {
              theScanner = new HashJoinRegionScanner(theScanner, p, j, ScanUtil.getTenantId(scan), env);
          }
          
          int batchSize = 0;
          long batchSizeBytes = 0L;
          List<Mutation> mutations = Collections.emptyList();
          boolean needToWrite = false;
          Configuration conf = c.getEnvironment().getConfiguration();
          long flushSize = region.getTableDesc().getMemStoreFlushSize();
  
          if (flushSize <= 0) {
              flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
                      HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);
          }
  
          /**
           * Slow down the writes if the memstore size more than
           * (hbase.hregion.memstore.block.multiplier - 1) times hbase.hregion.memstore.flush.size
           * bytes. This avoids flush storm to hdfs for cases like index building where reads and
           * write happen to all the table regions in the server.
           */
          final long blockingMemStoreSize = flushSize * (
                  conf.getLong(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER,
                          HConstants.DEFAULT_HREGION_MEMSTORE_BLOCK_MULTIPLIER)-1) ;
  
          boolean buildLocalIndex = indexMaintainers != null && dataColumns==null && !localIndexScan;
          if (isDescRowKeyOrderUpgrade || isDelete || isUpsert || (deleteCQ != null && deleteCF != null) || emptyCF != null || buildLocalIndex) {
              needToWrite = true;
              // TODO: size better
              mutations = Lists.newArrayListWithExpectedSize(1024);
              batchSize = env.getConfiguration().getInt(MUTATE_BATCH_SIZE_ATTRIB, QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
              batchSizeBytes = env.getConfiguration().getLong(MUTATE_BATCH_SIZE_BYTES_ATTRIB,
                  QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE_BYTES);
          }
          Aggregators aggregators = ServerAggregators.deserialize(
                  scan.getAttribute(BaseScannerRegionObserver.AGGREGATORS), env.getConfiguration());
          Aggregator[] rowAggregators = aggregators.getAggregators();
          boolean hasMore;
          boolean hasAny = false;
          MultiKeyValueTuple result = new MultiKeyValueTuple();
          if (logger.isDebugEnabled()) {
              logger.debug(LogUtil.addCustomAnnotations("Starting ungrouped coprocessor scan " + scan + " "+region.getRegionInfo(), ScanUtil.getCustomAnnotations(scan)));
          }
          long rowCount = 0;
          final RegionScanner innerScanner = theScanner;
          byte[] indexMaintainersPtr = scan.getAttribute(PhoenixIndexCodec.INDEX_MD);
          boolean acquiredLock = false;
          try {
              if(needToWrite) {
                  synchronized (lock) {
                      scansReferenceCount++;
                  }
              }
              region.startRegionOperation();
              acquiredLock = true;
              synchronized (innerScanner) {
                  do {
                      List<Cell> results = new ArrayList<Cell>();
                      // Results are potentially returned even when the return value of s.next is false
                      // since this is an indication of whether or not there are more values after the
                      // ones returned
                      hasMore = innerScanner.nextRaw(results);
                      if (!results.isEmpty()) {
                          rowCount++;
                          result.setKeyValues(results);
                          if (isDescRowKeyOrderUpgrade) {
                              Arrays.fill(values, null);
                              Cell firstKV = results.get(0);
                              RowKeySchema schema = projectedTable.getRowKeySchema();
                              int maxOffset = schema.iterator(firstKV.getRowArray(), firstKV.getRowOffset() + offset, firstKV.getRowLength(), ptr);
                              for (int i = 0; i < schema.getFieldCount(); i++) {
                                  Boolean hasValue = schema.next(ptr, i, maxOffset);
                                  if (hasValue == null) {
                                      break;
                                  }
                                  Field field = schema.getField(i);
                                  if (field.getSortOrder() == SortOrder.DESC) {
                                      // Special case for re-writing DESC ARRAY, as the actual byte value needs to change in this case
                                      if (field.getDataType().isArrayType()) {
                                          field.getDataType().coerceBytes(ptr, null, field.getDataType(),
                                              field.getMaxLength(), field.getScale(), field.getSortOrder(), 
                                              field.getMaxLength(), field.getScale(), field.getSortOrder(), true); // force to use correct separator byte
                                      }
                                      // Special case for re-writing DESC CHAR or DESC BINARY, to force the re-writing of trailing space characters
                                      else if (field.getDataType() == PChar.INSTANCE || field.getDataType() == PBinary.INSTANCE) {
                                          int len = ptr.getLength();
                                          while (len > 0 && ptr.get()[ptr.getOffset() + len - 1] == StringUtil.SPACE_UTF8) {
                                              len--;
                                          }
                                          ptr.set(ptr.get(), ptr.getOffset(), len);
                                          // Special case for re-writing DESC FLOAT and DOUBLE, as they're not inverted like they should be (PHOENIX-2171)
                                      } else if (field.getDataType() == PFloat.INSTANCE || field.getDataType() == PDouble.INSTANCE) {
                                          byte[] invertedBytes = SortOrder.invert(ptr.get(), ptr.getOffset(), ptr.getLength());
                                          ptr.set(invertedBytes);
                                      }
                                  } else if (field.getDataType() == PBinary.INSTANCE) {
                                      // Remove trailing space characters so that the setValues call below will replace them
                                      // with the correct zero byte character. Note this is somewhat dangerous as these
                                      // could be legit, but I don't know what the alternative is.
                                      int len = ptr.getLength();
                                      while (len > 0 && ptr.get()[ptr.getOffset() + len - 1] == StringUtil.SPACE_UTF8) {
                                          len--;
                                      }
                                      ptr.set(ptr.get(), ptr.getOffset(), len);                                        
                                  }
                                  values[i] = ptr.copyBytes();
                              }
                              writeToTable.newKey(ptr, values);
                              if (Bytes.compareTo(
                                  firstKV.getRowArray(), firstKV.getRowOffset() + offset, firstKV.getRowLength(), 
                                  ptr.get(),ptr.getOffset() + offset,ptr.getLength()) == 0) {
                                  continue;
                              }
                              byte[] newRow = ByteUtil.copyKeyBytesIfNecessary(ptr);
                              if (offset > 0) { // for local indexes (prepend region start key)
                                  byte[] newRowWithOffset = new byte[offset + newRow.length];
                                  System.arraycopy(firstKV.getRowArray(), firstKV.getRowOffset(), newRowWithOffset, 0, offset);;
                                  System.arraycopy(newRow, 0, newRowWithOffset, offset, newRow.length);
                                  newRow = newRowWithOffset;
                              }
                              byte[] oldRow = Bytes.copy(firstKV.getRowArray(), firstKV.getRowOffset(), firstKV.getRowLength());
                              for (Cell cell : results) {
                                  // Copy existing cell but with new row key
                                  Cell newCell = new KeyValue(newRow, 0, newRow.length,
                                      cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
                                      cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
                                      cell.getTimestamp(), KeyValue.Type.codeToType(cell.getTypeByte()),
                                      cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                                  switch (KeyValue.Type.codeToType(cell.getTypeByte())) {
                                  case Put:
                                      // If Put, point delete old Put
                                      Delete del = new Delete(oldRow);
                                      del.addDeleteMarker(new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
                                          cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
                                          cell.getQualifierArray(), cell.getQualifierOffset(),
                                          cell.getQualifierLength(), cell.getTimestamp(), KeyValue.Type.Delete,
                                          ByteUtil.EMPTY_BYTE_ARRAY, 0, 0));
                                      mutations.add(del);
  
                                      Put put = new Put(newRow);
                                      put.add(newCell);
                                      mutations.add(put);
                                      break;
                                  case Delete:
                                  case DeleteColumn:
                                  case DeleteFamily:
                                  case DeleteFamilyVersion:
                                      Delete delete = new Delete(newRow);
                                      delete.addDeleteMarker(newCell);
                                      mutations.add(delete);
                                      break;
                                  }
                              }
                          } else if (buildLocalIndex) {
                              for (IndexMaintainer maintainer : indexMaintainers) {
                                  if (!results.isEmpty()) {
                                      result.getKey(ptr);
                                      ValueGetter valueGetter =
                                              maintainer.createGetterFromKeyValues(
                                                  ImmutableBytesPtr.copyBytesIfNecessary(ptr),
                                                  results);
                                      Put put = maintainer.buildUpdateMutation(kvBuilder,
                                          valueGetter, ptr, results.get(0).getTimestamp(),
                                          env.getRegion().getRegionInfo().getStartKey(),
                                          env.getRegion().getRegionInfo().getEndKey());
                                      indexMutations.add(put);
                                  }
                              }
                              result.setKeyValues(results);
                          } else if (isDelete) {
                              // FIXME: the version of the Delete constructor without the lock
                              // args was introduced in 0.94.4, thus if we try to use it here
                              // we can no longer use the 0.94.2 version of the client.
                              Cell firstKV = results.get(0);
                              Delete delete = new Delete(firstKV.getRowArray(),
                                  firstKV.getRowOffset(), firstKV.getRowLength(),ts);
                              mutations.add(delete);
                              // force tephra to ignore this deletes
                              delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
                          } else if (isUpsert) {
                              Arrays.fill(values, null);
                              int bucketNumOffset = 0;
                              if (projectedTable.getBucketNum() != null) {
                                  values[0] = new byte[] { 0 };
                                  bucketNumOffset = 1;
                              }
                              int i = bucketNumOffset;
                              List<PColumn> projectedColumns = projectedTable.getColumns();
                              for (; i < projectedTable.getPKColumns().size(); i++) {
                                  Expression expression = selectExpressions.get(i - bucketNumOffset);
                                  if (expression.evaluate(result, ptr)) {
                                      values[i] = ptr.copyBytes();
                                      // If SortOrder from expression in SELECT doesn't match the
                                      // column being projected into then invert the bits.
                                      if (expression.getSortOrder() !=
                                              projectedColumns.get(i).getSortOrder()) {
                                          SortOrder.invert(values[i], 0, values[i], 0,
                                              values[i].length);
                                      }
                                  }else{
                                      values[i] = ByteUtil.EMPTY_BYTE_ARRAY;
                                  }
                              }
                              projectedTable.newKey(ptr, values);
                              PRow row = projectedTable.newRow(kvBuilder, ts, ptr, false);
                              for (; i < projectedColumns.size(); i++) {
                                  Expression expression = selectExpressions.get(i - bucketNumOffset);
                                  if (expression.evaluate(result, ptr)) {
                                      PColumn column = projectedColumns.get(i);
                                      if (!column.getDataType().isSizeCompatible(ptr, null,
                                          expression.getDataType(), expression.getSortOrder(),
                                          expression.getMaxLength(), expression.getScale(),
                                          column.getMaxLength(), column.getScale())) {
                                          throw new DataExceedsCapacityException(
                                              column.getDataType(), column.getMaxLength(),
                                              column.getScale(), column.getName().getString(), ptr);
                                      }
                                      column.getDataType().coerceBytes(ptr, null,
                                          expression.getDataType(), expression.getMaxLength(),
                                          expression.getScale(), expression.getSortOrder(), 
                                          column.getMaxLength(), column.getScale(),
                                          column.getSortOrder(), projectedTable.rowKeyOrderOptimizable());
                                      byte[] bytes = ByteUtil.copyKeyBytesIfNecessary(ptr);
                                      row.setValue(column, bytes);
                                  }
                              }
                              for (Mutation mutation : row.toRowMutations()) {
                                  mutations.add(mutation);
                              }
                              for (i = 0; i < selectExpressions.size(); i++) {
                                  selectExpressions.get(i).reset();
                              }
                          } else if (deleteCF != null && deleteCQ != null) {
                              // No need to search for delete column, since we project only it
                              // if no empty key value is being set
                              if (emptyCF == null ||
                                      result.getValue(deleteCF, deleteCQ) != null) {
                                  Delete delete = new Delete(results.get(0).getRowArray(),
                                      results.get(0).getRowOffset(),
                                      results.get(0).getRowLength());
                                  delete.deleteColumns(deleteCF,  deleteCQ, ts);
                                  // force tephra to ignore this deletes
                                  delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
                                  mutations.add(delete);
                              }
                          }
                          if (emptyCF != null) {
                              /*
                               * If we've specified an emptyCF, then we need to insert an empty
                               * key value "retroactively" for any key value that is visible at
                               * the timestamp that the DDL was issued. Key values that are not
                               * visible at this timestamp will not ever be projected up to
                               * scans past this timestamp, so don't need to be considered.
                               * We insert one empty key value per row per timestamp.
                               */
                              Set<Long> timeStamps =
                                      Sets.newHashSetWithExpectedSize(results.size());
                              for (Cell kv : results) {
                                  long kvts = kv.getTimestamp();
                                  if (!timeStamps.contains(kvts)) {
                                      Put put = new Put(kv.getRowArray(), kv.getRowOffset(),
                                          kv.getRowLength());
                                      put.add(emptyCF, QueryConstants.EMPTY_COLUMN_BYTES, kvts,
                                          ByteUtil.EMPTY_BYTE_ARRAY);
                                      mutations.add(put);
                                  }
                              }
                              // Commit in batches based on UPSERT_BATCH_SIZE_BYTES_ATTRIB in config
                              List<List<Mutation>> batchMutationList =
                                  MutationState.getMutationBatchList(batchSize, batchSizeBytes, mutations);
                              for (List<Mutation> batchMutations : batchMutationList) {
                                  commit(region, batchMutations, indexUUID, blockingMemStoreSize, indexMaintainersPtr,
                                          txState, areMutationInSameRegion, targetHTable);
                                  batchMutations.clear();
                              }
                              mutations.clear();
                              // Commit in batches based on UPSERT_BATCH_SIZE_BYTES_ATTRIB in config
                              List<List<Mutation>> batchIndexMutationList =
                                  MutationState.getMutationBatchList(batchSize, batchSizeBytes, indexMutations);
                              for (List<Mutation> batchIndexMutations : batchIndexMutationList) {
                                  commitBatch(region, batchIndexMutations, null, blockingMemStoreSize, null, txState);
                                  batchIndexMutations.clear();
                              }
                              indexMutations.clear();
                          }
                          aggregators.aggregate(rowAggregators, result);
                          hasAny = true;
                      }
                  } while (hasMore);
                  if (!mutations.isEmpty()) {
                      commit(region, mutations, indexUUID, blockingMemStoreSize, indexMaintainersPtr, txState,
                              areMutationInSameRegion, targetHTable);
                      mutations.clear();
                  }
  
                  if (!indexMutations.isEmpty()) {
                      commitBatch(region, indexMutations, null, blockingMemStoreSize, indexMaintainersPtr, txState);
                      indexMutations.clear();
                  }
              }
          } finally {
              if(needToWrite) {
                  synchronized (lock) {
                      scansReferenceCount--;
                  }
              }
              if (targetHTable != null) {
                  targetHTable.close();
              }
              try {
                  innerScanner.close();
              } finally {
                  if (acquiredLock) region.closeRegionOperation();
              }
          }
          if (logger.isDebugEnabled()) {
              logger.debug(LogUtil.addCustomAnnotations("Finished scanning " + rowCount + " rows for ungrouped coprocessor scan " + scan, ScanUtil.getCustomAnnotations(scan)));
          }
  
          final boolean hadAny = hasAny;
          KeyValue keyValue = null;
          if (hadAny) {
              byte[] value = aggregators.toBytes(rowAggregators);
              keyValue = KeyValueUtil.newKeyValue(UNGROUPED_AGG_ROW_KEY, SINGLE_COLUMN_FAMILY, SINGLE_COLUMN, AGG_TIMESTAMP, value, 0, value.length);
          }
          final KeyValue aggKeyValue = keyValue;
  
          RegionScanner scanner = new BaseRegionScanner(innerScanner) {
              private boolean done = !hadAny;
  
              @Override
              public boolean isFilterDone() {
                  return done;
              }
  
              @Override
              public boolean next(List<Cell> results) throws IOException {
                  if (done) return false;
                  done = true;
                  results.add(aggKeyValue);
                  return false;
              }
  
              @Override
              public long getMaxResultSize() {
                  return scan.getMaxResultSize();
              }
          };
          return scanner;
  
      }
input: #vulnerable code 
                                  @Override
      protected RegionScanner doPostScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws IOException, SQLException {
          RegionCoprocessorEnvironment env = c.getEnvironment();
          Region region = env.getRegion();
          long ts = scan.getTimeRange().getMax();
          boolean localIndexScan = ScanUtil.isLocalIndex(scan);
          if (ScanUtil.isAnalyzeTable(scan)) {
              byte[] gp_width_bytes =
                      scan.getAttribute(BaseScannerRegionObserver.GUIDEPOST_WIDTH_BYTES);
              byte[] gp_per_region_bytes =
                      scan.getAttribute(BaseScannerRegionObserver.GUIDEPOST_PER_REGION);
              // Let this throw, as this scan is being done for the sole purpose of collecting stats
              StatisticsCollector statsCollector = StatisticsCollectorFactory.createStatisticsCollector(
                      env, region.getRegionInfo().getTable().getNameAsString(), ts,
                      gp_width_bytes, gp_per_region_bytes);
              return collectStats(s, statsCollector, region, scan, env.getConfiguration());
          } else if (ScanUtil.isIndexRebuild(scan)) { return rebuildIndices(s, region, scan, env.getConfiguration()); }
          int offsetToBe = 0;
          if (localIndexScan) {
              /*
               * For local indexes, we need to set an offset on row key expressions to skip
               * the region start key.
               */
              offsetToBe = region.getRegionInfo().getStartKey().length != 0 ? region.getRegionInfo().getStartKey().length :
                  region.getRegionInfo().getEndKey().length;
              ScanUtil.setRowKeyOffset(scan, offsetToBe);
          }
          final int offset = offsetToBe;
          
          PTable projectedTable = null;
          PTable writeToTable = null;
          byte[][] values = null;
          byte[] descRowKeyTableBytes = scan.getAttribute(UPGRADE_DESC_ROW_KEY);
          boolean isDescRowKeyOrderUpgrade = descRowKeyTableBytes != null;
          if (isDescRowKeyOrderUpgrade) {
              logger.debug("Upgrading row key for " + region.getRegionInfo().getTable().getNameAsString());
              projectedTable = deserializeTable(descRowKeyTableBytes);
              try {
                  writeToTable = PTableImpl.makePTable(projectedTable, true);
              } catch (SQLException e) {
                  ServerUtil.throwIOException("Upgrade failed", e); // Impossible
              }
              values = new byte[projectedTable.getPKColumns().size()][];
          }
          byte[] localIndexBytes = scan.getAttribute(LOCAL_INDEX_BUILD);
          List<IndexMaintainer> indexMaintainers = localIndexBytes == null ? null : IndexMaintainer.deserialize(localIndexBytes);
          List<Mutation> indexMutations = localIndexBytes == null ? Collections.<Mutation>emptyList() : Lists.<Mutation>newArrayListWithExpectedSize(1024);
          
          RegionScanner theScanner = s;
          
          byte[] indexUUID = scan.getAttribute(PhoenixIndexCodec.INDEX_UUID);
          byte[] txState = scan.getAttribute(BaseScannerRegionObserver.TX_STATE);
          List<Expression> selectExpressions = null;
          byte[] upsertSelectTable = scan.getAttribute(BaseScannerRegionObserver.UPSERT_SELECT_TABLE);
          boolean isUpsert = false;
          boolean isDelete = false;
          byte[] deleteCQ = null;
          byte[] deleteCF = null;
          byte[] emptyCF = null;
          ImmutableBytesWritable ptr = new ImmutableBytesWritable();
          if (upsertSelectTable != null) {
              isUpsert = true;
              projectedTable = deserializeTable(upsertSelectTable);
              selectExpressions = deserializeExpressions(scan.getAttribute(BaseScannerRegionObserver.UPSERT_SELECT_EXPRS));
              values = new byte[projectedTable.getPKColumns().size()][];
              
          } else {
              byte[] isDeleteAgg = scan.getAttribute(BaseScannerRegionObserver.DELETE_AGG);
              isDelete = isDeleteAgg != null && Bytes.compareTo(PDataType.TRUE_BYTES, isDeleteAgg) == 0;
              if (!isDelete) {
                  deleteCF = scan.getAttribute(BaseScannerRegionObserver.DELETE_CF);
                  deleteCQ = scan.getAttribute(BaseScannerRegionObserver.DELETE_CQ);
              }
              emptyCF = scan.getAttribute(BaseScannerRegionObserver.EMPTY_CF);
          }
          TupleProjector tupleProjector = null;
          byte[][] viewConstants = null;
          ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan);
          final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan);
          final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
          if ((localIndexScan && !isDelete && !isDescRowKeyOrderUpgrade) || (j == null && p != null)) {
              if (dataColumns != null) {
                  tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns);
                  viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan);
              }
              ImmutableBytesWritable tempPtr = new ImmutableBytesWritable();
              theScanner =
                      getWrappedScanner(c, theScanner, offset, scan, dataColumns, tupleProjector, 
                          region, indexMaintainers == null ? null : indexMaintainers.get(0), viewConstants, p, tempPtr);
          } 
          
          if (j != null)  {
              theScanner = new HashJoinRegionScanner(theScanner, p, j, ScanUtil.getTenantId(scan), env);
          }
          
          int batchSize = 0;
          long batchSizeBytes = 0L;
          List<Mutation> mutations = Collections.emptyList();
          boolean needToWrite = false;
          Configuration conf = c.getEnvironment().getConfiguration();
          long flushSize = region.getTableDesc().getMemStoreFlushSize();
  
          if (flushSize <= 0) {
              flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
                      HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);
          }
  
          /**
           * Slow down the writes if the memstore size more than
           * (hbase.hregion.memstore.block.multiplier - 1) times hbase.hregion.memstore.flush.size
           * bytes. This avoids flush storm to hdfs for cases like index building where reads and
           * write happen to all the table regions in the server.
           */
          final long blockingMemStoreSize = flushSize * (
                  conf.getLong(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER,
                          HConstants.DEFAULT_HREGION_MEMSTORE_BLOCK_MULTIPLIER)-1) ;
  
          boolean buildLocalIndex = indexMaintainers != null && dataColumns==null && !localIndexScan;
          if (isDescRowKeyOrderUpgrade || isDelete || isUpsert || (deleteCQ != null && deleteCF != null) || emptyCF != null || buildLocalIndex) {
              needToWrite = true;
              // TODO: size better
              mutations = Lists.newArrayListWithExpectedSize(1024);
              batchSize = env.getConfiguration().getInt(MUTATE_BATCH_SIZE_ATTRIB, QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
              batchSizeBytes = env.getConfiguration().getLong(MUTATE_BATCH_SIZE_BYTES_ATTRIB,
                  QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE_BYTES);
          }
          Aggregators aggregators = ServerAggregators.deserialize(
                  scan.getAttribute(BaseScannerRegionObserver.AGGREGATORS), env.getConfiguration());
          Aggregator[] rowAggregators = aggregators.getAggregators();
          boolean hasMore;
          boolean hasAny = false;
          MultiKeyValueTuple result = new MultiKeyValueTuple();
          if (logger.isDebugEnabled()) {
              logger.debug(LogUtil.addCustomAnnotations("Starting ungrouped coprocessor scan " + scan + " "+region.getRegionInfo(), ScanUtil.getCustomAnnotations(scan)));
          }
          long rowCount = 0;
          final RegionScanner innerScanner = theScanner;
          byte[] indexMaintainersPtr = scan.getAttribute(PhoenixIndexCodec.INDEX_MD);
          boolean acquiredLock = false;
          try {
              if(needToWrite) {
                  synchronized (lock) {
                      scansReferenceCount++;
                  }
              }
              region.startRegionOperation();
              acquiredLock = true;
              synchronized (innerScanner) {
                  do {
                      List<Cell> results = new ArrayList<Cell>();
                      // Results are potentially returned even when the return value of s.next is false
                      // since this is an indication of whether or not there are more values after the
                      // ones returned
                      hasMore = innerScanner.nextRaw(results);
                      if (!results.isEmpty()) {
                          rowCount++;
                          result.setKeyValues(results);
                          if (isDescRowKeyOrderUpgrade) {
                              Arrays.fill(values, null);
                              Cell firstKV = results.get(0);
                              RowKeySchema schema = projectedTable.getRowKeySchema();
                              int maxOffset = schema.iterator(firstKV.getRowArray(), firstKV.getRowOffset() + offset, firstKV.getRowLength(), ptr);
                              for (int i = 0; i < schema.getFieldCount(); i++) {
                                  Boolean hasValue = schema.next(ptr, i, maxOffset);
                                  if (hasValue == null) {
                                      break;
                                  }
                                  Field field = schema.getField(i);
                                  if (field.getSortOrder() == SortOrder.DESC) {
                                      // Special case for re-writing DESC ARRAY, as the actual byte value needs to change in this case
                                      if (field.getDataType().isArrayType()) {
                                          field.getDataType().coerceBytes(ptr, null, field.getDataType(),
                                              field.getMaxLength(), field.getScale(), field.getSortOrder(), 
                                              field.getMaxLength(), field.getScale(), field.getSortOrder(), true); // force to use correct separator byte
                                      }
                                      // Special case for re-writing DESC CHAR or DESC BINARY, to force the re-writing of trailing space characters
                                      else if (field.getDataType() == PChar.INSTANCE || field.getDataType() == PBinary.INSTANCE) {
                                          int len = ptr.getLength();
                                          while (len > 0 && ptr.get()[ptr.getOffset() + len - 1] == StringUtil.SPACE_UTF8) {
                                              len--;
                                          }
                                          ptr.set(ptr.get(), ptr.getOffset(), len);
                                          // Special case for re-writing DESC FLOAT and DOUBLE, as they're not inverted like they should be (PHOENIX-2171)
                                      } else if (field.getDataType() == PFloat.INSTANCE || field.getDataType() == PDouble.INSTANCE) {
                                          byte[] invertedBytes = SortOrder.invert(ptr.get(), ptr.getOffset(), ptr.getLength());
                                          ptr.set(invertedBytes);
                                      }
                                  } else if (field.getDataType() == PBinary.INSTANCE) {
                                      // Remove trailing space characters so that the setValues call below will replace them
                                      // with the correct zero byte character. Note this is somewhat dangerous as these
                                      // could be legit, but I don't know what the alternative is.
                                      int len = ptr.getLength();
                                      while (len > 0 && ptr.get()[ptr.getOffset() + len - 1] == StringUtil.SPACE_UTF8) {
                                          len--;
                                      }
                                      ptr.set(ptr.get(), ptr.getOffset(), len);                                        
                                  }
                                  values[i] = ptr.copyBytes();
                              }
                              writeToTable.newKey(ptr, values);
                              if (Bytes.compareTo(
                                  firstKV.getRowArray(), firstKV.getRowOffset() + offset, firstKV.getRowLength(), 
                                  ptr.get(),ptr.getOffset() + offset,ptr.getLength()) == 0) {
                                  continue;
                              }
                              byte[] newRow = ByteUtil.copyKeyBytesIfNecessary(ptr);
                              if (offset > 0) { // for local indexes (prepend region start key)
                                  byte[] newRowWithOffset = new byte[offset + newRow.length];
                                  System.arraycopy(firstKV.getRowArray(), firstKV.getRowOffset(), newRowWithOffset, 0, offset);;
                                  System.arraycopy(newRow, 0, newRowWithOffset, offset, newRow.length);
                                  newRow = newRowWithOffset;
                              }
                              byte[] oldRow = Bytes.copy(firstKV.getRowArray(), firstKV.getRowOffset(), firstKV.getRowLength());
                              for (Cell cell : results) {
                                  // Copy existing cell but with new row key
                                  Cell newCell = new KeyValue(newRow, 0, newRow.length,
                                      cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
                                      cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
                                      cell.getTimestamp(), KeyValue.Type.codeToType(cell.getTypeByte()),
                                      cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                                  switch (KeyValue.Type.codeToType(cell.getTypeByte())) {
                                  case Put:
                                      // If Put, point delete old Put
                                      Delete del = new Delete(oldRow);
                                      del.addDeleteMarker(new KeyValue(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
                                          cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
                                          cell.getQualifierArray(), cell.getQualifierOffset(),
                                          cell.getQualifierLength(), cell.getTimestamp(), KeyValue.Type.Delete,
                                          ByteUtil.EMPTY_BYTE_ARRAY, 0, 0));
                                      mutations.add(del);
  
                                      Put put = new Put(newRow);
                                      put.add(newCell);
                                      mutations.add(put);
                                      break;
                                  case Delete:
                                  case DeleteColumn:
                                  case DeleteFamily:
                                  case DeleteFamilyVersion:
                                      Delete delete = new Delete(newRow);
                                      delete.addDeleteMarker(newCell);
                                      mutations.add(delete);
                                      break;
                                  }
                              }
                          } else if (buildLocalIndex) {
                              for (IndexMaintainer maintainer : indexMaintainers) {
                                  if (!results.isEmpty()) {
                                      result.getKey(ptr);
                                      ValueGetter valueGetter =
                                              maintainer.createGetterFromKeyValues(
                                                  ImmutableBytesPtr.copyBytesIfNecessary(ptr),
                                                  results);
                                      Put put = maintainer.buildUpdateMutation(kvBuilder,
                                          valueGetter, ptr, results.get(0).getTimestamp(),
                                          env.getRegion().getRegionInfo().getStartKey(),
                                          env.getRegion().getRegionInfo().getEndKey());
                                      indexMutations.add(put);
                                  }
                              }
                              result.setKeyValues(results);
                          } else if (isDelete) {
                              // FIXME: the version of the Delete constructor without the lock
                              // args was introduced in 0.94.4, thus if we try to use it here
                              // we can no longer use the 0.94.2 version of the client.
                              Cell firstKV = results.get(0);
                              Delete delete = new Delete(firstKV.getRowArray(),
                                  firstKV.getRowOffset(), firstKV.getRowLength(),ts);
                              mutations.add(delete);
                              // force tephra to ignore this deletes
                              delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
                          } else if (isUpsert) {
                              Arrays.fill(values, null);
                              int i = 0;
                              List<PColumn> projectedColumns = projectedTable.getColumns();
                              for (; i < projectedTable.getPKColumns().size(); i++) {
                                  Expression expression = selectExpressions.get(i);
                                  if (expression.evaluate(result, ptr)) {
                                      values[i] = ptr.copyBytes();
                                      // If SortOrder from expression in SELECT doesn't match the
                                      // column being projected into then invert the bits.
                                      if (expression.getSortOrder() !=
                                              projectedColumns.get(i).getSortOrder()) {
                                          SortOrder.invert(values[i], 0, values[i], 0,
                                              values[i].length);
                                      }
                                  }
                              }
                              projectedTable.newKey(ptr, values);
                              PRow row = projectedTable.newRow(kvBuilder, ts, ptr, false);
                              for (; i < projectedColumns.size(); i++) {
                                  Expression expression = selectExpressions.get(i);
                                  if (expression.evaluate(result, ptr)) {
                                      PColumn column = projectedColumns.get(i);
                                      if (!column.getDataType().isSizeCompatible(ptr, null,
                                          expression.getDataType(), expression.getSortOrder(),
                                          expression.getMaxLength(), expression.getScale(),
                                          column.getMaxLength(), column.getScale())) {
                                          throw new DataExceedsCapacityException(
                                              column.getDataType(), column.getMaxLength(),
                                              column.getScale(), column.getName().getString(), ptr);
                                      }
                                      column.getDataType().coerceBytes(ptr, null,
                                          expression.getDataType(), expression.getMaxLength(),
                                          expression.getScale(), expression.getSortOrder(), 
                                          column.getMaxLength(), column.getScale(),
                                          column.getSortOrder(), projectedTable.rowKeyOrderOptimizable());
                                      byte[] bytes = ByteUtil.copyKeyBytesIfNecessary(ptr);
                                      row.setValue(column, bytes);
                                  }
                              }
                              for (Mutation mutation : row.toRowMutations()) {
                                  mutations.add(mutation);
                              }
                              for (i = 0; i < selectExpressions.size(); i++) {
                                  selectExpressions.get(i).reset();
                              }
                          } else if (deleteCF != null && deleteCQ != null) {
                              // No need to search for delete column, since we project only it
                              // if no empty key value is being set
                              if (emptyCF == null ||
                                      result.getValue(deleteCF, deleteCQ) != null) {
                                  Delete delete = new Delete(results.get(0).getRowArray(),
                                      results.get(0).getRowOffset(),
                                      results.get(0).getRowLength());
                                  delete.deleteColumns(deleteCF,  deleteCQ, ts);
                                  // force tephra to ignore this deletes
                                  delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
                                  mutations.add(delete);
                              }
                          }
                          if (emptyCF != null) {
                              /*
                               * If we've specified an emptyCF, then we need to insert an empty
                               * key value "retroactively" for any key value that is visible at
                               * the timestamp that the DDL was issued. Key values that are not
                               * visible at this timestamp will not ever be projected up to
                               * scans past this timestamp, so don't need to be considered.
                               * We insert one empty key value per row per timestamp.
                               */
                              Set<Long> timeStamps =
                                      Sets.newHashSetWithExpectedSize(results.size());
                              for (Cell kv : results) {
                                  long kvts = kv.getTimestamp();
                                  if (!timeStamps.contains(kvts)) {
                                      Put put = new Put(kv.getRowArray(), kv.getRowOffset(),
                                          kv.getRowLength());
                                      put.add(emptyCF, QueryConstants.EMPTY_COLUMN_BYTES, kvts,
                                          ByteUtil.EMPTY_BYTE_ARRAY);
                                      mutations.add(put);
                                  }
                              }
                              // Commit in batches based on UPSERT_BATCH_SIZE_BYTES_ATTRIB in config
                              List<List<Mutation>> batchMutationList =
                                  MutationState.getMutationBatchList(batchSize, batchSizeBytes, mutations);
                              for (List<Mutation> batchMutations : batchMutationList) {
                                  commitBatch(region, batchMutations, indexUUID, blockingMemStoreSize, indexMaintainersPtr,
                                      txState);
                                  batchMutations.clear();
                              }
                              mutations.clear();
                              // Commit in batches based on UPSERT_BATCH_SIZE_BYTES_ATTRIB in config
                              List<List<Mutation>> batchIndexMutationList =
                                  MutationState.getMutationBatchList(batchSize, batchSizeBytes, indexMutations);
                              for (List<Mutation> batchIndexMutations : batchIndexMutationList) {
                                  commitBatch(region, batchIndexMutations, null, blockingMemStoreSize, null, txState);
                                  batchIndexMutations.clear();
                              }
                              indexMutations.clear();
                          }
                          aggregators.aggregate(rowAggregators, result);
                          hasAny = true;
                      }
                  } while (hasMore);
                  if (!mutations.isEmpty()) {
                      commitBatch(region, mutations, indexUUID, blockingMemStoreSize, indexMaintainersPtr, txState);
                  }
  
                  if (!indexMutations.isEmpty()) {
                      commitBatch(region, indexMutations, null, blockingMemStoreSize, indexMaintainersPtr, txState);
                      indexMutations.clear();
                  }
              }
          } finally {
              if(needToWrite) {
                  synchronized (lock) {
                      scansReferenceCount--;
                  }
              }
              try {
                  innerScanner.close();
              } finally {
                  if (acquiredLock) region.closeRegionOperation();
              }
          }
          if (logger.isDebugEnabled()) {
              logger.debug(LogUtil.addCustomAnnotations("Finished scanning " + rowCount + " rows for ungrouped coprocessor scan " + scan, ScanUtil.getCustomAnnotations(scan)));
          }
  
          final boolean hadAny = hasAny;
          KeyValue keyValue = null;
          if (hadAny) {
              byte[] value = aggregators.toBytes(rowAggregators);
              keyValue = KeyValueUtil.newKeyValue(UNGROUPED_AGG_ROW_KEY, SINGLE_COLUMN_FAMILY, SINGLE_COLUMN, AGG_TIMESTAMP, value, 0, value.length);
          }
          final KeyValue aggKeyValue = keyValue;
  
          RegionScanner scanner = new BaseRegionScanner(innerScanner) {
              private boolean done = !hadAny;
  
              @Override
              public boolean isFilterDone() {
                  return done;
              }
  
              @Override
              public boolean next(List<Cell> results) throws IOException {
                  if (done) return false;
                  done = true;
                  results.add(aggKeyValue);
                  return false;
              }
  
              @Override
              public long getMaxResultSize() {
                  return scan.getMaxResultSize();
              }
          };
          return scanner;
      } 
                              #location 200 
                              #vulnerability type THREAD_SAFETY_VIOLATION