// build map from physical table to mutation list
@SuppressWarnings("deprecation")
    private void send(Iterator<TableRef> tableRefIterator) throws SQLException {
        int i = 0;
        long[] serverTimeStamps = null;
        boolean sendAll = false;
        if (tableRefIterator == null) {
            serverTimeStamps = validateAll();
            tableRefIterator = mutations.keySet().iterator();
            sendAll = true;
        }
        Map<ImmutableBytesPtr, RowMutationState> valuesMap;
        Map<TableInfo,List<Mutation>> physicalTableMutationMap = Maps.newLinkedHashMap(); 
        // add tracing for this operation
        try (TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables")) {
            Span span = trace.getSpan();
            ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable();
            while (tableRefIterator.hasNext()) {
                // at this point we are going through mutations for each table
                final TableRef tableRef = tableRefIterator.next();
                valuesMap = mutations.get(tableRef);
                if (valuesMap == null || valuesMap.isEmpty()) {
                    continue;
                }
                // Validate as we go if transactional since we can undo if a problem occurs (which is unlikely)
                long serverTimestamp = serverTimeStamps == null ? validateAndGetServerTimestamp(tableRef, valuesMap) : serverTimeStamps[i++];
                Long scn = connection.getSCN();
                long mutationTimestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
                final PTable table = tableRef.getTable();
                Iterator<Pair<PName,List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, mutationTimestamp, serverTimestamp, false, sendAll);
                // build map from physical table to mutation list
                boolean isDataTable = true;
                while (mutationsIterator.hasNext()) {
                    Pair<PName,List<Mutation>> pair = mutationsIterator.next();
                    PName hTableName = pair.getFirst();
                    List<Mutation> mutationList = pair.getSecond();
                    TableInfo tableInfo = new TableInfo(isDataTable, hTableName, tableRef);
                    List<Mutation> oldMutationList = physicalTableMutationMap.put(tableInfo, mutationList);
                    if (oldMutationList!=null)
                        mutationList.addAll(0, oldMutationList);
                    isDataTable = false;
                }
                // For transactions, track the statement indexes as we send data
                // over because our CommitException should include all statements
                // involved in the transaction since none of them would have been
                // committed in the event of a failure.
                if (table.isTransactional()) {
                    addUncommittedStatementIndexes(valuesMap.values());
                    if (txMutations.isEmpty()) {
                        txMutations = Maps.newHashMapWithExpectedSize(mutations.size());
                    }
                    // Keep all mutations we've encountered until a commit or rollback.
                    // This is not ideal, but there's not good way to get the values back
                    // in the event that we need to replay the commit.
                    // Copy TableRef so we have the original PTable and know when the
                    // indexes have changed.
                    joinMutationState(new TableRef(tableRef), valuesMap, txMutations);
                }
            }
            long serverTimestamp = HConstants.LATEST_TIMESTAMP;
            Iterator<Entry<TableInfo, List<Mutation>>> mutationsIterator = physicalTableMutationMap.entrySet().iterator();
            while (mutationsIterator.hasNext()) {
                Entry<TableInfo, List<Mutation>> pair = mutationsIterator.next();
                TableInfo tableInfo = pair.getKey();
                byte[] htableName = tableInfo.getHTableName().getBytes();
                List<Mutation> mutationList = pair.getValue();
                //create a span per target table
                //TODO maybe we can be smarter about the table name to string here?
                Span child = Tracing.child(span,"Writing mutation batch for table: "+Bytes.toString(htableName));
                int retryCount = 0;
                boolean shouldRetry = false;
                long numMutations = 0;
                long mutationSizeBytes = 0;
                long mutationCommitTime = 0;
                long numFailedMutations = 0;;
                long startTime = 0;
                do {
                    TableRef origTableRef = tableInfo.getOrigTableRef();
                    PTable table = origTableRef.getTable();
                    table.getIndexMaintainers(indexMetaDataPtr, connection);
                    final ServerCache cache = tableInfo.isDataTable() ? setMetaDataOnMutations(origTableRef, mutationList, indexMetaDataPtr) : null;
                    // If we haven't retried yet, retry for this case only, as it's possible that
                    // a split will occur after we send the index metadata cache to all known
                    // region servers.
                    shouldRetry = cache!=null;
                    SQLException sqlE = null;
                    HTableInterface hTable = connection.getQueryServices().getTable(htableName);
                    try {
                        if (table.isTransactional()) {
                            // Track tables to which we've sent uncommitted data
                            uncommittedPhysicalNames.add(table.getPhysicalName().getString());
                            // If we have indexes, wrap the HTable in a delegate HTable that
                            // will attach the necessary index meta data in the event of a
                            // rollback
                            if (!table.getIndexes().isEmpty()) {
                                hTable = new MetaDataAwareHTable(hTable, origTableRef);
                            }
                            hTable = TransactionUtil.getPhoenixTransactionTable(phoenixTransactionContext, hTable, table);
                        }
                        numMutations = mutationList.size();
                        GLOBAL_MUTATION_BATCH_SIZE.update(numMutations);
                        mutationSizeBytes = calculateMutationSize(mutationList);
                        startTime = System.currentTimeMillis();
                        child.addTimelineAnnotation("Attempt " + retryCount);
                        List<List<Mutation>> mutationBatchList = getMutationBatchList(batchSize, batchSizeBytes, mutationList);
                        for (List<Mutation> mutationBatch : mutationBatchList) {
                            hTable.batch(mutationBatch);
                            batchCount++;
                        }
                        if (logger.isDebugEnabled()) logger.debug("Sent batch of " + numMutations + " for " + Bytes.toString(htableName));
                        child.stop();
                        child.stop();
                        shouldRetry = false;
                        mutationCommitTime = System.currentTimeMillis() - startTime;
                        GLOBAL_MUTATION_COMMIT_TIME.update(mutationCommitTime);
                        numFailedMutations = 0;
                        if (tableInfo.isDataTable()) {
                            numRows -= numMutations;
                        }
                        // Remove batches as we process them
                        mutations.remove(origTableRef);
                    } catch (Exception e) {
                    	mutationCommitTime = System.currentTimeMillis() - startTime;
                        serverTimestamp = ServerUtil.parseServerTimestamp(e);
                        SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
                        if (inferredE != null) {
                            if (shouldRetry && retryCount == 0 && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND.getErrorCode()) {
                                // Swallow this exception once, as it's possible that we split after sending the index metadata
                                // and one of the region servers doesn't have it. This will cause it to have it the next go around.
                                // If it fails again, we don't retry.
                                String msg = "Swallowing exception and retrying after clearing meta cache on connection. " + inferredE;
                                logger.warn(LogUtil.addCustomAnnotations(msg, connection));
                                connection.getQueryServices().clearTableRegionCache(htableName);
                                // add a new child span as this one failed
                                child.addTimelineAnnotation(msg);
                                child.stop();
                                child = Tracing.child(span,"Failed batch, attempting retry");
                                continue;
                            }
                            e = inferredE;
                        }
                        // Throw to client an exception that indicates the statements that
                        // were not committed successfully.
                        int[] uncommittedStatementIndexes = getUncommittedStatementIndexes();
						sqlE = new CommitException(e, uncommittedStatementIndexes, serverTimestamp);
						numFailedMutations = uncommittedStatementIndexes.length;
						GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations);
                    } finally {
                    	MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
                        mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
                        try {
                            if (cache!=null) 
                                cache.close();
                        } finally {
                            try {
                                hTable.close();
                            } 
                            catch (IOException e) {
                                if (sqlE != null) {
                                    sqlE.setNextException(ServerUtil.parseServerException(e));
                                } else {
                                    sqlE = ServerUtil.parseServerException(e);
                                }
                            } 
                            if (sqlE != null) {
                                throw sqlE;
                            }
                        }
                    }
                } while (shouldRetry && retryCount++ < 1);
            }
        }
    }
