src/java/org/apache/cassandra/db/CollationController.java
/**
     * Collects data in order of recency, using the sstable maxtimestamp data.
     * Once we have data for all requests columns that is newer than the newest remaining maxtimestamp,
     * we stop.
     */
Collects data in order of recency, using the sstable maxtimestamp data.
Once we have data for all requests columns that is newer than the newest remaining maxtimestamp,
we stop.
Collects data in order of recency using the sstable maxtimestamp data . Once we have data for all requests columns that is newer than the newest remaining maxtimestamp we stop .
private ColumnFamily collectTimeOrderedData()
    {
        logger.debug("collectTimeOrderedData");
        ISortedColumns.Factory factory = mutableColumns
                                       ? AtomicSortedColumns.factory()
                                       : TreeMapBackedSortedColumns.factory();
        ColumnFamily container = ColumnFamily.create(cfs.metadata, factory, filter.filter.isReversed());
        List<IColumnIterator> iterators = new ArrayList<IColumnIterator>();
        ColumnFamilyStore.ViewFragment view = cfs.markReferenced(filter.key);
        try
        {
            for (Memtable memtable : view.memtables)
            {
                IColumnIterator iter = filter.getMemtableColumnIterator(memtable);
                if (iter != null)
                {
                    iterators.add(iter);
                    container.delete(iter.getColumnFamily());
                    while (iter.hasNext())
                        container.addColumn(iter.next());
                }
            }
            TreeSet<ByteBuffer> filterColumns = new TreeSet<ByteBuffer>(((NamesQueryFilter) filter.filter).columns);
            QueryFilter reducedFilter = new QueryFilter(filter.key, filter.path, new NamesQueryFilter(filterColumns));
            /* add the SSTables on disk */
            Collections.sort(view.sstables, SSTable.maxTimestampComparator);
            for (SSTableReader sstable : view.sstables)
            {
                long currentMaxTs = sstable.getMaxTimestamp();
                reduceNameFilter(reducedFilter, container, currentMaxTs);
                if (((NamesQueryFilter) reducedFilter.filter).columns.isEmpty())
                    break;
                IColumnIterator iter = reducedFilter.getSSTableColumnIterator(sstable);
                iterators.add(iter);
                if (iter.getColumnFamily() != null)
                {
                    container.delete(iter.getColumnFamily());
                    sstablesIterated++;
                    while (iter.hasNext())
                        container.addColumn(iter.next());
                }
            }
            if (iterators.isEmpty())
                return null;
            final ColumnFamily c2 = container;
            CloseableIterator<IColumn> toCollate = new SimpleAbstractColumnIterator()
            {
                final Iterator<IColumn> iter = c2.iterator();
                protected IColumn computeNext()
                {
                    return iter.hasNext() ? iter.next() : endOfData();
                }
                public ColumnFamily getColumnFamily()
                {
                    return c2;
                }
                public DecoratedKey getKey()
                {
                    return filter.key;
                }
            };
            ColumnFamily returnCF = container.cloneMeShallow();
            filter.collateColumns(returnCF, Collections.singletonList(toCollate), gcBefore);
            if (sstablesIterated > cfs.getMinimumCompactionThreshold()
                && !cfs.isCompactionDisabled()
                && cfs.getCompactionStrategy() instanceof SizeTieredCompactionStrategy)
            {
                RowMutation rm = new RowMutation(cfs.table.name, new Row(filter.key, returnCF.cloneMe()));
                try
                {
                    Table.open(rm.getTable()).apply(rm, false, false);
                }
                catch (IOException e)
                {
                    logger.error("Error re-writing read results", e);
                }
            }
            return returnCF;
        }
        finally
        {
            for (IColumnIterator iter : iterators)
                FileUtils.closeQuietly(iter);
            SSTableReader.releaseReferences(view.sstables);
        }
    }
