#vulnerable code 
                                @Override
    protected RegionScanner doPostScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws IOException {
        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());
        }
        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);
        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;
        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);
        }

        /**
         * Upper bound of memstore size allowed for region. Updates will be blocked until the flush
         * happen if the memstore reaches this threshold.
         */
        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);
        }
        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;
        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);
                        try {
                            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);
                                for (; i < projectedColumns.size(); i++) {
                                    Expression expression = selectExpressions.get(i);
                                    if (expression.evaluate(result, ptr)) {
                                        PColumn column = projectedColumns.get(i);
                                        Object value = expression.getDataType()
                                                .toObject(ptr, column.getSortOrder());
                                        // We are guaranteed that the two column will have the
                                        // same type.
                                        if (!column.getDataType().isSizeCompatible(ptr, value,
                                            column.getDataType(), expression.getMaxLength(),
                                            expression.getScale(), column.getMaxLength(),
                                            column.getScale())) {
                                            throw new DataExceedsCapacityException(
                                                column.getDataType(), column.getMaxLength(),
                                                column.getScale());
                                        }
                                        column.getDataType().coerceBytes(ptr, value,
                                            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_ATTRIB in config
                            if (!mutations.isEmpty() && batchSize > 0 &&
                                    mutations.size() % batchSize == 0) {
                                commitBatch(region, mutations, indexUUID, blockingMemStoreSize);
                                mutations.clear();
                            }
                            // Commit in batches based on UPSERT_BATCH_SIZE_ATTRIB in config
                            if (!indexMutations.isEmpty() && batchSize > 0 &&
                                    indexMutations.size() % batchSize == 0) {
                                commitBatch(region, indexMutations, null, blockingMemStoreSize);
                                indexMutations.clear();
                            }
                        } catch (ConstraintViolationException e) {
                            // Log and ignore in count
                            logger.error(LogUtil.addCustomAnnotations("Failed to create row in " +
                                    region.getRegionInfo().getRegionNameAsString() + " with values " +
                                    SchemaUtil.toString(values),
                                    ScanUtil.getCustomAnnotations(scan)), e);
                            continue;
                        }
                        aggregators.aggregate(rowAggregators, result);
                        hasAny = true;
                    }
                } while (hasMore);
                if (!mutations.isEmpty()) {
                    commitBatch(region,mutations, indexUUID, blockingMemStoreSize);
                }

                if (!indexMutations.isEmpty()) {
                    commitBatch(region,indexMutations, null, blockingMemStoreSize);
                    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 404 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                                @Override
    protected RegionScanner doPostScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Scan scan, final RegionScanner s) throws IOException {
        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());
        }
        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);
        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;
        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);
        }
        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;
        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);
                        try {
                            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);
                                for (; i < projectedColumns.size(); i++) {
                                    Expression expression = selectExpressions.get(i);
                                    if (expression.evaluate(result, ptr)) {
                                        PColumn column = projectedColumns.get(i);
                                        Object value = expression.getDataType()
                                                .toObject(ptr, column.getSortOrder());
                                        // We are guaranteed that the two column will have the
                                        // same type.
                                        if (!column.getDataType().isSizeCompatible(ptr, value,
                                            column.getDataType(), expression.getMaxLength(),
                                            expression.getScale(), column.getMaxLength(),
                                            column.getScale())) {
                                            throw new DataExceedsCapacityException(
                                                column.getDataType(), column.getMaxLength(),
                                                column.getScale());
                                        }
                                        column.getDataType().coerceBytes(ptr, value,
                                            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_ATTRIB in config
                            if (!mutations.isEmpty() && batchSize > 0 &&
                                    mutations.size() % batchSize == 0) {
                                commitBatch(region, mutations, indexUUID, blockingMemStoreSize);
                                mutations.clear();
                            }
                            // Commit in batches based on UPSERT_BATCH_SIZE_ATTRIB in config
                            if (!indexMutations.isEmpty() && batchSize > 0 &&
                                    indexMutations.size() % batchSize == 0) {
                                commitBatch(region, indexMutations, null, blockingMemStoreSize);
                                indexMutations.clear();
                            }
                        } catch (ConstraintViolationException e) {
                            // Log and ignore in count
                            logger.error(LogUtil.addCustomAnnotations("Failed to create row in " +
                                    region.getRegionInfo().getRegionNameAsString() + " with values " +
                                    SchemaUtil.toString(values),
                                    ScanUtil.getCustomAnnotations(scan)), e);
                            continue;
                        }
                        aggregators.aggregate(rowAggregators, result);
                        hasAny = true;
                    }
                } while (hasMore);
                if (!mutations.isEmpty()) {
                    commitBatch(region,mutations, indexUUID, blockingMemStoreSize);
                }

                if (!indexMutations.isEmpty()) {
                    commitBatch(region,indexMutations, null, blockingMemStoreSize);
                    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;
    }