final PTable table = tableRef.getTable();
Iterator<Pair<PName,List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, mutationTimestamp, serverTimestamp, false, sendAll);
// build map from physical table to mutation list
boolean isDataTable = true;
while (mutationsIterator.hasNext()) {
valuesMap = mutations.get(tableRef);
if (valuesMap == null || valuesMap.isEmpty()) {
continue;
}
// Validate as we go if transactional since we can undo if a problem occurs (which is unlikely)
long serverTimestamp = serverTimeStamps == null ? validateAndGetServerTimestamp(tableRef, valuesMap) : serverTimeStamps[i++];
Long scn = connection.getSCN();
long mutationTimestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
final PTable table = tableRef.getTable();
Iterator<Pair<PName,List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, mutationTimestamp, serverTimestamp, false, sendAll);
// build map from physical table to mutation list
boolean isDataTable = true;
while (mutationsIterator.hasNext()) {
Pair<PName,List<Mutation>> pair = mutationsIterator.next();
PName hTableName = pair.getFirst();
List<Mutation> mutationList = pair.getSecond();
TableInfo tableInfo = new TableInfo(isDataTable, hTableName, tableRef);
List<Mutation> oldMutationList = physicalTableMutationMap.put(tableInfo, mutationList);
if (oldMutationList!=null)
mutationList.addAll(0, oldMutationList);
isDataTable = false;
}
Map<ImmutableBytesPtr, RowMutationState> valuesMap;
Map<TableInfo,List<Mutation>> physicalTableMutationMap = Maps.newLinkedHashMap();
// add tracing for this operation
try (TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables")) {
Span span = trace.getSpan();
ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable();
while (tableRefIterator.hasNext()) {
// at this point we are going through mutations for each table
final TableRef tableRef = tableRefIterator.next();
valuesMap = mutations.get(tableRef);
if (valuesMap == null || valuesMap.isEmpty()) {
continue;
}
// Validate as we go if transactional since we can undo if a problem occurs (which is unlikely)
long serverTimestamp = serverTimeStamps == null ? validateAndGetServerTimestamp(tableRef, valuesMap) : serverTimeStamps[i++];
Long scn = connection.getSCN();
long mutationTimestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
final PTable table = tableRef.getTable();
Iterator<Pair<PName,List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, mutationTimestamp, serverTimestamp, false, sendAll);
// build map from physical table to mutation list
boolean isDataTable = true;
while (mutationsIterator.hasNext()) {
Pair<PName,List<Mutation>> pair = mutationsIterator.next();
PName hTableName = pair.getFirst();
List<Mutation> mutationList = pair.getSecond();
TableInfo tableInfo = new TableInfo(isDataTable, hTableName, tableRef);
List<Mutation> oldMutationList = physicalTableMutationMap.put(tableInfo, mutationList);
if (oldMutationList!=null)
mutationList.addAll(0, oldMutationList);
isDataTable = false;
}
// For transactions, track the statement indexes as we send data
// over because our CommitException should include all statements
// involved in the transaction since none of them would have been
// committed in the event of a failure.
if (table.isTransactional()) {
addUncommittedStatementIndexes(valuesMap.values());
if (txMutations.isEmpty()) {
txMutations = Maps.newHashMapWithExpectedSize(mutations.size());
}