@Test public void testGetMutationBatchList() { byte[] r1 = Bytes.toBytes(1); byte[] r2 = Bytes.toBytes(2); byte[] r3 = Bytes.toBytes(3); byte[] r4 = Bytes.toBytes(4); { List<Mutation> list = ImmutableList.of(new Put(r1), new Put(r2), new Delete(r2)); List<List<Mutation>> batchLists = MutationState.getMutationBatchList(2, 10, list); assertTrue(batchLists.size() == 2); assertEquals(batchLists.get(0).size(), 1); assertEquals(batchLists.get(1).size(), 2); } { List<Mutation> list = ImmutableList.of(new Put(r1), new Delete(r1), new Put(r2)); List<List<Mutation>> batchLists = MutationState.getMutationBatchList(2, 10, list); assertTrue(batchLists.size() == 2); assertEquals(batchLists.get(0).size(), 2); assertEquals(batchLists.get(1).size(), 1); } { List<Mutation> list = ImmutableList.of(new Put(r3), new Put(r1), new Delete(r1), new Put(r2), new Put(r4), new Delete(r4)); List<List<Mutation>> batchLists = MutationState.getMutationBatchList(2, 10, list); assertTrue(batchLists.size() == 4); assertEquals(batchLists.get(0).size(), 1); assertEquals(batchLists.get(1).size(), 2); assertEquals(batchLists.get(2).size(), 1); assertEquals(batchLists.get(3).size(), 2); } }
public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { Preconditions.checkArgument(batchSize> 1, "Mutation types are put or delete, for one row all mutations must be in one batch."); Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); List<List<Mutation>> mutationBatchList = Lists.newArrayList(); List<Mutation> currentList = Lists.newArrayList(); List<Mutation> sameRowList = Lists.newArrayList(); long currentBatchSizeBytes = 0L; for (int i = 0; i < allMutationList.size(); ) { long sameRowBatchSize = 1L; Mutation mutation = allMutationList.get(i); long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); sameRowList.add(mutation); while (i + 1 < allMutationList.size() && Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { Mutation sameRowMutation = allMutationList.get(i + 1); sameRowList.add(sameRowMutation); sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); sameRowBatchSize++; i++; } if (currentList.size() + sameRowBatchSize > batchSize || currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { if (currentList.size() > 0) { mutationBatchList.add(currentList); currentList = Lists.newArrayList(); currentBatchSizeBytes = 0L; } } currentList.addAll(sameRowList); currentBatchSizeBytes += sameRowMutationSizeBytes; sameRowList.clear(); i++; } if (currentList.size() > 0) { mutationBatchList.add(currentList); } return mutationBatchList; }
MutationState implements SQLCloseable { public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { Preconditions.checkArgument(batchSize> 1, "Mutation types are put or delete, for one row all mutations must be in one batch."); Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); List<List<Mutation>> mutationBatchList = Lists.newArrayList(); List<Mutation> currentList = Lists.newArrayList(); List<Mutation> sameRowList = Lists.newArrayList(); long currentBatchSizeBytes = 0L; for (int i = 0; i < allMutationList.size(); ) { long sameRowBatchSize = 1L; Mutation mutation = allMutationList.get(i); long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); sameRowList.add(mutation); while (i + 1 < allMutationList.size() && Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { Mutation sameRowMutation = allMutationList.get(i + 1); sameRowList.add(sameRowMutation); sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); sameRowBatchSize++; i++; } if (currentList.size() + sameRowBatchSize > batchSize || currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { if (currentList.size() > 0) { mutationBatchList.add(currentList); currentList = Lists.newArrayList(); currentBatchSizeBytes = 0L; } } currentList.addAll(sameRowList); currentBatchSizeBytes += sameRowMutationSizeBytes; sameRowList.clear(); i++; } if (currentList.size() > 0) { mutationBatchList.add(currentList); } return mutationBatchList; } }
MutationState implements SQLCloseable { public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { Preconditions.checkArgument(batchSize> 1, "Mutation types are put or delete, for one row all mutations must be in one batch."); Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); List<List<Mutation>> mutationBatchList = Lists.newArrayList(); List<Mutation> currentList = Lists.newArrayList(); List<Mutation> sameRowList = Lists.newArrayList(); long currentBatchSizeBytes = 0L; for (int i = 0; i < allMutationList.size(); ) { long sameRowBatchSize = 1L; Mutation mutation = allMutationList.get(i); long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); sameRowList.add(mutation); while (i + 1 < allMutationList.size() && Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { Mutation sameRowMutation = allMutationList.get(i + 1); sameRowList.add(sameRowMutation); sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); sameRowBatchSize++; i++; } if (currentList.size() + sameRowBatchSize > batchSize || currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { if (currentList.size() > 0) { mutationBatchList.add(currentList); currentList = Lists.newArrayList(); currentBatchSizeBytes = 0L; } } currentList.addAll(sameRowList); currentBatchSizeBytes += sameRowMutationSizeBytes; sameRowList.clear(); i++; } if (currentList.size() > 0) { mutationBatchList.add(currentList); } return mutationBatchList; } MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           PhoenixTransactionContext txContext);  MutationState(MutationState mutationState);  MutationState(MutationState mutationState, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           long sizeOffset); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext, long sizeOffset);   MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            Map<TableRef, MultiRowMutationState> mutations, boolean subTask, PhoenixTransactionContext txContext);  MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset,
           int maxSize, long maxSizeBytes, PhoenixConnection connection); }
