A VoltDB extension to support LIMIT PARTITION ROWS
976974
Constraint getLimitConstraint() {
        Constraint result = null;
        for (Constraint constraint : getConstraints()) {
            if (constraint.getConstraintType() == Constraint.LIMIT) {
                // We're assuming only one LIMIT constraint at the moment
                result = constraint;
                break;
            }
        }
        return result;
    }
38857
public void putIndex(IndexRecord rec, int partition) {
    final int pos = partition * MapTask.MAP_OUTPUT_INDEX_RECORD_LENGTH / 8;
    entries.put(pos, rec.startOffset);
    entries.put(pos + 1, rec.rawLength);
    entries.put(pos + 2, rec.partLength);
  }
758005
@SuppressWarnings("unchecked")
    public CassandraJavaPairRDD<K, V> perPartitionLimit(Long rowsNumber) {
        CassandraRDD<Tuple2<K, V>> newRDD = rdd().perPartitionLimit(rowsNumber);
        return wrap(newRDD);
    }
696784
def partition(cls, iterable, pred):
    

    t1, t2 = itertools.tee(iterable)
    return cls(itertools.filterfalse(pred, t1), filter(pred, t2))
106279
private Iterator<Record> rowQuery(final List<Map.Entry<ByteBuffer, Key>> rowKeys, final ReadConsistency consistency,
                                      final DeltaPlacement placement) {
        List<ByteBuffer> keys = Lists.newArrayListWithCapacity(rowKeys.size());
        final Map<ByteBuffer, Key> rawKeyMap = Maps.newHashMap();
        for (Map.Entry<ByteBuffer, Key> entry : rowKeys) {
            keys.add(entry.getKey());
            rawKeyMap.put(entry.getKey(), entry.getValue());
        }

        BlockedDeltaTableDDL tableDDL = placement.getBlockedDeltaTableDDL();

        Statement statement = selectDeltaFrom(tableDDL)
                .where(in(tableDDL.getRowKeyColumnName(), keys))
                .setConsistencyLevel(SorConsistencies.toCql(consistency));

        Iterator<Iterable<Row>> rowGroups = deltaQueryAsync(placement, statement, false, "Failed to read records %s", rawKeyMap.values());

        return Iterators.concat(
                // First iterator reads the row groups found and transforms them to Records
                Iterators.transform(rowGroups, rows -> {
                    ByteBuffer keyBytes = getRawKeyFromRowGroup(rows);
                    Key key = rawKeyMap.remove(keyBytes);
                    assert key != null : "Query returned row with a key out of bound";
                    return newRecordFromCql(key, rows, placement, ByteBufferUtil.bytesToHex(keyBytes));
                }),
                // Second iterator returns an empty Record for each key queried but not found.
                new AbstractIterator<Record>() {
                    private Iterator<Key> _nonExistentKeyIterator;

                    @Override
                    protected Record computeNext() {
                        // Lazily return an empty record for each key not found in the previous iterator.
                        // rawKeyMap.iterator() must not be called until the first iterator is completely spent.
                        if (_nonExistentKeyIterator == null) {
                            _nonExistentKeyIterator = rawKeyMap.values().iterator();
                        }
                        return _nonExistentKeyIterator.hasNext() ?
                                emptyRecord(_nonExistentKeyIterator.next()) :
                                endOfData();
                    }
                });
    }
1090574
public Result runPartitionIndexOrPartitionScanQueryOnGivenOwnedPartition(Query query, int partitionId) {
        MapContainer mapContainer = mapServiceContext.getMapContainer(query.getMapName());
        PartitionIdSet partitions = singletonPartitionIdSet(partitionCount, partitionId);

        // first we optimize the query
        Predicate predicate = queryOptimizer.optimize(query.getPredicate(), mapContainer.getIndexes(partitionId));

        Collection<QueryableEntry> entries = null;
        Indexes indexes = mapContainer.getIndexes(partitionId);
        if (indexes != null && !indexes.isGlobal()) {
            entries = indexes.query(predicate);
        }

        Result result;
        if (entries == null) {
            result = createResult(query, partitions);
            partitionScanExecutor.execute(query.getMapName(), predicate, partitions, result);
            result.completeConstruction(partitions);
        } else {
            result = populateNonEmptyResult(query, entries, partitions);
        }

        return result;
    }
749087
private DBObject createQueryPartition(MongoPartition partition) {

        QueryBuilder queryBuilderMin = QueryBuilder.start(partition.getKey());
        DBObject bsonObjectMin = queryBuilderMin.greaterThanEquals(partition.splitWrapper().getStartToken()).get();

        QueryBuilder queryBuilderMax = QueryBuilder.start(partition.getKey());
        DBObject bsonObjectMax = queryBuilderMax.lessThan(partition.splitWrapper().getEndToken()).get();

        QueryBuilder queryBuilder = QueryBuilder.start();
        if (partition.splitWrapper().getStartToken() != null) {
            queryBuilder.and(bsonObjectMin);
        }

        if (partition.splitWrapper().getEndToken() != null) {
            queryBuilder.and(bsonObjectMax);
        }

        LOG.debug("mongodb query "+queryBuilder.get());

        return queryBuilder.get();
    }
956008
private Iterator<Record> rowQuery(DeltaPlacement placement,
                                      List<Map.Entry<ByteBuffer, Key>> keys,
                                      ReadConsistency consistency) {
        // Build the list of row IDs to query for.
        List<ByteBuffer> rowIds = Lists.transform(keys, entryKeyFunction());

        // Query for Delta & Compaction info, just the first 50 columns for now.
        final Rows<ByteBuffer, DeltaKey> rows = execute(placement.getKeyspace()
                        .prepareQuery(placement.getBlockedDeltaColumnFamily(), SorConsistencies.toAstyanax(consistency))
                        .getKeySlice(rowIds)
                        .withColumnRange(_maxColumnsRange),
                "query %d keys from placement %s", rowIds.size(), placement.getName());

        // Track metrics
        _randomReadMeter.mark(rowIds.size());

        // Return an iterator that decodes the row results, avoiding pinning multiple decoded rows into memory at once.
        return decodeRows(keys, rows, _maxColumnsRange.getLimit(), consistency);
    }
240776
private Result runOnAllPartitions(Query query) {
        PartitionIdSet mutablePartitionIds = getAllPartitionIds();

        Result result = doRunOnQueryThreads(query, mutablePartitionIds, ALL_NODES);
        if (isResultFromAnyPartitionMissing(mutablePartitionIds)) {
            doRunOnPartitionThreads(query, mutablePartitionIds, result);
        }
        assertAllPartitionsQueried(mutablePartitionIds);

        return result;
    }
176527
private static Pair<List<DimensionSpec>, List<DimensionSpec>> partitionDimensionList(
      StorageAdapter adapter,
      List<DimensionSpec> dimensions
  )
  {
    final List<DimensionSpec> bitmapDims = new ArrayList<>();
    final List<DimensionSpec> nonBitmapDims = new ArrayList<>();
    final List<DimensionSpec> dimsToSearch = getDimsToSearch(
        adapter.getAvailableDimensions(),
        dimensions
    );

    for (DimensionSpec spec : dimsToSearch) {
      ColumnCapabilities capabilities = adapter.getColumnCapabilities(spec.getDimension());
      if (capabilities == null) {
        continue;
      }

      if (capabilities.hasBitmapIndexes()) {
        bitmapDims.add(spec);
      } else {
        nonBitmapDims.add(spec);
      }
    }

    return new Pair<>(bitmapDims, nonBitmapDims);
  }
292255
private void moveRecords(Segment seg, ByteBuffer max, boolean deleteFromSource) {
        checkWritesAllowed();
        ByteBuffer from = seg.getMin();
        ByteBuffer toExclusive = successor(max);

        int batchSize = scanBatchSize();
        Iterator<List<ByteBuffer>> batchIter = Iterators.partition(
                _dao.scanRecords(seg.getDataId(), from, toExclusive, batchSize, Integer.MAX_VALUE),
                batchSize);
        while (batchIter.hasNext()) {
            List<ByteBuffer> records = batchIter.next();

            // Write to the destination.  Go through addAll() to update stats, do splitting, etc.
            addAll(records);

            // Delete individual records from the source.
            if (deleteFromSource) {
                _dao.prepareUpdate(_name).deleteRecords(seg.getDataId(), records).execute();
            }
        }

        seg.setMin(toExclusive);
    }
623414
private Iterable<MultiTableScanResult> scanMultiTableRows(
            final TableSet tables, final DeltaPlacement placement, final ByteBufferRange rowRange,
            final LimitCounter limit, final boolean includeDroppedTables, final boolean includeMirrorTables,
            final ReadConsistency consistency, final Instant cutoffTime) {

        // Avoiding pinning multiple decoded rows into memory at once.
        return () -> limit.limit(new AbstractIterator<MultiTableScanResult>() {
            private PeekingIterator<Iterable<Row>> _iter = Iterators.peekingIterator(
                    rowScan(placement, null, rowRange, consistency));

            private long _lastTableUuid = -1;
            private AstyanaxTable _table = null;
            private boolean _droppedTable;
            private boolean _primaryTable;

            @Override
            protected MultiTableScanResult computeNext() {
                while (_iter.hasNext()) {
                    // Get the next rows from the grouping iterator.  All rows in the returned Iterable
                    // are from the same Cassandra wide row (in other words, they share the same key).
                    final Iterable<Row> rows = _iter.next();

                    // filter the rows if a cutOff time is specified.
                    Iterable<Row> filteredRows = rows;
                    if (cutoffTime != null) {
                        filteredRows = getFilteredRows(rows, cutoffTime);
                    }

                    // Convert the filteredRows into a Record object
                    ByteBuffer rowKey = getRawKeyFromRowGroupOrNull(filteredRows);
                    // rowKey can be null if "all" the rows of the cassandra record are after the cutoff time. In such case ignore that record and continue.
                    if (rowKey == null) {
                        continue;
                    }

                    long tableUuid = AstyanaxStorage.getTableUuid(rowKey);
                    if (_lastTableUuid != tableUuid) {
                        _lastTableUuid = tableUuid;
                        try {
                            _table = (AstyanaxTable) tables.getByUuid(tableUuid);
                        } catch (UnknownTableException e) {
                            _table = AstyanaxTable.createUnknown(tableUuid, placement, e.getTable());
                        } catch (DroppedTableException e) {
                            _table = AstyanaxTable.createUnknown(tableUuid, placement, e.getPriorTable());
                        }
                        _droppedTable = _table.isUnknownTable();
                        _primaryTable = _table.getReadStorage().hasUUID(tableUuid);
                    }

                    // Skip dropped and mirror tables if configured
                    if ((!includeDroppedTables && _droppedTable) || (!includeMirrorTables && !_primaryTable)) {
                        _iter = skipToNextTable(tableUuid);
                        continue;
                    }

                    int shardId = AstyanaxStorage.getShardId(rowKey);
                    String key = AstyanaxStorage.getContentKey(rowKey);
                    Record record = newRecordFromCql(new Key(_table, key), filteredRows);
                    return new MultiTableScanResult(rowKey, shardId, tableUuid, _droppedTable, record);
                }

                return endOfData();
            }

            private PeekingIterator<Iterable<Row>> skipToNextTable(long tableUuid) {
                // Iterate over the next 10 row groups first to check for a table switch.  This avoids starting a new range
                // query if the number of rows in the undesired table is small.
                int skipLimit = 10;
                Iterable<Row> rowGroup = null;

                while (skipLimit != 0 && _iter.hasNext()) {
                    rowGroup = _iter.peek();
                    ByteBuffer rawKey = getRawKeyFromRowGroup(rowGroup);
                    long nextTableUuid = AstyanaxStorage.getTableUuid(rawKey);
                    if (nextTableUuid != tableUuid) {
                        // This is the first row of a new table
                        return _iter;
                    } else {
                        _iter.next();
                        skipLimit -= 1;
                    }
                }

                if (_iter.hasNext()) {
                    // Skip the table entirely by starting a new query on the next possible table
                    assert rowGroup != null;
                    int shardId = AstyanaxStorage.getShardId(getRawKeyFromRowGroup(rowGroup));
                    ByteBuffer nextPossibleTableStart = AstyanaxStorage.getRowKeyRaw(shardId, tableUuid + 1, "");
                    ByteBuffer end = rowRange.getEnd();

                    if (AstyanaxStorage.compareKeys(nextPossibleTableStart, end) < 0) {
                        // We haven't reached the last end boundary of the original range scan
                        ByteBufferRange updatedRange = new ByteBufferRangeImpl(nextPossibleTableStart, end, -1, false);
                        return Iterators.peekingIterator(rowScan(placement, null, updatedRange, consistency));
                    }
                }

                return Iterators.peekingIterator(Iterators.<Iterable<Row>>emptyIterator());
            }
        });
    }
