package com.facebook.presto.operator;

import com.facebook.presto.block.BlockStream;
import com.facebook.presto.block.Cursor;
import com.facebook.presto.block.position.PositionsBlock;
import com.facebook.presto.Range;
import com.facebook.presto.TupleInfo;
import com.facebook.presto.block.Block;
import com.facebook.presto.block.BlockCursor;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.ImmutableList;

import java.util.Iterator;

public class DataScan1
        implements BlockStream<Block>
{
    private static final int RANGES_PER_BLOCK = 100;
    private static final TupleInfo INFO = new TupleInfo();

    private final BlockStream<? extends Block> source;
    private final Predicate<BlockCursor> predicate;

    public DataScan1(BlockStream<? extends Block> source, Predicate<BlockCursor> predicate)
    {
        Preconditions.checkNotNull(source, "source is null");
        Preconditions.checkNotNull(predicate, "predicate is null");
        this.source = source;
        this.predicate = predicate;
    }

    @Override
    public TupleInfo getTupleInfo()
    {
        return INFO;
    }

    @Override
    public Iterator<Block> iterator()
    {
        return new AbstractIterator<Block>()
        {
            Iterator<? extends Block> sourceIterator = source.iterator();

            @Override
            protected Block computeNext()
            {
                int rangesCount = 0;
                ImmutableList.Builder<Range> ranges = ImmutableList.builder();
                while (rangesCount < RANGES_PER_BLOCK && sourceIterator.hasNext()) {
                    BlockCursor blockCursor = sourceIterator.next().blockCursor();
                    while (blockCursor.advanceToNextValue()) {
                        if (predicate.apply(blockCursor)) {
                            ranges.add(new Range(blockCursor.getPosition(), blockCursor.getValuePositionEnd()));
                            rangesCount++;
                        }
                    }
                }
                if (rangesCount == 0) {
                    endOfData();
                    return null;
                }
                return new PositionsBlock(ranges.build());
            }
        };
    }

    @Override
    public Cursor cursor()
    {
        return new ValueCursor(INFO, source.iterator());
    }
}

@@ -32,16 +32,128 @@
 rator;%0A%0A
+import com.facebook.presto.Range;%0Aimport com.facebook.presto.TupleInfo;%0Aimport com.facebook.presto.block.Block;%0A
 import c
@@ -293,166 +293,8 @@
 ck;%0A
-import com.facebook.presto.Range;%0Aimport com.facebook.presto.TupleInfo;%0Aimport com.facebook.presto.block.Block;%0Aimport com.facebook.presto.block.BlockCursor;%0A
 impo
@@ -754,37 +754,32 @@
 final Predicate%3C
-Block
 Cursor%3E predicat
@@ -850,21 +850,16 @@
 edicate%3C
-Block
 Cursor%3E 
@@ -1158,38 +1158,24 @@
 NFO;%0A    %7D%0A%0A
-    @Override%0A
     public I
@@ -1277,46 +1277,19 @@
     
-Iterator%3C? extends Block%3E sourceIterat
+Cursor curs
 or =
@@ -1292,30 +1292,28 @@
 or = source.
-iterat
+curs
 or();%0A%0A     
@@ -1562,152 +1562,9 @@
  && 
-sourceIterator.hasNext()) %7B%0A                    BlockCursor blockCursor = sourceIterator.next().blockCursor();%0A                    while (blockC
+c
 urso
@@ -1576,10 +1576,8 @@
 ance
-To
 Next
@@ -1603,28 +1603,24 @@
             
- 
-   
 if (predicat
@@ -1627,22 +1627,17 @@
 e.apply(
-blockC
+c
 ursor)) 
@@ -1634,28 +1634,24 @@
 (cursor)) %7B%0A
-    
             
@@ -1683,22 +1683,17 @@
 w Range(
-blockC
+c
 ursor.ge
@@ -1709,14 +1709,9 @@
 (), 
-blockC
+c
 urso
@@ -1719,21 +1719,31 @@
 .get
+Current
 Value
+End
 Position
 End(
@@ -1742,21 +1742,14 @@
 tion
-End
 ()));%0A
-    
     
@@ -1787,34 +1787,8 @@
 ++;%0A
-                        %7D%0A
     
@@ -2122,23 +2122,16 @@
 r(INFO, 
-source.
 iterator

Change DataScan1 to use cursor instead of iterator of blocks