MutationState implements SQLCloseable { public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { Preconditions.checkArgument(batchSize> 1, "Mutation types are put or delete, for one row all mutations must be in one batch."); Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); List<List<Mutation>> mutationBatchList = Lists.newArrayList(); List<Mutation> currentList = Lists.newArrayList(); List<Mutation> sameRowList = Lists.newArrayList(); long currentBatchSizeBytes = 0L; for (int i = 0; i < allMutationList.size(); ) { long sameRowBatchSize = 1L; Mutation mutation = allMutationList.get(i); long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); sameRowList.add(mutation); while (i + 1 < allMutationList.size() && Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { Mutation sameRowMutation = allMutationList.get(i + 1); sameRowList.add(sameRowMutation); sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); sameRowBatchSize++; i++; } if (currentList.size() + sameRowBatchSize > batchSize || currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { if (currentList.size() > 0) { mutationBatchList.add(currentList); currentList = Lists.newArrayList(); currentBatchSizeBytes = 0L; } } currentList.addAll(sameRowList); currentBatchSizeBytes += sameRowMutationSizeBytes; sameRowList.clear(); i++; } if (currentList.size() > 0) { mutationBatchList.add(currentList); } return mutationBatchList; } MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           PhoenixTransactionContext txContext);  MutationState(MutationState mutationState);  MutationState(MutationState mutationState, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           long sizeOffset); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext, long sizeOffset);   MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            Map<TableRef, MultiRowMutationState> mutations, boolean subTask, PhoenixTransactionContext txContext);  MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset,
           int maxSize, long maxSizeBytes, PhoenixConnection connection); long getEstimatedSize(); int getMaxSize(); long getMaxSizeBytes(); PhoenixTransactionContext getPhoenixTransactionContext(); void commitDDLFence(PTable dataTable); boolean checkpointIfNeccessary(MutationPlan plan); Table getHTable(PTable table); PhoenixConnection getConnection(); boolean isTransactionStarted(); long getInitialWritePointer(); long getWritePointer(); PhoenixVisibilityLevel getVisibilityLevel(); boolean startTransaction(Provider provider); static MutationState emptyMutationState(int maxSize, long maxSizeBytes,
                  PhoenixConnection connection); long getUpdateCount(); int getNumRows(); void join(MutationState newMutationState); Iterator<Pair<byte[], List<Mutation>>> toMutations(Long timestamp); Iterator<Pair<byte[], List<Mutation>>> toMutations(); Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes); Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes,
            final Long tableTimestamp); static long getTableTimestamp(final Long tableTimestamp, Long scn); static long getMutationTimestamp(final Long scn); long getBatchSizeBytes(); long getBatchCount(); static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList); byte[] encodeTransaction(); @Override void close(); void rollback(); void commit(); boolean sendUncommitted(); boolean sendUncommitted(Iterator<TableRef> tableRefs); void send(); static int[] joinSortedIntArrays(int[] a, int[] b); ReadMetricQueue getReadMetricQueue(); void setReadMetricQueue(ReadMetricQueue readMetricQueue); MutationMetricQueue getMutationMetricQueue(); }
