@Test public void respondsToChangesInSingleQueryLimitBatchParameter() { AtomicReference<CassandraCellLoadingConfig> config = new AtomicReference<>(LOADING_CONFIG); CellLoadingBatcher reloadingBatcher = new CellLoadingBatcher(config::get, rebatchingCallback); updateConfig(config, CROSS_COLUMN_LIMIT, 2 * SINGLE_QUERY_LIMIT); List<List<Cell>> largerBatches = reloadingBatcher.partitionIntoBatches( rowRange(2 * SINGLE_QUERY_LIMIT, 0), ADDRESS, TABLE_REFERENCE); assertBatchContentsMatch(largerBatches, rowRange(2 * SINGLE_QUERY_LIMIT, 0)); verify(rebatchingCallback, never()).consume(any(), any(), anyInt()); }
List<List<Cell>> partitionIntoBatches( Collection<Cell> cellsToPartition, InetSocketAddress host, TableReference tableReference) { CassandraCellLoadingConfig config = loadingConfigSupplier.get(); ListMultimap<byte[], Cell> cellsByColumn = indexCellsByColumnName(cellsToPartition); List<List<Cell>> batches = Lists.newArrayList(); List<Cell> cellsForCrossColumnBatching = Lists.newArrayList(); for (Map.Entry<byte[], List<Cell>> cellColumnPair : Multimaps.asMap(cellsByColumn).entrySet()) { if (shouldExplicitlyAllocateBatchToColumn(config, cellColumnPair.getValue())) { batches.addAll(partitionBySingleQueryLoadBatchLimit( cellColumnPair.getValue(), config, host, tableReference)); } else { cellsForCrossColumnBatching.addAll(cellColumnPair.getValue()); } } batches.addAll(Lists.partition(cellsForCrossColumnBatching, config.crossColumnLoadBatchLimit())); return batches; }
CellLoadingBatcher { List<List<Cell>> partitionIntoBatches( Collection<Cell> cellsToPartition, InetSocketAddress host, TableReference tableReference) { CassandraCellLoadingConfig config = loadingConfigSupplier.get(); ListMultimap<byte[], Cell> cellsByColumn = indexCellsByColumnName(cellsToPartition); List<List<Cell>> batches = Lists.newArrayList(); List<Cell> cellsForCrossColumnBatching = Lists.newArrayList(); for (Map.Entry<byte[], List<Cell>> cellColumnPair : Multimaps.asMap(cellsByColumn).entrySet()) { if (shouldExplicitlyAllocateBatchToColumn(config, cellColumnPair.getValue())) { batches.addAll(partitionBySingleQueryLoadBatchLimit( cellColumnPair.getValue(), config, host, tableReference)); } else { cellsForCrossColumnBatching.addAll(cellColumnPair.getValue()); } } batches.addAll(Lists.partition(cellsForCrossColumnBatching, config.crossColumnLoadBatchLimit())); return batches; } }
CellLoadingBatcher { List<List<Cell>> partitionIntoBatches( Collection<Cell> cellsToPartition, InetSocketAddress host, TableReference tableReference) { CassandraCellLoadingConfig config = loadingConfigSupplier.get(); ListMultimap<byte[], Cell> cellsByColumn = indexCellsByColumnName(cellsToPartition); List<List<Cell>> batches = Lists.newArrayList(); List<Cell> cellsForCrossColumnBatching = Lists.newArrayList(); for (Map.Entry<byte[], List<Cell>> cellColumnPair : Multimaps.asMap(cellsByColumn).entrySet()) { if (shouldExplicitlyAllocateBatchToColumn(config, cellColumnPair.getValue())) { batches.addAll(partitionBySingleQueryLoadBatchLimit( cellColumnPair.getValue(), config, host, tableReference)); } else { cellsForCrossColumnBatching.addAll(cellColumnPair.getValue()); } } batches.addAll(Lists.partition(cellsForCrossColumnBatching, config.crossColumnLoadBatchLimit())); return batches; } CellLoadingBatcher(Supplier<CassandraCellLoadingConfig> loadingConfigSupplier,
            BatchCallback rebatchingManyRowsForColumnCallback); }
CellLoadingBatcher { List<List<Cell>> partitionIntoBatches( Collection<Cell> cellsToPartition, InetSocketAddress host, TableReference tableReference) { CassandraCellLoadingConfig config = loadingConfigSupplier.get(); ListMultimap<byte[], Cell> cellsByColumn = indexCellsByColumnName(cellsToPartition); List<List<Cell>> batches = Lists.newArrayList(); List<Cell> cellsForCrossColumnBatching = Lists.newArrayList(); for (Map.Entry<byte[], List<Cell>> cellColumnPair : Multimaps.asMap(cellsByColumn).entrySet()) { if (shouldExplicitlyAllocateBatchToColumn(config, cellColumnPair.getValue())) { batches.addAll(partitionBySingleQueryLoadBatchLimit( cellColumnPair.getValue(), config, host, tableReference)); } else { cellsForCrossColumnBatching.addAll(cellColumnPair.getValue()); } } batches.addAll(Lists.partition(cellsForCrossColumnBatching, config.crossColumnLoadBatchLimit())); return batches; } CellLoadingBatcher(Supplier<CassandraCellLoadingConfig> loadingConfigSupplier,
            BatchCallback rebatchingManyRowsForColumnCallback);  }
CellLoadingBatcher { List<List<Cell>> partitionIntoBatches( Collection<Cell> cellsToPartition, InetSocketAddress host, TableReference tableReference) { CassandraCellLoadingConfig config = loadingConfigSupplier.get(); ListMultimap<byte[], Cell> cellsByColumn = indexCellsByColumnName(cellsToPartition); List<List<Cell>> batches = Lists.newArrayList(); List<Cell> cellsForCrossColumnBatching = Lists.newArrayList(); for (Map.Entry<byte[], List<Cell>> cellColumnPair : Multimaps.asMap(cellsByColumn).entrySet()) { if (shouldExplicitlyAllocateBatchToColumn(config, cellColumnPair.getValue())) { batches.addAll(partitionBySingleQueryLoadBatchLimit( cellColumnPair.getValue(), config, host, tableReference)); } else { cellsForCrossColumnBatching.addAll(cellColumnPair.getValue()); } } batches.addAll(Lists.partition(cellsForCrossColumnBatching, config.crossColumnLoadBatchLimit())); return batches; } CellLoadingBatcher(Supplier<CassandraCellLoadingConfig> loadingConfigSupplier,
            BatchCallback rebatchingManyRowsForColumnCallback);   }