private ColumnFamily collectTimeOrderedData ( ) { logger . debug ( "collectTimeOrderedData" ) ; ISortedColumns . Factory factory = mutableColumns ? AtomicSortedColumns . factory ( ) : TreeMapBackedSortedColumns . factory ( ) ; ColumnFamily container = ColumnFamily . create ( cfs . metadata , factory , filter . filter . isReversed ( ) ) ; List < IColumnIterator > iterators = new ArrayList < IColumnIterator > ( ) ; ColumnFamilyStore . ViewFragment view = cfs . markReferenced ( filter . key ) ; try { for ( Memtable memtable : view . memtables ) { IColumnIterator iter = filter . getMemtableColumnIterator ( memtable ) ; if ( iter != null ) { iterators . add ( iter ) ; container . delete ( iter . getColumnFamily ( ) ) ; while ( iter . hasNext ( ) ) container . addColumn ( iter . next ( ) ) ; } } TreeSet < ByteBuffer > filterColumns = new TreeSet < ByteBuffer > ( ( ( NamesQueryFilter ) filter . filter ) . columns ) ; QueryFilter reducedFilter = new QueryFilter ( filter . key , filter . path , new NamesQueryFilter ( filterColumns ) ) ; /* add the SSTables on disk */ Collections . sort ( view . sstables , SSTable . maxTimestampComparator ) ; for ( SSTableReader sstable : view . sstables ) { long currentMaxTs = sstable . getMaxTimestamp ( ) ; reduceNameFilter ( reducedFilter , container , currentMaxTs ) ; if ( ( ( NamesQueryFilter ) reducedFilter . filter ) . columns . isEmpty ( ) ) break ; IColumnIterator iter = reducedFilter . getSSTableColumnIterator ( sstable ) ; iterators . add ( iter ) ; if ( iter . getColumnFamily ( ) != null ) { container . delete ( iter . getColumnFamily ( ) ) ; sstablesIterated ++ ; while ( iter . hasNext ( ) ) container . addColumn ( iter . next ( ) ) ; } } if ( iterators . isEmpty ( ) ) return null ; final ColumnFamily c2 = container ; CloseableIterator < IColumn > toCollate = new SimpleAbstractColumnIterator ( ) { final Iterator < IColumn > iter = c2 . iterator ( ) ; protected IColumn computeNext ( ) { return iter . hasNext ( ) ? iter . next ( ) : endOfData ( ) ; } public ColumnFamily getColumnFamily ( ) { return c2 ; } public DecoratedKey getKey ( ) { return filter . key ; } } ; ColumnFamily returnCF = container . cloneMeShallow ( ) ; filter . collateColumns ( returnCF , Collections . singletonList ( toCollate ) , gcBefore ) ; if ( sstablesIterated > cfs . getMinimumCompactionThreshold ( ) && ! cfs . isCompactionDisabled ( ) && cfs . getCompactionStrategy ( ) instanceof SizeTieredCompactionStrategy ) { RowMutation rm = new RowMutation ( cfs . table . name , new Row ( filter . key , returnCF . cloneMe ( ) ) ) ; try { Table . open ( rm . getTable ( ) ) . apply ( rm , false , false ) ; } catch ( IOException e ) { logger . error ( "Error re-writing read results" , e ) ; } } return returnCF ; } finally { for ( IColumnIterator iter : iterators ) FileUtils . closeQuietly ( iter ) ; SSTableReader . releaseReferences ( view . sstables ) ; } }
Collects data in order of recency, using the sstable maxtimestamp data.
Collects data in order of recency using the sstable maxtimestamp data .
// avoid changing the filter columns of the original filter // (reduceNameFilter removes columns that are known to be irrelevant) // read sorted sstables // we need to distinguish between "there is no data at all for this row" (BF will let us rebuild that efficiently) // and "there used to be data, but it's gone now" (we should cache the empty CF so we don't need to rebuild that slower) // do a final collate.  toCollate is boilerplate required to provide a CloseableIterator // "hoist up" the requested data into a more recent sstable // skipping commitlog and index updates is fine since we're just de-fragmenting existing data // log and allow the result to be returned // Caller is responsible for final removeDeletedCF.  This is important for cacheRow to work correctly: