Mailing-List: contact oak-commits-help@jackrabbit.apache.org; run by ezmlm
Precedence: bulk
Reply-To: oak-dev@jackrabbit.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r1444167 - in /jackrabbit/oak/trunk/oak-core/src:
 main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java
 test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java
Date: Fri, 08 Feb 2013 17:55:40 -0000
To: oak-commits@jackrabbit.apache.org
From: jukka@apache.org
Message-Id: <20130208175540.6DF632388900@eris.apache.org>

Author: jukka
Date: Fri Feb  8 17:55:39 2013
New Revision: 1444167

URL: http://svn.apache.org/r1444167
Log:
OAK-593: Segment-based MK

Implement iteration over all entries in a map record

Modified:
    jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java
    jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java

Modified: jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java?rev=1444167&r1=1444166&r2=1444167&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/segment/MapRecord.java Fri Feb  8 17:55:39 2013
@@ -1,74 +1,146 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.jackrabbit.oak.plugins.segment;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-class MapRecord extends Record {
-
-    static final int LEVEL_BITS = 6;
-
-    MapRecord(RecordId id) {
-        super(id);
-    }
-
-    public int size(SegmentReader reader) {
-        return reader.readInt(getRecordId(), 0);
-    }
-
-    public RecordId getEntry(SegmentReader reader, String key) {
-        checkNotNull(key);
-        return getEntry(reader, key, 0);
-    }
-
-    private RecordId getEntry(SegmentReader reader, String key, int level) {
-        int size = 1 << LEVEL_BITS;
-        int mask = size - 1;
-        int shift = level * LEVEL_BITS;
-
-        int code = key.hashCode();
-        int bucketSize = reader.readInt(getRecordId(), 0);
-        if (bucketSize == 0) {
-            return null;
-        } else if (bucketSize <= size || shift >= 32) {
-            int offset = 0;
-            while (offset < bucketSize && reader.readInt(getRecordId(), 4 + offset * 4) < code) {
-                offset++;
-            }
-            while (offset < bucketSize && reader.readInt(getRecordId(), 4 + offset * 4) == code) {
-                RecordId keyId = reader.readRecordId(getRecordId(), 4 + (bucketSize + offset) * 4);
-                if (key.equals(reader.readString(keyId))) {
-                    return reader.readRecordId(getRecordId(), 4 + (2 * bucketSize + offset) * 4);
-                }
-                offset++;
-            }
-            return null;
-        } else {
-            long bucketMap = reader.readLong(getRecordId(), 4);
-            int bucketIndex = (code >> shift) & mask;
-            long bucketBit = 1L << bucketIndex;
-            if ((bucketMap & bucketBit) != 0) {
-                bucketIndex = Long.bitCount(bucketMap & (bucketBit - 1));
-                RecordId bucketId = reader.readRecordId(getRecordId(), 12 + bucketIndex * 4);
-                return new MapRecord(bucketId).getEntry(reader, key, level + 1);
-            } else {
-                return null;
-            }
-        }
-    }
-
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.segment;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+class MapRecord extends Record {
+
+    public interface Entry extends Map.Entry<String, RecordId> {    
+    }
+
+    static final int LEVEL_BITS = 6;
+
+    MapRecord(RecordId id) {
+        super(id);
+    }
+
+    public int size(SegmentReader reader) {
+        return reader.readInt(getRecordId(), 0);
+    }
+
+    public RecordId getEntry(SegmentReader reader, String key) {
+        checkNotNull(key);
+        return getEntry(reader, key, 0);
+    }
+
+    private RecordId getEntry(SegmentReader reader, String key, int level) {
+        int size = 1 << LEVEL_BITS;
+        int mask = size - 1;
+        int shift = level * LEVEL_BITS;
+
+        int code = key.hashCode();
+        int bucketSize = reader.readInt(getRecordId(), 0);
+        if (bucketSize == 0) {
+            return null;
+        } else if (bucketSize <= size || shift >= 32) {
+            int offset = 0;
+            while (offset < bucketSize && reader.readInt(getRecordId(), 4 + offset * 4) < code) {
+                offset++;
+            }
+            while (offset < bucketSize && reader.readInt(getRecordId(), 4 + offset * 4) == code) {
+                RecordId keyId = reader.readRecordId(getRecordId(), 4 + (bucketSize + offset) * 4);
+                if (key.equals(reader.readString(keyId))) {
+                    return reader.readRecordId(getRecordId(), 4 + (2 * bucketSize + offset) * 4);
+                }
+                offset++;
+            }
+            return null;
+        } else {
+            long bucketMap = reader.readLong(getRecordId(), 4);
+            int bucketIndex = (code >> shift) & mask;
+            long bucketBit = 1L << bucketIndex;
+            if ((bucketMap & bucketBit) != 0) {
+                bucketIndex = Long.bitCount(bucketMap & (bucketBit - 1));
+                RecordId bucketId = reader.readRecordId(getRecordId(), 12 + bucketIndex * 4);
+                return new MapRecord(bucketId).getEntry(reader, key, level + 1);
+            } else {
+                return null;
+            }
+        }
+    }
+
+    public Iterable<Entry> getEntries(SegmentReader reader) {
+        return getEntries(reader, 0);
+    }
+
+    private Iterable<Entry> getEntries(
+            final SegmentReader reader, int level) {
+        int size = 1 << LEVEL_BITS;
+        int shift = level * LEVEL_BITS;
+
+        final int bucketSize = reader.readInt(getRecordId(), 0);
+        if (bucketSize == 0) {
+            return Collections.emptyList();
+        } else if (bucketSize <= size || shift >= 32) {
+            return new Iterable<Entry>() {
+                @Override
+                public Iterator<Entry> iterator() {
+                    return new Iterator<Entry>() {
+                        private int index = 0;
+                        @Override
+                        public boolean hasNext() {
+                            return index < bucketSize;
+                        }
+                        @Override
+                        public Entry next() {
+                            final int offset = index++;
+                            return new Entry() {
+                                @Override
+                                public String getKey() {
+                                    RecordId id = reader.readRecordId(getRecordId(), 4 + (bucketSize + offset) * 4);
+                                    return reader.readString(id);
+                                }
+                                @Override
+                                public RecordId getValue() {
+                                    return reader.readRecordId(getRecordId(), 4 + (2 * bucketSize + offset) * 4);
+                                }
+                                @Override
+                                public RecordId setValue(RecordId arg0) {
+                                    throw new UnsupportedOperationException();
+                                }
+                            };
+                        }
+                        @Override
+                        public void remove() {
+                            throw new UnsupportedOperationException();
+                        }
+                    };
+                }
+            };
+        } else {
+            long bucketMap = reader.readLong(getRecordId(), 4);
+            int bucketCount = Long.bitCount(bucketMap);
+            List<Iterable<Entry>> iterables =
+                    Lists.newArrayListWithCapacity(bucketCount);
+            for (int i = 0; i < bucketCount; i++) {
+                RecordId bucketId = reader.readRecordId(getRecordId(), 12 + i * 4);
+                iterables.add(new MapRecord(bucketId).getEntries(reader, level + 1));
+            }
+            return Iterables.concat(iterables);
+        }
+    }
+
+}

Modified: jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java?rev=1444167&r1=1444166&r2=1444167&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java (original)
+++ jackrabbit/oak/trunk/oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/segment/RecordTest.java Fri Feb  8 17:55:39 2013
@@ -1,199 +1,224 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.jackrabbit.oak.plugins.segment;
-
-import static org.apache.jackrabbit.oak.plugins.segment.ListRecord.LEVEL_SIZE;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-
-import org.junit.Test;
-
-import com.google.common.base.Charsets;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Maps;
-
-public class RecordTest {
-
-    private String hello = "Hello, World!";
-
-    private byte[] bytes = hello.getBytes(Charsets.UTF_8);
-
-    private SegmentStore store = new MemoryStore();
-
-    private SegmentWriter writer = new SegmentWriter(store);
-
-    private SegmentReader reader = new SegmentReader(store);
-
-    private final Random random = new Random(0xcafefaceL);
-
-    @Test
-    public void testBlockRecord() {
-        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
-        writer.flush();
-        BlockRecord block = new BlockRecord(blockId, bytes.length);
-
-        // Check reading with all valid positions and lengths
-        for (int n = 1; n < bytes.length; n++) {
-            for (int i = 0; i + n <= bytes.length; i++) {
-                Arrays.fill(bytes, i, i + n, (byte) '.');
-                assertEquals(n, block.read(reader, i, bytes, i, n));
-                assertEquals(hello, new String(bytes, Charsets.UTF_8));
-            }
-        }
-
-        // Check reading with a too long length
-        byte[] large = new byte[bytes.length * 2];
-        assertEquals(bytes.length, block.read(reader, 0, large, 0, large.length));
-        assertEquals(hello, new String(large, 0, bytes.length, Charsets.UTF_8));
-    }
-
-    @Test
-    public void testListRecord() {
-        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
-
-        ListRecord zero = writeList(0, blockId);
-        ListRecord one = writeList(1, blockId);
-        ListRecord level1 = writeList(LEVEL_SIZE, blockId);
-        ListRecord level1p = writeList(LEVEL_SIZE + 1, blockId);
-        ListRecord level2 = writeList(LEVEL_SIZE * LEVEL_SIZE, blockId);
-        ListRecord level2p = writeList(LEVEL_SIZE * LEVEL_SIZE + 1, blockId);
-        writer.flush();
-
-        assertEquals(0, zero.size());
-        assertEquals(1, one.size());
-        assertEquals(blockId, one.getEntry(reader, 0));
-        assertEquals(LEVEL_SIZE, level1.size());
-        assertEquals(blockId, level1.getEntry(reader, 0));
-        assertEquals(blockId, level1.getEntry(reader, LEVEL_SIZE - 1));
-        assertEquals(LEVEL_SIZE + 1, level1p.size());
-        assertEquals(blockId, level1p.getEntry(reader, 0));
-        assertEquals(blockId, level1p.getEntry(reader, LEVEL_SIZE));
-        assertEquals(LEVEL_SIZE * LEVEL_SIZE, level2.size());
-        assertEquals(blockId, level2.getEntry(reader, 0));
-        assertEquals(blockId, level2.getEntry(reader, LEVEL_SIZE * LEVEL_SIZE - 1));
-        assertEquals(LEVEL_SIZE * LEVEL_SIZE + 1, level2p.size());
-        assertEquals(blockId, level2p.getEntry(reader, 0));
-        assertEquals(blockId, level2p.getEntry(reader, LEVEL_SIZE * LEVEL_SIZE));
-    }
-
-    private ListRecord writeList(int size, RecordId id) {
-        List<RecordId> list = Collections.nCopies(size, id);
-        return new ListRecord(writer.writeList(list), size);
-    }
-
-    @Test
-    public void testStreamRecord() throws IOException {
-        checkRandomStreamRecord(0);
-        checkRandomStreamRecord(1);
-        checkRandomStreamRecord(SegmentWriter.BLOCK_SIZE);
-        checkRandomStreamRecord(SegmentWriter.BLOCK_SIZE + 1);
-        checkRandomStreamRecord(SegmentWriter.INLINE_SIZE);
-        checkRandomStreamRecord(SegmentWriter.INLINE_SIZE + 1);
-        checkRandomStreamRecord(store.getMaxSegmentSize());
-        checkRandomStreamRecord(store.getMaxSegmentSize() + 1);
-        checkRandomStreamRecord(store.getMaxSegmentSize() * 2);
-        checkRandomStreamRecord(store.getMaxSegmentSize() * 2 + 1);
-    }
-
-    private void checkRandomStreamRecord(int size) throws IOException {
-        byte[] source = new byte[size];
-        random.nextBytes(source);
-
-        RecordId valueId = writer.writeStream(new ByteArrayInputStream(source));
-        writer.flush();
-
-        InputStream stream = reader.readStream(valueId);
-        try {
-            byte[] b = new byte[349]; // prime number
-            int offset = 0;
-            for (int n = stream.read(b); n != -1; n = stream.read(b)) {
-                for (int i = 0; i < n; i++) {
-                    assertEquals(source[offset + i], b[i]);
-                }
-                offset += n;
-            }
-            assertEquals(offset, size);
-            assertEquals(-1, stream.read());
-        } finally {
-            stream.close();
-        }
-    }
-
-    @Test
-    public void testStringRecord() {
-        RecordId empty = writer.writeString("");
-        RecordId space = writer.writeString(" ");
-        RecordId hello = writer.writeString("Hello, World!");
-
-        StringBuilder builder = new StringBuilder();
-        for (int i = 0; i < 100000; i++) {
-            builder.append((char) ('0' + i % 10));
-        }
-        RecordId large = writer.writeString(builder.toString());
-
-        writer.flush();
-
-        assertEquals("", reader.readString(empty));
-        assertEquals(" ", reader.readString(space));
-        assertEquals("Hello, World!", reader.readString(hello));
-        assertEquals(builder.toString(), reader.readString(large));
-    }
-
-    @Test
-    public void testMapRecord() {
-        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
-
-        MapRecord zero = new MapRecord(writer.writeMap(
-                ImmutableMap.<String, RecordId>of()));
-        MapRecord one = new MapRecord(writer.writeMap(
-                ImmutableMap.of("one", blockId)));
-        MapRecord two = new MapRecord(writer.writeMap(
-                ImmutableMap.of("one", blockId, "two", blockId)));
-        Map<String, RecordId> map = Maps.newHashMap();
-        for (int i = 0; i < 1000; i++) {
-            map.put("key" + i, blockId);
-        }
-        MapRecord many = new MapRecord(writer.writeMap(map));
-
-        writer.flush();
-
-        assertEquals(0, zero.size(reader));
-        assertNull(zero.getEntry(reader, "one"));
-        assertEquals(1, one.size(reader));
-        assertEquals(blockId, one.getEntry(reader, "one"));
-        assertNull(one.getEntry(reader, "two"));
-        assertEquals(2, two.size(reader));
-        assertEquals(blockId, two.getEntry(reader, "one"));
-        assertEquals(blockId, two.getEntry(reader, "two"));
-        assertNull(two.getEntry(reader, "three"));
-        assertEquals(1000, many.size(reader));
-        for (int i = 0; i < 1000; i++) {
-            assertEquals(blockId, many.getEntry(reader, "key" + i));
-        }
-        assertNull(many.getEntry(reader, "foo"));
-    }
-
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.oak.plugins.segment;
+
+import static org.apache.jackrabbit.oak.plugins.segment.ListRecord.LEVEL_SIZE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+
+public class RecordTest {
+
+    private String hello = "Hello, World!";
+
+    private byte[] bytes = hello.getBytes(Charsets.UTF_8);
+
+    private SegmentStore store = new MemoryStore();
+
+    private SegmentWriter writer = new SegmentWriter(store);
+
+    private SegmentReader reader = new SegmentReader(store);
+
+    private final Random random = new Random(0xcafefaceL);
+
+    @Test
+    public void testBlockRecord() {
+        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
+        writer.flush();
+        BlockRecord block = new BlockRecord(blockId, bytes.length);
+
+        // Check reading with all valid positions and lengths
+        for (int n = 1; n < bytes.length; n++) {
+            for (int i = 0; i + n <= bytes.length; i++) {
+                Arrays.fill(bytes, i, i + n, (byte) '.');
+                assertEquals(n, block.read(reader, i, bytes, i, n));
+                assertEquals(hello, new String(bytes, Charsets.UTF_8));
+            }
+        }
+
+        // Check reading with a too long length
+        byte[] large = new byte[bytes.length * 2];
+        assertEquals(bytes.length, block.read(reader, 0, large, 0, large.length));
+        assertEquals(hello, new String(large, 0, bytes.length, Charsets.UTF_8));
+    }
+
+    @Test
+    public void testListRecord() {
+        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
+
+        ListRecord zero = writeList(0, blockId);
+        ListRecord one = writeList(1, blockId);
+        ListRecord level1 = writeList(LEVEL_SIZE, blockId);
+        ListRecord level1p = writeList(LEVEL_SIZE + 1, blockId);
+        ListRecord level2 = writeList(LEVEL_SIZE * LEVEL_SIZE, blockId);
+        ListRecord level2p = writeList(LEVEL_SIZE * LEVEL_SIZE + 1, blockId);
+        writer.flush();
+
+        assertEquals(0, zero.size());
+        assertEquals(1, one.size());
+        assertEquals(blockId, one.getEntry(reader, 0));
+        assertEquals(LEVEL_SIZE, level1.size());
+        assertEquals(blockId, level1.getEntry(reader, 0));
+        assertEquals(blockId, level1.getEntry(reader, LEVEL_SIZE - 1));
+        assertEquals(LEVEL_SIZE + 1, level1p.size());
+        assertEquals(blockId, level1p.getEntry(reader, 0));
+        assertEquals(blockId, level1p.getEntry(reader, LEVEL_SIZE));
+        assertEquals(LEVEL_SIZE * LEVEL_SIZE, level2.size());
+        assertEquals(blockId, level2.getEntry(reader, 0));
+        assertEquals(blockId, level2.getEntry(reader, LEVEL_SIZE * LEVEL_SIZE - 1));
+        assertEquals(LEVEL_SIZE * LEVEL_SIZE + 1, level2p.size());
+        assertEquals(blockId, level2p.getEntry(reader, 0));
+        assertEquals(blockId, level2p.getEntry(reader, LEVEL_SIZE * LEVEL_SIZE));
+    }
+
+    private ListRecord writeList(int size, RecordId id) {
+        List<RecordId> list = Collections.nCopies(size, id);
+        return new ListRecord(writer.writeList(list), size);
+    }
+
+    @Test
+    public void testStreamRecord() throws IOException {
+        checkRandomStreamRecord(0);
+        checkRandomStreamRecord(1);
+        checkRandomStreamRecord(SegmentWriter.BLOCK_SIZE);
+        checkRandomStreamRecord(SegmentWriter.BLOCK_SIZE + 1);
+        checkRandomStreamRecord(SegmentWriter.INLINE_SIZE);
+        checkRandomStreamRecord(SegmentWriter.INLINE_SIZE + 1);
+        checkRandomStreamRecord(store.getMaxSegmentSize());
+        checkRandomStreamRecord(store.getMaxSegmentSize() + 1);
+        checkRandomStreamRecord(store.getMaxSegmentSize() * 2);
+        checkRandomStreamRecord(store.getMaxSegmentSize() * 2 + 1);
+    }
+
+    private void checkRandomStreamRecord(int size) throws IOException {
+        byte[] source = new byte[size];
+        random.nextBytes(source);
+
+        RecordId valueId = writer.writeStream(new ByteArrayInputStream(source));
+        writer.flush();
+
+        InputStream stream = reader.readStream(valueId);
+        try {
+            byte[] b = new byte[349]; // prime number
+            int offset = 0;
+            for (int n = stream.read(b); n != -1; n = stream.read(b)) {
+                for (int i = 0; i < n; i++) {
+                    assertEquals(source[offset + i], b[i]);
+                }
+                offset += n;
+            }
+            assertEquals(offset, size);
+            assertEquals(-1, stream.read());
+        } finally {
+            stream.close();
+        }
+    }
+
+    @Test
+    public void testStringRecord() {
+        RecordId empty = writer.writeString("");
+        RecordId space = writer.writeString(" ");
+        RecordId hello = writer.writeString("Hello, World!");
+
+        StringBuilder builder = new StringBuilder();
+        for (int i = 0; i < 100000; i++) {
+            builder.append((char) ('0' + i % 10));
+        }
+        RecordId large = writer.writeString(builder.toString());
+
+        writer.flush();
+
+        assertEquals("", reader.readString(empty));
+        assertEquals(" ", reader.readString(space));
+        assertEquals("Hello, World!", reader.readString(hello));
+        assertEquals(builder.toString(), reader.readString(large));
+    }
+
+    @Test
+    public void testMapRecord() {
+        RecordId blockId = writer.writeBlock(bytes, 0, bytes.length);
+
+        MapRecord zero = new MapRecord(writer.writeMap(
+                ImmutableMap.<String, RecordId>of()));
+        MapRecord one = new MapRecord(writer.writeMap(
+                ImmutableMap.of("one", blockId)));
+        MapRecord two = new MapRecord(writer.writeMap(
+                ImmutableMap.of("one", blockId, "two", blockId)));
+        Map<String, RecordId> map = Maps.newHashMap();
+        for (int i = 0; i < 1000; i++) {
+            map.put("key" + i, blockId);
+        }
+        MapRecord many = new MapRecord(writer.writeMap(map));
+
+        writer.flush();
+        Iterator<MapRecord.Entry> iterator;
+
+        assertEquals(0, zero.size(reader));
+        assertNull(zero.getEntry(reader, "one"));
+        iterator = zero.getEntries(reader).iterator();
+        assertFalse(iterator.hasNext());
+
+        assertEquals(1, one.size(reader));
+        assertEquals(blockId, one.getEntry(reader, "one"));
+        assertNull(one.getEntry(reader, "two"));
+        iterator = one.getEntries(reader).iterator();
+        assertTrue(iterator.hasNext());
+        assertEquals("one", iterator.next().getKey());
+        assertFalse(iterator.hasNext());
+
+        assertEquals(2, two.size(reader));
+        assertEquals(blockId, two.getEntry(reader, "one"));
+        assertEquals(blockId, two.getEntry(reader, "two"));
+        assertNull(two.getEntry(reader, "three"));
+        iterator = two.getEntries(reader).iterator();
+        assertTrue(iterator.hasNext());
+        iterator.next();
+        assertTrue(iterator.hasNext());
+        iterator.next();
+        assertFalse(iterator.hasNext());
+
+        assertEquals(1000, many.size(reader));
+        iterator = many.getEntries(reader).iterator();
+        for (int i = 0; i < 1000; i++) {
+            assertTrue(iterator.hasNext());
+            iterator.next();
+            assertEquals(blockId, many.getEntry(reader, "key" + i));
+        }
+        assertFalse(iterator.hasNext());
+        assertNull(many.getEntry(reader, "foo"));
+    }
+
+}