1050747
public CassandraJavaRDD<R> perPartitionLimit(Long rowsNumber) {
        CassandraRDD<R> newRDD = rdd().perPartitionLimit(rowsNumber);
        return wrap(newRDD);
    }
99162
def ensure_row_dep_constraint(
            self, M_c, T, X_L, X_D, row1, row2, dependent=True, wrt=None,
            max_iter=100, force=False):
        """"""
        X_L_list, X_D_list, was_multistate = su.ensure_multistate(X_L, X_D)

        if force:
            raise NotImplementedError
        else:
            kernel_list = ('row_partition_assignements',)
            for i, (X_L_i, X_D_i) in enumerate(zip(X_L_list, X_D_list)):

                iters = 0
                X_L_tmp = copy.deepcopy(X_L_i)
                X_D_tmp = copy.deepcopy(X_D_i)

                while not self.assert_row(
                        X_L_tmp, X_D_tmp, row1, row2,
                        dependent=dependent, wrt=wrt):
                    if iters >= max_iter:
                        raise RuntimeError(
                            'Maximum ensure iterations reached.')
                    # XXX No seed?
                    res = self.analyze(
                        M_c, T, X_L_i, X_D_i, kernel_list=kernel_list,
                        n_steps=1, r=(row1,))
                    X_L_tmp = res[0]
                    X_D_tmp = res[1]
                    iters += 1

                X_L_list[i] = X_L_tmp
                X_D_list[i] = X_D_tmp

        if was_multistate:
            return X_L_list, X_D_list
        else:
            return X_L_list[0], X_D_list[0]
675657
protected ArrayDBIDs rowsBitsetToIDs(long[] rows) {
    ArrayModifiableDBIDs rowIDs = DBIDUtil.newArray(BitsUtil.cardinality(rows));
    DBIDArrayIter iter = this.rowIDs.iter();
    outer: for(int rlpos = 0; rlpos < rows.length; ++rlpos) {
      long rlong = rows[rlpos];
      // Fast skip blocks of 64 masked values.
      if(rlong == 0L) {
        iter.advance(Long.SIZE);
        continue;
      }
      for(int i = 0; i < Long.SIZE; ++i, rlong >>>= 1, iter.advance()) {
        if(!iter.valid()) {
          break outer;
        }
        if((rlong & 1L) == 1L) {
          rowIDs.add(iter);
        }
      }
    }
    return rowIDs;
  }
100721
public synchronized long update(final PartitionInstance partitionInstance, final long vbuuid, final long seqno) {
    return update(partitionInstance.partition(), partitionInstance.slot(), vbuuid, seqno);
  }
858015
def records_for_rebuild(batch_size = 1000)
      transaction do
        if use_fast_batches?
          offset = 0
          while (rows = where([ "#{table_name}.id > ?", offset ]).limit(batch_size).all).any?
            offset = rows.last.id
            yield rows, offset
          end
        else
          order = "#{primary_key} ASC" # fixes #212
          0.step(self.count, batch_size) do |offset|
            yield scoped.limit(batch_size).offset(offset).order(order).all, offset
          end
        end
      end
    end