Skip to content

Commit

Permalink
PHOENIX-4381 Calculate the estimatedSize of MutationState incrementally
Browse files Browse the repository at this point in the history
  • Loading branch information
@twdsilva
twdsilva committed Nov 16, 2017
1 parent 693fa65 commit 6a45c40
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 37 deletions.
15 changes: 13 additions & 2 deletions phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
View file
Expand Up @@ -128,6 +128,7 @@ public class MutationState implements SQLCloseable {

private long sizeOffset;
private int numRows = 0;
private long estimatedSize = 0;
private int[] uncommittedStatementIndexes = EMPTY_STATEMENT_INDEX_ARRAY;
private boolean isExternalTxContext = false;
private Map<TableRef, Map<ImmutableBytesPtr,RowMutationState>> txMutations = Collections.emptyMap();
Expand Down Expand Up @@ -194,6 +195,7 @@ public MutationState(TableRef table, Map<ImmutableBytesPtr,RowMutationState> mut
this.mutations.put(table, mutations);
}
this.numRows = mutations.size();
this.estimatedSize = KeyValueUtil.getEstimatedRowSize(table, mutations);
throwIfTooBig();
}

Expand Down Expand Up @@ -355,7 +357,6 @@ private void throwIfTooBig() throws SQLException {
throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_EXCEEDED).build()
.buildException();
}
long estimatedSize = PhoenixKeyValueUtil.getEstimatedRowSize(mutations);
if (estimatedSize > maxSizeBytes) {
resetState();
throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_BYTES_EXCEEDED)
Expand Down Expand Up @@ -434,7 +435,12 @@ public void join(MutationState newMutationState) throws SQLException {
phoenixTransactionContext.join(newMutationState.getPhoenixTransactionContext());

this.sizeOffset += newMutationState.sizeOffset;
int oldNumRows = this.numRows;
joinMutationState(newMutationState.mutations, this.mutations);
// here we increment the estimated size by the fraction of new rows we added from the newMutationState
if (newMutationState.numRows>0) {
this.estimatedSize += ((double)(this.numRows-oldNumRows)/newMutationState.numRows) * newMutationState.estimatedSize;
}
if (!newMutationState.txMutations.isEmpty()) {
if (txMutations.isEmpty()) {
txMutations = Maps.newHashMapWithExpectedSize(mutations.size());
Expand Down Expand Up @@ -968,6 +974,8 @@ private void send(Iterator<TableRef> tableRefIterator) throws SQLException {
long mutationCommitTime = 0;
long numFailedMutations = 0;;
long startTime = 0;
long startNumRows = numRows;
long startEstimatedSize = estimatedSize;
do {
TableRef origTableRef = tableInfo.getOrigTableRef();
PTable table = origTableRef.getTable();
Expand Down Expand Up @@ -1005,8 +1013,8 @@ private void send(Iterator<TableRef> tableRefIterator) throws SQLException {
// TODO need to get the the results of batch and fail if any exceptions.
hTable.batch(mutationBatch, null);
batchCount++;
if (logger.isDebugEnabled()) logger.debug("Sent batch of " + mutationBatch.size() + " for " + Bytes.toString(htableName));
}
if (logger.isDebugEnabled()) logger.debug("Sent batch of " + numMutations + " for " + Bytes.toString(htableName));
child.stop();
child.stop();
shouldRetry = false;
Expand All @@ -1016,6 +1024,8 @@ private void send(Iterator<TableRef> tableRefIterator) throws SQLException {

if (tableInfo.isDataTable()) {
numRows -= numMutations;
// decrement estimated size by the fraction of rows we sent to hbase
estimatedSize -= ((double)numMutations/startNumRows)*startEstimatedSize;
}
// Remove batches as we process them
mutations.remove(origTableRef);
Expand Down Expand Up @@ -1181,6 +1191,7 @@ public void close() throws SQLException {

private void resetState() {
numRows = 0;
estimatedSize = 0;
this.mutations.clear();
resetTransactionalState();
}
Expand Down
65 changes: 30 additions & 35 deletions phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixKeyValueUtil.java
View file
Expand Up @@ -189,46 +189,41 @@ public static long calculateMutationDiskSize(Mutation m) {
* @return estimated row size
*/
public static long
getEstimatedRowSize(Map<TableRef, Map<ImmutableBytesPtr, RowMutationState>> mutations) {
getEstimatedRowSize(TableRef tableRef, Map<ImmutableBytesPtr, RowMutationState> mutations) {
long size = 0;
// iterate over tables
for (Entry<TableRef, Map<ImmutableBytesPtr, RowMutationState>> tableEntry : mutations
.entrySet()) {
PTable table = tableEntry.getKey().getTable();
// iterate over rows
for (Entry<ImmutableBytesPtr, RowMutationState> rowEntry : tableEntry.getValue()
.entrySet()) {
int rowLength = rowEntry.getKey().getLength();
Map<PColumn, byte[]> colValueMap = rowEntry.getValue().getColumnValues();
switch (table.getImmutableStorageScheme()) {
case ONE_CELL_PER_COLUMN:
// iterate over columns
for (Entry<PColumn, byte[]> colValueEntry : colValueMap.entrySet()) {
PColumn pColumn = colValueEntry.getKey();
size +=
KeyValue.getKeyValueDataStructureSize(rowLength,
pColumn.getFamilyName().getBytes().length,
pColumn.getColumnQualifierBytes().length,
colValueEntry.getValue().length);
}
break;
case SINGLE_CELL_ARRAY_WITH_OFFSETS:
// we store all the column values in a single key value that contains all the
// column values followed by an offset array
PTable table = tableRef.getTable();
// iterate over rows
for (Entry<ImmutableBytesPtr, RowMutationState> rowEntry : mutations.entrySet()) {
int rowLength = rowEntry.getKey().getLength();
Map<PColumn, byte[]> colValueMap = rowEntry.getValue().getColumnValues();
switch (table.getImmutableStorageScheme()) {
case ONE_CELL_PER_COLUMN:
// iterate over columns
for (Entry<PColumn, byte[]> colValueEntry : colValueMap.entrySet()) {
PColumn pColumn = colValueEntry.getKey();
size +=
PArrayDataTypeEncoder.getEstimatedByteSize(table, rowLength,
colValueMap);
break;
KeyValue.getKeyValueDataStructureSize(rowLength,
pColumn.getFamilyName().getBytes().length,
pColumn.getColumnQualifierBytes().length,
colValueEntry.getValue().length);
}
// count the empty key value
Pair<byte[], byte[]> emptyKeyValueInfo =
EncodedColumnsUtil.getEmptyKeyValueInfo(table);
break;
case SINGLE_CELL_ARRAY_WITH_OFFSETS:
// we store all the column values in a single key value that contains all the
// column values followed by an offset array
size +=
KeyValue.getKeyValueDataStructureSize(rowLength,
SchemaUtil.getEmptyColumnFamilyPtr(table).getLength(),
emptyKeyValueInfo.getFirst().length,
emptyKeyValueInfo.getSecond().length);
PArrayDataTypeEncoder.getEstimatedByteSize(table, rowLength,
colValueMap);
break;
}
// count the empty key value
Pair<byte[], byte[]> emptyKeyValueInfo =
EncodedColumnsUtil.getEmptyKeyValueInfo(table);
size +=
KeyValue.getKeyValueDataStructureSize(rowLength,
SchemaUtil.getEmptyColumnFamilyPtr(table).getLength(),
emptyKeyValueInfo.getFirst().length,
emptyKeyValueInfo.getSecond().length);
}
return size;
}
Expand Down

0 comments on commit 6a45c40

Please sign in to comment.