From a30209bcc744c80d4e724549ba0ede886badc92a Mon Sep 17 00:00:00 2001
From: Chris Masone <cmasone@atticlabs.io>
Date: Tue, 15 Dec 2015 17:38:22 -0800
Subject: [PATCH] Implement compoundMap Filter()

---
 types/compound_map.go      | 12 +++++++++++-
 types/compound_map_test.go | 28 ++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/types/compound_map.go b/types/compound_map.go
index 90e86ed9db..0958e48092 100644
--- a/types/compound_map.go
+++ b/types/compound_map.go
@@ -122,7 +122,17 @@ func (cm compoundMap) IterAllP(concurrency int, f mapIterAllCallback) {
 }
 
 func (cm compoundMap) Filter(cb mapFilterCallback) Map {
-	panic("Not implemented")
+	seq := newEmptySequenceChunker(makeMapLeafChunkFn(cm.t, cm.cs), newMapMetaSequenceChunkFn(cm.t, cm.cs), newMapLeafBoundaryChecker(), newOrderedMetaSequenceBoundaryChecker)
+
+	// Could conceivably use IterAllP() here.
+	cm.IterAll(func(k, v Value) {
+		if cb(k, v) {
+			seq.Append(mapEntry{k, v})
+		}
+	})
+
+	m := seq.Done()
+	return internalValueFromType(m, m.Type()).(Map)
 }
 
 func (cm compoundMap) Has(key Value) bool {
diff --git a/types/compound_map_test.go b/types/compound_map_test.go
index 91a18b5166..f57d434372 100644
--- a/types/compound_map_test.go
+++ b/types/compound_map_test.go
@@ -322,3 +322,31 @@ func TestCompoundMapRemoveNonexistentKey(t *testing.T) {
 	assert.Equal(original.Len(), actual.Len())
 	assert.True(original.Equals(actual))
 }
+
+func TestCompoundMapFilter(t *testing.T) {
+	assert := assert.New(t)
+
+	doTest := func(tm testMap) {
+		cs := chunks.NewMemoryStore()
+		m := tm.toCompoundMap(cs)
+		sort.Sort(tm)
+		pivotPoint := 10
+		pivot := tm.entries[pivotPoint].key
+		actual := m.Filter(func(k, v Value) bool {
+			return tm.less(k, pivot)
+		})
+		assert.True(newTypedMap(cs, tm.tr, tm.entries[:pivotPoint+1]...).Equals(actual))
+
+		idx := 0
+		actual.IterAll(func(k, v Value) {
+			assert.True(tm.entries[idx].key.Equals(k), "%v != %v", k, tm.entries[idx].key)
+			assert.True(tm.entries[idx].value.Equals(v), "%v != %v", v, tm.entries[idx].value)
+			idx++
+		})
+	}
+
+	doTest(getTestNativeOrderMap(16))
+	doTest(getTestRefValueOrderMap(2))
+	doTest(getTestRefToNativeOrderMap(2))
+	doTest(getTestRefToValueOrderMap(2))
+}