From ac3a9dc4d80241a62c76ad8d481f99ae4e5bca98 Mon Sep 17 00:00:00 2001
From: David Phillips <david@acz.org>
Date: Mon, 20 Aug 2012 18:16:45 -0700
Subject: [PATCH] Add basic UncompressedColumnInput

---
 .../presto/UncompressedColumnInput.java       | 63 +++++++++++++++++++
 .../presto/TestUncompressedColumnInput.java   | 47 ++++++++++++++
 2 files changed, 110 insertions(+)
 create mode 100644 src/main/java/com/facebook/presto/UncompressedColumnInput.java
 create mode 100644 src/test/java/com/facebook/presto/TestUncompressedColumnInput.java

diff --git a/src/main/java/com/facebook/presto/UncompressedColumnInput.java b/src/main/java/com/facebook/presto/UncompressedColumnInput.java
new file mode 100644
index 0000000000000..c60df7e3a4c7d
--- /dev/null
+++ b/src/main/java/com/facebook/presto/UncompressedColumnInput.java
@@ -0,0 +1,63 @@
+package com.facebook.presto;
+
+import com.google.common.collect.AbstractIterator;
+
+import java.util.Iterator;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+public class UncompressedColumnInput
+        implements Iterable<UncompressedValueBlock>
+{
+    private Slice slice;
+    private final int valueLength;
+
+    public UncompressedColumnInput(Slice slice, int valueLength)
+    {
+        this.slice = checkNotNull(slice, "slice");
+        this.valueLength = valueLength;
+    }
+
+    @Override
+    public Iterator<UncompressedValueBlock> iterator()
+    {
+        return new ColumnIterator(slice, valueLength);
+    }
+
+    private static class ColumnIterator
+            extends AbstractIterator<UncompressedValueBlock>
+    {
+        private final Slice slice;
+        private final int valueLength;
+        private int position = 0;
+
+        public ColumnIterator(Slice slice, int valueLength)
+        {
+            this.slice = slice;
+            this.valueLength = valueLength;
+            checkArgument((slice.length() % valueLength) == 0, "data must be a multiple of value length");
+            checkArgument(valueLength == 8, "types other than long are not currently supported");
+        }
+
+        @Override
+        protected UncompressedValueBlock computeNext()
+        {
+            if (position == slice.length()) {
+                endOfData();
+                return null;
+            }
+
+            TupleInfo tupleInfo = new TupleInfo(valueLength);
+            BlockBuilder blockBuilder = new BlockBuilder(position, tupleInfo);
+
+            do {
+                blockBuilder.append(slice.getLong(position));
+                position += valueLength;
+            }
+            while ((position < slice.length()) && !blockBuilder.isFull());
+
+            return blockBuilder.build();
+        }
+    }
+}
diff --git a/src/test/java/com/facebook/presto/TestUncompressedColumnInput.java b/src/test/java/com/facebook/presto/TestUncompressedColumnInput.java
new file mode 100644
index 0000000000000..761fe997be311
--- /dev/null
+++ b/src/test/java/com/facebook/presto/TestUncompressedColumnInput.java
@@ -0,0 +1,47 @@
+package com.facebook.presto;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.ByteArrayDataOutput;
+import com.google.common.io.ByteStreams;
+import com.google.common.primitives.Longs;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static com.facebook.presto.SizeOf.SIZE_OF_LONG;
+
+public class TestUncompressedColumnInput
+{
+    @Test
+    public void testIterator()
+            throws Exception
+    {
+        Slice slice = uncompressedLongs(0, 1, 2, 4, 9, 12345);
+        UncompressedColumnInput firstColumn = new UncompressedColumnInput(slice, SIZE_OF_LONG);
+
+        ImmutableList<Pair> actual = ImmutableList.copyOf(new PairsIterator(firstColumn.iterator()));
+        Assert.assertEquals(actual,
+                ImmutableList.of(
+                        new Pair(0, createTuple(0)),
+                        new Pair(1, createTuple(1)),
+                        new Pair(2, createTuple(2)),
+                        new Pair(3, createTuple(4)),
+                        new Pair(4, createTuple(9)),
+                        new Pair(5, createTuple(12345))));
+    }
+
+    private Tuple createTuple(long value)
+    {
+        Slice slice = Slices.allocate(SIZE_OF_LONG);
+        slice.setLong(0, value);
+        return new Tuple(slice, new TupleInfo(SIZE_OF_LONG));
+    }
+
+    private static Slice uncompressedLongs(long... longs)
+    {
+        ByteArrayDataOutput buffer = ByteStreams.newDataOutput(longs.length * 8);
+        for (long v : longs) {
+            buffer.write(Longs.toByteArray(Long.reverseBytes(v)));
+        }
+        return new Slice(buffer.toByteArray());
+    }
+}