MutationState implements SQLCloseable { public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { Preconditions.checkArgument(batchSize> 1, "Mutation types are put or delete, for one row all mutations must be in one batch."); Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); List<List<Mutation>> mutationBatchList = Lists.newArrayList(); List<Mutation> currentList = Lists.newArrayList(); List<Mutation> sameRowList = Lists.newArrayList(); long currentBatchSizeBytes = 0L; for (int i = 0; i < allMutationList.size(); ) { long sameRowBatchSize = 1L; Mutation mutation = allMutationList.get(i); long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); sameRowList.add(mutation); while (i + 1 < allMutationList.size() && Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { Mutation sameRowMutation = allMutationList.get(i + 1); sameRowList.add(sameRowMutation); sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); sameRowBatchSize++; i++; } if (currentList.size() + sameRowBatchSize > batchSize || currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { if (currentList.size() > 0) { mutationBatchList.add(currentList); currentList = Lists.newArrayList(); currentBatchSizeBytes = 0L; } } currentList.addAll(sameRowList); currentBatchSizeBytes += sameRowMutationSizeBytes; sameRowList.clear(); i++; } if (currentList.size() > 0) { mutationBatchList.add(currentList); } return mutationBatchList; } MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           PhoenixTransactionContext txContext);  MutationState(MutationState mutationState);  MutationState(MutationState mutationState, PhoenixConnection connection);  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
           long sizeOffset); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext); private  MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            boolean subTask, PhoenixTransactionContext txContext, long sizeOffset);   MutationState(int maxSize, long maxSizeBytes, PhoenixConnection connection,
            Map<TableRef, MultiRowMutationState> mutations, boolean subTask, PhoenixTransactionContext txContext);  MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset,
           int maxSize, long maxSizeBytes, PhoenixConnection connection); long getEstimatedSize(); int getMaxSize(); long getMaxSizeBytes(); PhoenixTransactionContext getPhoenixTransactionContext(); void commitDDLFence(PTable dataTable); boolean checkpointIfNeccessary(MutationPlan plan); Table getHTable(PTable table); PhoenixConnection getConnection(); boolean isTransactionStarted(); long getInitialWritePointer(); long getWritePointer(); PhoenixVisibilityLevel getVisibilityLevel(); boolean startTransaction(Provider provider); static MutationState emptyMutationState(int maxSize, long maxSizeBytes,
                  PhoenixConnection connection); long getUpdateCount(); int getNumRows(); void join(MutationState newMutationState); Iterator<Pair<byte[], List<Mutation>>> toMutations(Long timestamp); Iterator<Pair<byte[], List<Mutation>>> toMutations(); Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes); Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes,
            final Long tableTimestamp); static long getTableTimestamp(final Long tableTimestamp, Long scn); static long getMutationTimestamp(final Long scn); long getBatchSizeBytes(); long getBatchCount(); static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList); byte[] encodeTransaction(); @Override void close(); void rollback(); void commit(); boolean sendUncommitted(); boolean sendUncommitted(Iterator<TableRef> tableRefs); void send(); static int[] joinSortedIntArrays(int[] a, int[] b); ReadMetricQueue getReadMetricQueue(); void setReadMetricQueue(ReadMetricQueue readMetricQueue); MutationMetricQueue getMutationMetricQueue();  }