From cbf80370cbcbdd93ce21863feb11038ce653f089 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Mon, 9 Mar 2020 23:09:07 -0500
Subject: [PATCH] support fragment sync for int and decimal fields

This PR adds support for anti-entropy syncing for integer
and decimal fields. It differs from the logic for other
field types in that it does not rely on a consensus to determine
what the value should be; instead, it considers the correct
values to be those of the primary replica. From there, data
is pushed to all non-primary replicas.
---
 api.go                  |  59 ++++++++---
 encoding/proto/proto.go |  10 +-
 field.go                |  25 +++++
 fragment.go             | 156 +++++++++++++++++++++++++---
 handler.go              |  12 ++-
 holder.go               |  19 +---
 holder_test.go          | 130 ++++++++++++++++++-----
 http/handler.go         |   2 +-
 internal/public.pb.go   | 224 ++++++++++++++++++++++++++++------------
 internal/public.proto   |   2 +
 pql/pql.peg.go          |  77 +++++++++-----
 server/handler_test.go  |  26 +++--
 test/holder.go          |  32 +++++-
 13 files changed, 594 insertions(+), 180 deletions(-)

diff --git a/api.go b/api.go
index 08750eb90..92c9b1eb5 100644
--- a/api.go
+++ b/api.go
@@ -319,26 +319,56 @@ func importWorker(importWork chan importJob) {
 	for j := range importWork {
 		err := func() error {
 			for viewName, viewData := range j.req.Views {
+				// The logic here corresponds to the logic in fragment.cleanViewName().
+				// Unfortunately, the logic in that method is not completely exclusive
+				// (i.e. an "other" view named with format YYYMMDD woud be handled
+				// incorrectly). One way to address this would be to change the logic
+				// overall so there weren't conflicts. For now, I'm just going to
+				// rely on the field type to inform the intended view name.
 				if viewName == "" {
 					viewName = viewStandard
-				} else {
+				} else if j.field.Type() == FieldTypeTime {
 					viewName = fmt.Sprintf("%s_%s", viewStandard, viewName)
 				}
 				if len(viewData) == 0 {
 					return fmt.Errorf("no data to import for view: %s", viewName)
 				}
-				fileMagic := uint32(binary.LittleEndian.Uint16(viewData[0:2]))
-				if fileMagic == roaring.MagicNumber { // if pilosa roaring format
-					if err := j.field.importRoaring(j.ctx, viewData, j.shard, viewName, j.req.Clear); err != nil {
-						return errors.Wrap(err, "importing pilosa roaring")
+
+				// TODO: deprecate ImportRoaringRequest.Clear, but
+				// until we do, we need to check its value to provide
+				// backward compatibility.
+				doAction := j.req.Action
+				if doAction == "" {
+					if j.req.Clear {
+						doAction = RequestActionClear
+					} else {
+						doAction = RequestActionSet
 					}
-				} else {
-					// must make a copy of data to operate on locally on standard roaring format.
-					// field.importRoaring changes the standard roaring run format to pilosa roaring
-					data := make([]byte, len(viewData))
-					copy(data, viewData)
-					if err := j.field.importRoaring(j.ctx, data, j.shard, viewName, j.req.Clear); err != nil {
-						return errors.Wrap(err, "importing standard roaring")
+				}
+
+				var doClear bool
+				switch doAction {
+				case RequestActionOverwrite:
+					if err := j.field.importRoaringOverwrite(j.ctx, viewData, j.shard, viewName, j.req.Block); err != nil {
+						return errors.Wrap(err, "importing roaring as overwrite")
+					}
+				case RequestActionClear:
+					doClear = true
+					fallthrough
+				case RequestActionSet:
+					fileMagic := uint32(binary.LittleEndian.Uint16(viewData[0:2]))
+					if fileMagic == roaring.MagicNumber { // if pilosa roaring format
+						if err := j.field.importRoaring(j.ctx, viewData, j.shard, viewName, doClear); err != nil {
+							return errors.Wrap(err, "importing pilosa roaring")
+						}
+					} else {
+						// must make a copy of data to operate on locally on standard roaring format.
+						// field.importRoaring changes the standard roaring run format to pilosa roaring
+						data := make([]byte, len(viewData))
+						copy(data, viewData)
+						if err := j.field.importRoaring(j.ctx, data, j.shard, viewName, doClear); err != nil {
+							return errors.Wrap(err, "importing standard roaring")
+						}
 					}
 				}
 			}
@@ -386,8 +416,9 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
 	}
 
 	// only set and time fields are supported
-	if field.Type() != FieldTypeSet && field.Type() != FieldTypeTime {
-		return NewBadRequestError(errors.Errorf("roaring import is only supported for set and time fields, not '%s' fields.", field.Type()))
+	// TODO: get rid of this (need to confirm other field types)
+	if field.Type() != FieldTypeSet && field.Type() != FieldTypeTime && field.Type() != FieldTypeInt && field.Type() != FieldTypeDecimal {
+		return NewBadRequestError(errors.Errorf("roaring import is only supported for set, time, int, and decimal fields, not '%s' fields.", field.Type()))
 	}
 
 	errCh := make(chan error, len(nodes))
diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go
index ebd2532fd..fc355d732 100644
--- a/encoding/proto/proto.go
+++ b/encoding/proto/proto.go
@@ -418,11 +418,13 @@ func encodeImportRoaringRequest(m *pilosa.ImportRoaringRequest) *internal.Import
 			Name: viewName,
 			Data: viewData,
 		}
-		i += 1
+		i++
 	}
 	return &internal.ImportRoaringRequest{
-		Clear: m.Clear,
-		Views: views,
+		Clear:  m.Clear,
+		Action: m.Action,
+		Block:  uint64(m.Block),
+		Views:  views,
 	}
 }
 
@@ -1071,6 +1073,8 @@ func decodeImportRoaringRequest(pb *internal.ImportRoaringRequest, m *pilosa.Imp
 		views[view.Name] = view.Data
 	}
 	m.Clear = pb.Clear
+	m.Action = pb.Action
+	m.Block = int(pb.Block)
 	m.Views = views
 }
 
diff --git a/field.go b/field.go
index 31f4873f7..85380a4ee 100644
--- a/field.go
+++ b/field.go
@@ -1747,6 +1747,31 @@ func (f *Field) importRoaring(ctx context.Context, data []byte, shard uint64, vi
 	return nil
 }
 
+func (f *Field) importRoaringOverwrite(ctx context.Context, data []byte, shard uint64, viewName string, block int) error {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Field.importRoaringOverwrite")
+	defer span.Finish()
+
+	if viewName == "" {
+		viewName = viewStandard
+	}
+	span.LogKV("view", viewName, "bytes", len(data), "shard", shard)
+	view, err := f.createViewIfNotExists(viewName)
+	if err != nil {
+		return errors.Wrap(err, "creating view")
+	}
+
+	frag, err := view.CreateFragmentIfNotExists(shard)
+	if err != nil {
+		return errors.Wrap(err, "creating fragment")
+	}
+
+	if err := frag.importRoaringOverwrite(ctx, data, block); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 type fieldSlice []*Field
 
 func (p fieldSlice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
diff --git a/fragment.go b/fragment.go
index e0e4743ac..58a868655 100644
--- a/fragment.go
+++ b/fragment.go
@@ -696,7 +696,7 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err
 	return changed, nil
 }
 
-// ClearRow clears a row for a given rowID within the fragment.
+// clearRow clears a row for a given rowID within the fragment.
 // This updates both the on-disk storage and the in-cache bitmap.
 func (f *fragment) clearRow(rowID uint64) (changed bool, err error) {
 	f.mu.Lock()
@@ -738,6 +738,25 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) {
 	return changed, nil
 }
 
+// unprotectedClearBlock clears all rows for a given block.
+// This updates both the on-disk storage and the in-cache bitmap.
+func (f *fragment) unprotectedClearBlock(block int) (changed bool, err error) {
+	firstRow := uint64(block * HashBlockSize)
+	err = f.gen.Transaction(&f.storage.OpWriter, func() error {
+		var rowChanged bool
+		for rowID := uint64(firstRow); rowID < firstRow+HashBlockSize; rowID++ {
+			if changed, err := f.unprotectedClearRow(rowID); err != nil {
+				return errors.Wrapf(err, "clearing row: %d", rowID)
+			} else if changed {
+				rowChanged = true
+			}
+		}
+		changed = rowChanged
+		return nil
+	})
+	return changed, err
+}
+
 func (f *fragment) bit(rowID, columnID uint64) (bool, error) {
 	pos, err := f.pos(rowID, columnID)
 	if err != nil {
@@ -1733,7 +1752,7 @@ func (f *fragment) blockData(id int) (rowIDs, columnIDs []uint64) {
 // The state of a bit is determined by consensus from all blocks being considered.
 //
 // For example, if 3 blocks are compared and two have a set bit and one has a
-// cleared bit then the bit is considered cleared. The function returns the
+// cleared bit then the bit is considered set. The function returns the
 // diff per incoming block so that all can be in sync.
 func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, err error) {
 	// Ensure that all pair sets are of equal length.
@@ -2122,14 +2141,19 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 // https://github.com/RoaringBitmap/RoaringFormatSpec or from pilosa's version
 // of the roaring format. The cache is updated to reflect the new data.
 func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) error {
-	rowSize := uint64(1 << shardVsContainerExponent)
 	span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring")
 	defer span.Finish()
 	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock")
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	span.Finish()
-	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
+
+	return f.unprotectedImportRoaring(ctx, data, clear)
+}
+
+func (f *fragment) unprotectedImportRoaring(ctx context.Context, data []byte, clear bool) error {
+	rowSize := uint64(1 << shardVsContainerExponent)
+	span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
 	var changed int
 	var rowSet map[uint64]int
 	err := f.gen.Transaction(&f.storage.OpWriter, func() (err error) {
@@ -2175,6 +2199,20 @@ func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) e
 	return nil
 }
 
+// importRoaringOverwrite overwrites the specified block with the provided data.
+func (f *fragment) importRoaringOverwrite(ctx context.Context, data []byte, block int) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	// Clear the existing data from fragment block.
+	if _, err := f.unprotectedClearBlock(block); err != nil {
+		return errors.Wrapf(err, "clearing block: %d", block)
+	}
+
+	// Union the new block data with the fragment data.
+	return f.unprotectedImportRoaring(ctx, data, false)
+}
+
 // incrementOpN increase the operation count by one.
 // If the count exceeds the maximum allowed then a snapshot is performed.
 func (f *fragment) incrementOpN(changed int) {
@@ -2611,6 +2649,19 @@ func (f *fragment) unprotectedRows(start uint64, filters ...rowFilter) []uint64
 	return rows
 }
 
+// blockToRoaringData converts a fragment block into a roaring.Bitmap
+// which represents a portion of the data within a single shard.
+// TODO: it seems like we should be able to get the
+// block data as roaring without having to go through
+// this rows/columns step.
+func (f *fragment) blockToRoaringData(block int) ([]byte, error) {
+	rowIDs, columnIDs := f.blockData(block)
+	return bitsToRoaringData(pairSet{
+		columnIDs: columnIDs,
+		rowIDs:    rowIDs,
+	})
+}
+
 // upgradeRoaringBSIv2 upgrades a fragment that contains old BSI formatting
 // to a new BSI format (v2). The new format moves the "exists" bit to the
 // beginning & adds a negative sign bit.
@@ -2731,6 +2782,9 @@ type fragmentSyncer struct {
 	Node    *Node
 	Cluster *cluster
 
+	// FieldType helps determine which method of syncing to use.
+	FieldType string
+
 	Closing <-chan struct{}
 }
 
@@ -2756,6 +2810,16 @@ func (s *fragmentSyncer) syncFragment() error {
 		return nil
 	}
 
+	// This is here solely to prevent unnecessary work;
+	// if this node isn't the primary replica, there's no need
+	// to continue processing int/decimal fields.
+	if nodes[0].ID != s.Node.ID {
+		switch s.FieldType {
+		case FieldTypeInt, FieldTypeDecimal:
+			return nil
+		}
+	}
+
 	// Create a set of blocks.
 	blockSets := make([][]FragmentBlock, 0, len(nodes))
 	for _, node := range nodes {
@@ -2814,11 +2878,79 @@ func (s *fragmentSyncer) syncFragment() error {
 		if byteSlicesEqual(checksums) {
 			continue
 		}
-		// Synchronize block.
-		if err := s.syncBlock(blockID); err != nil {
-			return fmt.Errorf("sync block: id=%d, err=%s", blockID, err)
+
+		// If we've gotten here, it means that the block differs
+		// between nodes. If this particular fragment is part of an
+		// `int` or `decimal` field, then instead of using a consensus
+		// to determine which bits to update, we consider the primary
+		// replica to be correct, and overwrite the non-primary replicas
+		// with the primary's data.
+		switch s.FieldType {
+		case FieldTypeInt, FieldTypeDecimal:
+			// Synchronize block from the primary replica.
+			if err := s.syncBlockFromPrimary(blockID); err != nil {
+				return fmt.Errorf("sync block from primary: id=%d, err=%s", blockID, err)
+			}
+			s.Fragment.stats.Count("BlockRepairPrimary", 1, 1.0)
+		default:
+			// Synchronize block.
+			if err := s.syncBlock(blockID); err != nil {
+				return fmt.Errorf("sync block: id=%d, err=%s", blockID, err)
+			}
+			s.Fragment.stats.Count("BlockRepair", 1, 1.0)
+		}
+	}
+
+	return nil
+}
+
+// syncBlockFromPrimary sends all rows for a given block
+// from the primary replica to non-primary replicas.
+// Since this is pushing updates out to replicas, it only
+// runs on the primary replica.
+// Returns an error if any remote hosts are unreachable.
+func (s *fragmentSyncer) syncBlockFromPrimary(id int) error {
+	span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncBlockFromPrimary")
+	defer span.Finish()
+
+	f := s.Fragment
+
+	// Determine replica set. Return early if this is not
+	// the primary node.
+	nodes := s.Cluster.shardNodes(f.index, f.shard)
+	if s.Node.ID != nodes[0].ID {
+		f.Logger.Debugf("non-primary replica expecting sync from primary: %s, index=%s, field=%s, shard=%d", nodes[0].ID, f.index, f.field, f.shard)
+		return nil
+	}
+
+	// Get the local block represented as roaring data.
+	localData, err := f.blockToRoaringData(id)
+	if err != nil {
+		return errors.Wrap(err, "converting block to roaring data")
+	}
+
+	// Verify sync is not prematurely closing.
+	if s.isClosing() {
+		return nil
+	}
+
+	// Create the overwrite request to be sent to non-primary replicas.
+	overwriteReq := &ImportRoaringRequest{
+		Action: RequestActionOverwrite,
+		Block:  id,
+		Views:  map[string][]byte{cleanViewName(f.view): localData},
+	}
+
+	// Write updates to remote blocks.
+	for _, node := range nodes {
+		if s.Node.ID == node.ID {
+			continue
+		}
+
+		uri := &node.URI
+		if err := s.Cluster.InternalClient.ImportRoaring(ctx, uri, f.index, f.field, f.shard, true, overwriteReq); err != nil {
+			return errors.Wrap(err, "sending roaring data (overwrite)")
 		}
-		s.Fragment.stats.Count("BlockRepair", 1, 1.0)
 	}
 
 	return nil
@@ -2883,8 +3015,8 @@ func (s *fragmentSyncer) syncBlock(id int) error {
 			}
 
 			setReq := &ImportRoaringRequest{
-				Clear: false,
-				Views: map[string][]byte{cleanViewName(f.view): setData},
+				Action: RequestActionSet,
+				Views:  map[string][]byte{cleanViewName(f.view): setData},
 			}
 
 			if err := s.Cluster.InternalClient.ImportRoaring(ctx, uris[i], f.index, f.field, f.shard, true, setReq); err != nil {
@@ -2900,8 +3032,8 @@ func (s *fragmentSyncer) syncBlock(id int) error {
 			}
 
 			clearReq := &ImportRoaringRequest{
-				Clear: true,
-				Views: map[string][]byte{"": clearData},
+				Action: RequestActionClear,
+				Views:  map[string][]byte{cleanViewName(f.view): clearData},
 			}
 
 			if err := s.Cluster.InternalClient.ImportRoaring(ctx, uris[i], f.index, f.field, f.shard, true, clearReq); err != nil {
diff --git a/handler.go b/handler.go
index e89791268..b726c6dff 100644
--- a/handler.go
+++ b/handler.go
@@ -184,11 +184,19 @@ type ImportRequest struct {
 	Timestamps []int64
 }
 
+const (
+	RequestActionSet       = "set"
+	RequestActionClear     = "clear"
+	RequestActionOverwrite = "overwrite"
+)
+
 // ImportRoaringRequest describes the import request structure
 // for an import containing roaring-encoded data.
 type ImportRoaringRequest struct {
-	Clear bool
-	Views map[string][]byte
+	Clear  bool
+	Action string // [set, clear, overwrite]
+	Block  int
+	Views  map[string][]byte
 }
 
 // ImportResponse is the structured response of an import.
diff --git a/holder.go b/holder.go
index dc58a8ab1..48796da8e 100644
--- a/holder.go
+++ b/holder.go
@@ -911,16 +911,6 @@ func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) err
 		return ErrFieldNotFound
 	}
 
-	// TODO: this is a temporary fix put in place to prevent
-	// the syncer from trying to sync replicas of fields
-	// other than `set` or `time` (i.e. `int`, `mutex`, `bool`,
-	// `decimal`) using ImportRoaring, because ImportRoaring
-	// only supports `set` and `time` fields.
-	if f.Type() != FieldTypeSet && f.Type() != FieldTypeTime {
-		s.Holder.Logger.Printf("temporarily skipping fragment sync: %s/%d", field, shard)
-		return nil
-	}
-
 	// Ensure view exists locally.
 	v, err := f.createViewIfNotExists(view)
 	if err != nil {
@@ -935,10 +925,11 @@ func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) err
 
 	// Sync fragments together.
 	fs := fragmentSyncer{
-		Fragment: frag,
-		Node:     s.Node,
-		Cluster:  s.Cluster,
-		Closing:  s.Closing,
+		Fragment:  frag,
+		Node:      s.Node,
+		Cluster:   s.Cluster,
+		FieldType: f.Type(),
+		Closing:   s.Closing,
 	}
 	if err := fs.syncFragment(); err != nil {
 		return errors.Wrap(err, "syncing fragment")
diff --git a/holder_test.go b/holder_test.go
index f88c39e0e..cc14798fa 100644
--- a/holder_test.go
+++ b/holder_test.go
@@ -17,6 +17,7 @@ package pilosa_test
 import (
 	"bytes"
 	"context"
+	"math"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -25,6 +26,7 @@ import (
 	"time"
 
 	"github.com/pilosa/pilosa/v2"
+	"github.com/pilosa/pilosa/v2/pql"
 	"github.com/pilosa/pilosa/v2/test"
 	"github.com/pkg/errors"
 )
@@ -526,37 +528,109 @@ func TestHolderSyncer_TimeQuantum(t *testing.T) {
 
 // Ensure holder can sync integer views with a remote holder.
 func TestHolderSyncer_IntField(t *testing.T) {
-	c := test.MustNewCluster(t, 2)
-	c[0].Config.Cluster.ReplicaN = 2
-	c[0].Config.AntiEntropy.Interval = 0
-	c[1].Config.Cluster.ReplicaN = 2
-	c[1].Config.AntiEntropy.Interval = 0
-	err := c.Start()
-	if err != nil {
-		t.Fatalf("starting cluster: %v", err)
-	}
-	defer c.Close()
+	t.Run("BasicSync", func(t *testing.T) {
+		c := test.MustNewCluster(t, 2)
+		c[0].Config.Cluster.ReplicaN = 2
+		c[0].Config.AntiEntropy.Interval = 0
+		c[1].Config.Cluster.ReplicaN = 2
+		c[1].Config.AntiEntropy.Interval = 0
+		err := c.Start()
+		if err != nil {
+			t.Fatalf("starting cluster: %v", err)
+		}
+		defer c.Close()
 
-	_, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
-	if err != nil {
-		t.Fatalf("creating index i: %v", err)
-	}
-	_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeInt(0, 100))
-	if err != nil {
-		t.Fatalf("creating field f: %v", err)
-	}
+		_, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
+		if err != nil {
+			t.Fatalf("creating index i: %v", err)
+		}
+		_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeInt(0, 100))
+		if err != nil {
+			t.Fatalf("creating field f: %v", err)
+		}
 
-	hldr0 := &test.Holder{Holder: c[0].Server.Holder()}
-	hldr1 := &test.Holder{Holder: c[1].Server.Holder()}
+		hldr0 := &test.Holder{Holder: c[0].Server.Holder()}
+		hldr1 := &test.Holder{Holder: c[1].Server.Holder()}
 
-	// Set data on the local holder for node0.
-	hldr0.SetValue("i", "f", 1, 1)
+		// Set data on the local holder for node0.
+		hldr0.SetValue("i", "f", 1, 1)
 
-	// Set data on node1.
-	hldr1.SetValue("i", "f", 2, 2)
+		// Set data on node1.
+		hldr1.SetValue("i", "f", 2, 2)
 
-	err = c[0].Server.SyncData()
-	if err != nil {
-		t.Fatalf("syncing node 0: %v", err)
-	}
+		err = c[0].Server.SyncData()
+		if err != nil {
+			t.Fatalf("syncing node 0: %v", err)
+		}
+
+		// Verify data is the same on both nodes.
+		for i, hldr := range []*test.Holder{hldr0, hldr1} {
+			if a, exists := hldr.Value("i", "f", 1); !exists || a != 1 {
+				t.Errorf("unexpected value(node%d/0): %d, exists: %v", i, a, exists)
+			}
+			if a, exists := hldr.Value("i", "f", 2); exists {
+				t.Errorf("unexpected value(node%d/1): %d, exists: %v", i, a, exists)
+			}
+		}
+	})
+
+	t.Run("MultiShard", func(t *testing.T) {
+		c := test.MustNewCluster(t, 2)
+		c[0].Config.Cluster.ReplicaN = 2
+		c[0].Config.AntiEntropy.Interval = 0
+		c[1].Config.Cluster.ReplicaN = 2
+		c[1].Config.AntiEntropy.Interval = 0
+		err := c.Start()
+		if err != nil {
+			t.Fatalf("starting cluster: %v", err)
+		}
+		defer c.Close()
+
+		_, err = c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{})
+		if err != nil {
+			t.Fatalf("creating index i: %v", err)
+		}
+		_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
+		if err != nil {
+			t.Fatalf("creating field f: %v", err)
+		}
+
+		hldr0 := &test.Holder{Holder: c[0].Server.Holder()}
+		hldr1 := &test.Holder{Holder: c[1].Server.Holder()}
+
+		// Set data on the local holder for node0.
+		hldr0.SetValue("i", "f", 1*pilosa.ShardWidth, 11)
+		hldr0.SetValue("i", "f", 3*pilosa.ShardWidth, 32)
+		hldr0.SetValue("i", "f", 4*pilosa.ShardWidth, math.MinInt32)
+		hldr0.SetValue("i", "f", 7*pilosa.ShardWidth, math.MinInt32)
+
+		// Set data on node1.
+		hldr1.SetValue("i", "f", 0*pilosa.ShardWidth, 2)
+		hldr1.SetValue("i", "f", 2*pilosa.ShardWidth, 22)
+		hldr1.SetValue("i", "f", 4*pilosa.ShardWidth, math.MaxInt32)
+		hldr1.SetValue("i", "f", 7*pilosa.ShardWidth, math.MaxInt32)
+
+		// Primary for shards (for index "i"):
+		// node0: [0,3,7]
+		// node1: [1,2,4]
+
+		err = c[0].Server.SyncData()
+		if err != nil {
+			t.Fatalf("syncing node 0: %v", err)
+		}
+		err = c[1].Server.SyncData()
+		if err != nil {
+			t.Fatalf("syncing node 1: %v", err)
+		}
+
+		// Verify data is the same on both nodes.
+		for i, hldr := range []*test.Holder{hldr0, hldr1} {
+			if a := hldr.Range("i", "f", pql.GT, 0); !reflect.DeepEqual(a.Columns(), []uint64{2 * pilosa.ShardWidth, 3 * pilosa.ShardWidth, 4 * pilosa.ShardWidth}) {
+				t.Errorf("unexpected columns(node%d/0): %d", i, a.Columns())
+			}
+			if a := hldr.Range("i", "f", pql.LT, 0); !reflect.DeepEqual(a.Columns(), []uint64{7 * pilosa.ShardWidth}) {
+				t.Errorf("unexpected columns(node%d/0): %d", i, a.Columns())
+			}
+		}
+	})
 }
diff --git a/http/handler.go b/http/handler.go
index dbb2c4e61..32808cb97 100644
--- a/http/handler.go
+++ b/http/handler.go
@@ -1174,7 +1174,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	} else {
-		// Field type: Set, Time
+		// Field type: set, time, mutex
 		// Marshal into request object.
 		req := &pilosa.ImportRequest{}
 		if err := h.api.Serializer.Unmarshal(body, req); err != nil {
diff --git a/internal/public.pb.go b/internal/public.pb.go
index 8c1e33061..d4d99e51c 100644
--- a/internal/public.pb.go
+++ b/internal/public.pb.go
@@ -1532,6 +1532,8 @@ func (m *ImportRoaringRequestView) GetData() []byte {
 type ImportRoaringRequest struct {
 	Clear                bool                        `protobuf:"varint,1,opt,name=Clear,proto3" json:"Clear,omitempty"`
 	Views                []*ImportRoaringRequestView `protobuf:"bytes,2,rep,name=views,proto3" json:"views,omitempty"`
+	Action               string                      `protobuf:"bytes,3,opt,name=Action,proto3" json:"Action,omitempty"`
+	Block                uint64                      `protobuf:"varint,4,opt,name=Block,proto3" json:"Block,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}                    `json:"-"`
 	XXX_unrecognized     []byte                      `json:"-"`
 	XXX_sizecache        int32                       `json:"-"`
@@ -1584,6 +1586,20 @@ func (m *ImportRoaringRequest) GetViews() []*ImportRoaringRequestView {
 	return nil
 }
 
+func (m *ImportRoaringRequest) GetAction() string {
+	if m != nil {
+		return m.Action
+	}
+	return ""
+}
+
+func (m *ImportRoaringRequest) GetBlock() uint64 {
+	if m != nil {
+		return m.Block
+	}
+	return 0
+}
+
 type ImportColumnAttrsRequest struct {
 	Index                string   `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"`
 	Shard                int64    `protobuf:"varint,2,opt,name=Shard,proto3" json:"Shard,omitempty"`
@@ -1693,76 +1709,78 @@ func init() {
 func init() { proto.RegisterFile("public.proto", fileDescriptor_413a91106d7bcce8) }
 
 var fileDescriptor_413a91106d7bcce8 = []byte{
-	// 1103 bytes of a gzipped FileDescriptorProto
+	// 1127 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x56, 0xcd, 0x8e, 0x1b, 0x45,
 	0x10, 0xa6, 0x3d, 0xe3, 0xb5, 0x5d, 0xf6, 0x2e, 0x51, 0xc7, 0x09, 0x23, 0x14, 0x6d, 0xac, 0x56,
-	0x84, 0x0c, 0x87, 0x8d, 0x12, 0x10, 0xca, 0x89, 0x9f, 0x8d, 0x37, 0x60, 0x45, 0xb1, 0x42, 0x7b,
-	0x65, 0x6e, 0x48, 0xb3, 0x71, 0xc7, 0x19, 0x69, 0x3c, 0x63, 0xe6, 0x07, 0x67, 0x9f, 0x83, 0x0b,
-	0xe2, 0x09, 0x78, 0x07, 0x5e, 0x80, 0x23, 0x8f, 0x00, 0xcb, 0x1b, 0x70, 0xe5, 0x82, 0xaa, 0x7a,
-	0xda, 0xdd, 0xe3, 0xf5, 0x2e, 0x51, 0xc4, 0xad, 0xfe, 0xba, 0xba, 0xbe, 0xea, 0xea, 0xaf, 0x1b,
-	0x7a, 0xab, 0xf2, 0x2c, 0x8e, 0x5e, 0x1c, 0xad, 0xb2, 0xb4, 0x48, 0x79, 0x3b, 0x4a, 0x0a, 0x95,
-	0x25, 0x61, 0x2c, 0x72, 0xf0, 0x64, 0xba, 0xe6, 0x01, 0xb4, 0x1e, 0xa7, 0x71, 0xb9, 0x4c, 0xf2,
-	0x80, 0x0d, 0xbc, 0xa1, 0x2f, 0x8d, 0xca, 0x39, 0xf8, 0x4f, 0xd5, 0x79, 0x1e, 0x78, 0x03, 0x6f,
-	0xd8, 0x91, 0x24, 0xf3, 0x7b, 0xd0, 0xfc, 0xb2, 0x28, 0xb2, 0x3c, 0x68, 0x0c, 0xbc, 0x61, 0xf7,
-	0xe1, 0xc1, 0x91, 0x49, 0x77, 0x84, 0x66, 0xa9, 0x9d, 0x98, 0x53, 0xa6, 0x61, 0x16, 0x25, 0x8b,
-	0xc0, 0x1f, 0xb0, 0x61, 0x4f, 0x1a, 0x55, 0x3c, 0x83, 0xce, 0x34, 0x5a, 0x24, 0x6a, 0x8e, 0x5b,
-	0xdf, 0x05, 0xef, 0x79, 0x8a, 0xdb, 0xb2, 0x61, 0xf7, 0xe1, 0xbe, 0x4d, 0x25, 0xd3, 0xb5, 0x44,
-	0x0f, 0x06, 0x4c, 0xd4, 0x22, 0x68, 0xec, 0x0c, 0x98, 0xa8, 0x85, 0x78, 0x04, 0x07, 0x32, 0x5d,
-	0x8f, 0xe7, 0x2a, 0x29, 0xa2, 0x97, 0x91, 0xca, 0xa8, 0x68, 0x99, 0xae, 0x0d, 0x16, 0x92, 0x37,
-	0x40, 0x1a, 0x16, 0x88, 0xf8, 0x0c, 0xfc, 0xe7, 0x61, 0x94, 0xf1, 0x03, 0x68, 0x8c, 0x47, 0x54,
-	0x82, 0x2f, 0x1b, 0xe3, 0x11, 0xbf, 0x01, 0xde, 0x53, 0x75, 0x1e, 0x78, 0x03, 0x36, 0xec, 0x48,
-	0x14, 0x79, 0x1f, 0x9a, 0x8f, 0xd3, 0x32, 0x29, 0xa8, 0x0c, 0x5f, 0x6a, 0x45, 0x9c, 0x40, 0x07,
-	0xd7, 0x3f, 0x89, 0x54, 0x3c, 0xe7, 0x42, 0x27, 0xab, 0x90, 0x38, 0x4d, 0x41, 0xab, 0xd4, 0x1b,
-	0xf5, 0xa1, 0x49, 0xc1, 0x94, 0xa6, 0x23, 0xb5, 0x22, 0xbe, 0x06, 0x40, 0x6f, 0xae, 0xf3, 0xdc,
-	0x83, 0x26, 0x69, 0x54, 0xfd, 0xe5, 0x44, 0xda, 0x79, 0x45, 0xa6, 0x09, 0xb4, 0x49, 0xc0, 0xc6,
-	0x6e, 0x22, 0x98, 0x13, 0x81, 0x56, 0x6c, 0xd6, 0xc8, 0x00, 0x21, 0x85, 0xdf, 0x86, 0x3d, 0x99,
-	0xae, 0x2d, 0xe6, 0x4a, 0x13, 0xdf, 0x01, 0x7c, 0x95, 0xa5, 0xe5, 0x8a, 0xe0, 0xf2, 0x21, 0x34,
-	0x49, 0xab, 0x2a, 0xe3, 0xb6, 0x32, 0xb3, 0xa9, 0xd4, 0x01, 0xbb, 0xdb, 0x85, 0x6d, 0x9d, 0x96,
-	0x4b, 0xda, 0xc2, 0x93, 0x28, 0x62, 0xbd, 0xb3, 0x30, 0xde, 0x78, 0x67, 0x61, 0x4c, 0xd5, 0x7a,
-	0x12, 0xc5, 0x7a, 0x16, 0xcf, 0x64, 0x79, 0x1f, 0xda, 0x4f, 0xe2, 0x34, 0x2c, 0x30, 0x18, 0x53,
-	0x31, 0xb9, 0xd1, 0xc5, 0xb7, 0xb0, 0xaf, 0x07, 0x17, 0x47, 0x70, 0xaa, 0x8a, 0x37, 0x38, 0xd9,
-	0x37, 0x1a, 0x66, 0xf1, 0x0b, 0x03, 0x1f, 0x25, 0x93, 0x80, 0xd9, 0x04, 0x1c, 0xfc, 0xd3, 0xf3,
-	0x95, 0xaa, 0xa0, 0x92, 0xcc, 0x07, 0xd0, 0x9d, 0x16, 0x38, 0xeb, 0xb3, 0x30, 0x2e, 0x55, 0xb5,
-	0x9d, 0x6b, 0x42, 0x14, 0xe3, 0xa4, 0xd0, 0x6e, 0x9f, 0xe0, 0x6d, 0x74, 0x7e, 0x07, 0x3a, 0xc7,
-	0x69, 0x1a, 0x6b, 0x67, 0x73, 0xc0, 0x86, 0x6d, 0x69, 0x0d, 0xfc, 0x10, 0xc0, 0xe0, 0x2d, 0x55,
-	0xb0, 0x47, 0x1d, 0x70, 0x2c, 0xe2, 0x3e, 0xb4, 0xb0, 0xd2, 0x67, 0xe1, 0xca, 0x62, 0x63, 0xd7,
-	0x61, 0xfb, 0x87, 0x41, 0xef, 0x9b, 0x52, 0x65, 0xe7, 0x52, 0x7d, 0x5f, 0xaa, 0xbc, 0xc0, 0xbe,
-	0x93, 0x6e, 0x26, 0x87, 0x14, 0x9c, 0x91, 0xe9, 0xab, 0x30, 0x9b, 0xeb, 0x4e, 0xf9, 0xb2, 0xd2,
-	0x10, 0xab, 0xed, 0x79, 0x4e, 0x58, 0xdb, 0xd2, 0x35, 0xd1, 0x74, 0xa9, 0x65, 0x5a, 0x18, 0x30,
-	0x95, 0xc6, 0x87, 0xf0, 0xee, 0xc9, 0xeb, 0x17, 0x71, 0x39, 0x57, 0x32, 0x5d, 0xeb, 0xd5, 0x7b,
-	0x14, 0xb0, 0x6d, 0xe6, 0x1f, 0xc0, 0x41, 0x65, 0x32, 0x34, 0xd5, 0xa2, 0xc0, 0x2d, 0x2b, 0x7f,
-	0x00, 0xbd, 0x93, 0xe5, 0x99, 0x9a, 0xcf, 0xd5, 0x7c, 0x14, 0x16, 0x61, 0xd0, 0x26, 0xdc, 0x5b,
-	0xa4, 0x51, 0x0b, 0x11, 0x3f, 0x32, 0xd8, 0xaf, 0xd0, 0xe7, 0xab, 0x34, 0xc9, 0x15, 0x1e, 0xf1,
-	0x49, 0x96, 0x99, 0x23, 0x3e, 0xc9, 0x32, 0x7e, 0x1f, 0x5a, 0x52, 0xe5, 0x65, 0x5c, 0x98, 0x29,
-	0xb9, 0x65, 0x33, 0x9a, 0xb5, 0x65, 0x5c, 0x48, 0x13, 0xc5, 0x3f, 0x87, 0x83, 0xda, 0x1c, 0x6a,
-	0xfe, 0xec, 0x3e, 0x7c, 0xcf, 0xae, 0xab, 0xf9, 0xe5, 0x56, 0xb8, 0xf8, 0xd5, 0x83, 0xae, 0x93,
-	0x79, 0x33, 0x64, 0xd8, 0x9f, 0xfd, 0x6a, 0xc8, 0xee, 0x12, 0x77, 0x5f, 0xc1, 0x9c, 0xc8, 0x00,
-	0x3d, 0x60, 0x93, 0x6a, 0x2c, 0xd9, 0xc4, 0xf2, 0x8a, 0x77, 0x1d, 0xaf, 0xe0, 0x4b, 0xf0, 0x2a,
-	0x4c, 0x16, 0x6a, 0x4e, 0x63, 0xd9, 0x96, 0x46, 0xe5, 0x47, 0xf6, 0xae, 0xd2, 0x39, 0xd6, 0x08,
-	0xc0, 0x78, 0xa4, 0xbd, 0xcf, 0x9a, 0x53, 0xc6, 0x23, 0x3c, 0x2b, 0x9a, 0x17, 0xad, 0xf1, 0x4f,
-	0xa1, 0x6b, 0x39, 0x25, 0xaf, 0x8e, 0xa8, 0x6f, 0x53, 0x59, 0xa7, 0x74, 0x03, 0xf9, 0x17, 0xdb,
-	0x34, 0x1f, 0x74, 0xa8, 0x8a, 0xa0, 0x86, 0xdc, 0xf1, 0xcb, 0xed, 0x67, 0xe1, 0x81, 0xf3, 0xee,
-	0x04, 0x40, 0x8b, 0x6f, 0xda, 0xc5, 0x1b, 0x97, 0x74, 0x5e, 0xa7, 0x4f, 0x5c, 0x6a, 0x0e, 0xba,
-	0xb4, 0xa6, 0x5f, 0xef, 0x9c, 0xf6, 0x49, 0x27, 0x4e, 0xfc, 0xc9, 0x60, 0x7f, 0xbc, 0x5c, 0xa5,
-	0x59, 0xe1, 0x5c, 0xa9, 0x71, 0x32, 0x57, 0xaf, 0xcd, 0x95, 0x22, 0x65, 0x37, 0x89, 0xa3, 0x95,
-	0xae, 0x16, 0x5d, 0x25, 0x5f, 0x6a, 0xc5, 0x69, 0xa7, 0x5f, 0x6b, 0xe7, 0x1d, 0xe8, 0xe8, 0xd9,
-	0x41, 0x57, 0x93, 0x5c, 0xd6, 0xa0, 0x1f, 0xe1, 0x35, 0x3d, 0x7c, 0x2d, 0x7a, 0xf8, 0x8c, 0x8a,
-	0x34, 0xa2, 0xc3, 0xc8, 0xd9, 0x26, 0xa7, 0x63, 0x41, 0xff, 0x69, 0xb4, 0x54, 0x79, 0x11, 0x2e,
-	0x57, 0x78, 0x2f, 0xbd, 0xa1, 0x27, 0x1d, 0x8b, 0xf8, 0x9b, 0x01, 0xd7, 0x18, 0x89, 0x76, 0xfe,
-	0x3f, 0xa0, 0xd7, 0x03, 0xaa, 0x97, 0xdd, 0xba, 0x54, 0xf6, 0x6d, 0xd8, 0xa3, 0x7a, 0x4c, 0xc9,
-	0x95, 0x86, 0x2c, 0x65, 0x39, 0x52, 0xe3, 0x65, 0xd2, 0x35, 0x71, 0x01, 0x3d, 0x87, 0xa0, 0x71,
-	0xba, 0x30, 0x77, 0xcd, 0x26, 0x66, 0xd0, 0x3f, 0xcd, 0xc2, 0x24, 0x8f, 0xc3, 0x42, 0xe1, 0x76,
-	0x6f, 0x83, 0x7a, 0xc7, 0x8f, 0x4a, 0x7c, 0x08, 0xb7, 0xb6, 0xf2, 0x5a, 0x2e, 0xc2, 0x36, 0x78,
-	0xd4, 0x06, 0x14, 0xc5, 0x14, 0x6e, 0x6e, 0x42, 0xc7, 0xa3, 0xb7, 0xaa, 0xe0, 0x72, 0xd2, 0x8f,
-	0x1c, 0x5c, 0x94, 0xb4, 0xda, 0x7e, 0x57, 0xad, 0xc7, 0x10, 0x54, 0xb3, 0xad, 0xbf, 0x73, 0x55,
-	0x05, 0xb3, 0x48, 0xad, 0x31, 0x7e, 0x12, 0x2e, 0x55, 0x55, 0x04, 0xc9, 0x68, 0x23, 0x2e, 0x6e,
-	0xd0, 0x27, 0x90, 0x64, 0xf1, 0x12, 0xfa, 0xbb, 0x72, 0xd0, 0x8b, 0x1f, 0xab, 0x50, 0x93, 0x6f,
-	0x5b, 0x6a, 0x85, 0x3f, 0x82, 0xe6, 0x0f, 0x91, 0x5a, 0x1b, 0xf2, 0x15, 0xf6, 0xfe, 0x5d, 0x55,
-	0x88, 0xd4, 0x0b, 0xc4, 0xcf, 0xcc, 0x14, 0xeb, 0xbc, 0x47, 0xff, 0xd9, 0x32, 0x3d, 0x94, 0xd5,
-	0xa7, 0x43, 0x0f, 0x65, 0xa0, 0x1f, 0x55, 0xfb, 0x77, 0x30, 0x2a, 0x3e, 0xe4, 0x28, 0xce, 0xc2,
-	0x58, 0xdf, 0xcc, 0x8e, 0xdc, 0xe8, 0xd7, 0x8f, 0xf2, 0xf1, 0x8d, 0xdf, 0x2e, 0x0e, 0xd9, 0xef,
-	0x17, 0x87, 0xec, 0x8f, 0x8b, 0x43, 0xf6, 0xd3, 0x5f, 0x87, 0xef, 0x9c, 0xed, 0xd1, 0xf7, 0xfc,
-	0xe3, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0xe1, 0xfe, 0xaf, 0x13, 0xae, 0x0b, 0x00, 0x00,
+	0x84, 0x0c, 0x87, 0x8d, 0x12, 0x10, 0xca, 0x09, 0xc8, 0xc6, 0x1b, 0xb0, 0xa2, 0x58, 0xa1, 0xbd,
+	0x32, 0x37, 0xa4, 0xd9, 0x75, 0xb3, 0x19, 0x31, 0x9e, 0x36, 0xf3, 0x83, 0xb3, 0xcf, 0x01, 0x07,
+	0xc4, 0x13, 0xf0, 0x0e, 0xbc, 0x00, 0x47, 0x1e, 0x01, 0x96, 0x37, 0xe0, 0xca, 0x05, 0x55, 0xf5,
+	0xb4, 0x7b, 0xc6, 0xeb, 0x5d, 0xa2, 0x88, 0x5b, 0x7d, 0x55, 0xd5, 0xd5, 0xf5, 0x55, 0x57, 0x57,
+	0x37, 0xf4, 0x96, 0xc5, 0x49, 0x1c, 0x9d, 0x1e, 0x2c, 0x53, 0x9d, 0x6b, 0xde, 0x8e, 0x92, 0x5c,
+	0xa5, 0x49, 0x18, 0x8b, 0x0c, 0x3c, 0xa9, 0x57, 0x3c, 0x80, 0xd6, 0x13, 0x1d, 0x17, 0x8b, 0x24,
+	0x0b, 0xd8, 0xc0, 0x1b, 0xfa, 0xd2, 0x42, 0xce, 0xc1, 0x7f, 0xa6, 0xce, 0xb3, 0xc0, 0x1b, 0x78,
+	0xc3, 0x8e, 0x24, 0x99, 0xdf, 0x83, 0xe6, 0xe3, 0x3c, 0x4f, 0xb3, 0xa0, 0x31, 0xf0, 0x86, 0xdd,
+	0x87, 0x7b, 0x07, 0x36, 0xdc, 0x01, 0xaa, 0xa5, 0x31, 0x62, 0x4c, 0xa9, 0xc3, 0x34, 0x4a, 0xce,
+	0x02, 0x7f, 0xc0, 0x86, 0x3d, 0x69, 0xa1, 0x78, 0x0e, 0x9d, 0x69, 0x74, 0x96, 0xa8, 0x39, 0x6e,
+	0x7d, 0x17, 0xbc, 0x17, 0x1a, 0xb7, 0x65, 0xc3, 0xee, 0xc3, 0x5d, 0x17, 0x4a, 0xea, 0x95, 0x44,
+	0x0b, 0x3a, 0x4c, 0xd4, 0x59, 0xd0, 0xd8, 0xea, 0x30, 0x51, 0x67, 0xe2, 0x11, 0xec, 0x49, 0xbd,
+	0x1a, 0xcf, 0x55, 0x92, 0x47, 0xdf, 0x44, 0x2a, 0xa5, 0xa4, 0xa5, 0x5e, 0x59, 0x2e, 0x24, 0xaf,
+	0x89, 0x34, 0x1c, 0x11, 0xf1, 0x09, 0xf8, 0x2f, 0xc2, 0x28, 0xe5, 0x7b, 0xd0, 0x18, 0x8f, 0x28,
+	0x05, 0x5f, 0x36, 0xc6, 0x23, 0x7e, 0x03, 0xbc, 0x67, 0xea, 0x3c, 0xf0, 0x06, 0x6c, 0xd8, 0x91,
+	0x28, 0xf2, 0x3e, 0x34, 0x9f, 0xe8, 0x22, 0xc9, 0x29, 0x0d, 0x5f, 0x1a, 0x20, 0x8e, 0xa0, 0x83,
+	0xeb, 0x9f, 0x46, 0x2a, 0x9e, 0x73, 0x61, 0x82, 0x95, 0x4c, 0x2a, 0x45, 0x41, 0xad, 0x34, 0x1b,
+	0xf5, 0xa1, 0x49, 0xce, 0x14, 0xa6, 0x23, 0x0d, 0x10, 0x5f, 0x00, 0xa0, 0x35, 0x33, 0x71, 0xee,
+	0x41, 0x93, 0x10, 0x65, 0x7f, 0x39, 0x90, 0x31, 0x5e, 0x11, 0x69, 0x02, 0x6d, 0x12, 0xb0, 0xb0,
+	0x6b, 0x0f, 0x56, 0xf1, 0x40, 0x2d, 0x16, 0x6b, 0x64, 0x89, 0x10, 0xe0, 0xb7, 0x61, 0x47, 0xea,
+	0x95, 0xe3, 0x5c, 0x22, 0xf1, 0x35, 0xc0, 0xe7, 0xa9, 0x2e, 0x96, 0x44, 0x97, 0x0f, 0xa1, 0x49,
+	0xa8, 0xcc, 0x8c, 0xbb, 0xcc, 0xec, 0xa6, 0xd2, 0x38, 0x6c, 0x2f, 0x17, 0x96, 0x75, 0x5a, 0x2c,
+	0x68, 0x0b, 0x4f, 0xa2, 0x88, 0xf9, 0xce, 0xc2, 0x78, 0x6d, 0x9d, 0x85, 0x31, 0x65, 0xeb, 0x49,
+	0x14, 0xeb, 0x51, 0x3c, 0x1b, 0xe5, 0x5d, 0x68, 0x3f, 0x8d, 0x75, 0x98, 0xa3, 0x33, 0x86, 0x62,
+	0x72, 0x8d, 0xc5, 0x57, 0xb0, 0x6b, 0x1a, 0x17, 0x5b, 0x70, 0xaa, 0xf2, 0xd7, 0x38, 0xd9, 0xd7,
+	0x6a, 0x66, 0xf1, 0x0b, 0x03, 0x1f, 0x25, 0x1b, 0x80, 0xb9, 0x00, 0x1c, 0xfc, 0xe3, 0xf3, 0xa5,
+	0x2a, 0xa9, 0x92, 0xcc, 0x07, 0xd0, 0x9d, 0xe6, 0xd8, 0xeb, 0xb3, 0x30, 0x2e, 0x54, 0xb9, 0x5d,
+	0x55, 0x85, 0x2c, 0xc6, 0x49, 0x6e, 0xcc, 0x3e, 0xd1, 0x5b, 0x63, 0x7e, 0x07, 0x3a, 0x87, 0x5a,
+	0xc7, 0xc6, 0xd8, 0x1c, 0xb0, 0x61, 0x5b, 0x3a, 0x05, 0xdf, 0x07, 0xb0, 0x7c, 0x0b, 0x15, 0xec,
+	0x50, 0x05, 0x2a, 0x1a, 0x71, 0x1f, 0x5a, 0x98, 0xe9, 0xf3, 0x70, 0xe9, 0xb8, 0xb1, 0xeb, 0xb8,
+	0xfd, 0xc3, 0xa0, 0xf7, 0x65, 0xa1, 0xd2, 0x73, 0xa9, 0xbe, 0x2b, 0x54, 0x96, 0x63, 0xdd, 0x09,
+	0xdb, 0xce, 0x21, 0x80, 0x3d, 0x32, 0x7d, 0x19, 0xa6, 0x73, 0x53, 0x29, 0x5f, 0x96, 0x08, 0xb9,
+	0xba, 0x9a, 0x67, 0xc4, 0xb5, 0x2d, 0xab, 0x2a, 0xea, 0x2e, 0xb5, 0xd0, 0xb9, 0x25, 0x53, 0x22,
+	0x3e, 0x84, 0xb7, 0x8f, 0x5e, 0x9d, 0xc6, 0xc5, 0x5c, 0x49, 0xbd, 0x32, 0xab, 0x77, 0xc8, 0x61,
+	0x53, 0xcd, 0xdf, 0x83, 0xbd, 0x52, 0x65, 0xc7, 0x54, 0x8b, 0x1c, 0x37, 0xb4, 0xfc, 0x01, 0xf4,
+	0x8e, 0x16, 0x27, 0x6a, 0x3e, 0x57, 0xf3, 0x51, 0x98, 0x87, 0x41, 0x9b, 0x78, 0x6f, 0x0c, 0x8d,
+	0x9a, 0x8b, 0xf8, 0x81, 0xc1, 0x6e, 0xc9, 0x3e, 0x5b, 0xea, 0x24, 0x53, 0x78, 0xc4, 0x47, 0x69,
+	0x6a, 0x8f, 0xf8, 0x28, 0x4d, 0xf9, 0x7d, 0x68, 0x49, 0x95, 0x15, 0x71, 0x6e, 0xbb, 0xe4, 0x96,
+	0x8b, 0x68, 0xd7, 0x16, 0x71, 0x2e, 0xad, 0x17, 0xff, 0x14, 0xf6, 0x6a, 0x7d, 0x68, 0xe6, 0x67,
+	0xf7, 0xe1, 0x3b, 0x6e, 0x5d, 0xcd, 0x2e, 0x37, 0xdc, 0xc5, 0xaf, 0x1e, 0x74, 0x2b, 0x91, 0xd7,
+	0x4d, 0x86, 0xf5, 0xd9, 0x2d, 0x9b, 0xec, 0x2e, 0xcd, 0xee, 0x2b, 0x26, 0x27, 0x4e, 0x80, 0x1e,
+	0xb0, 0x49, 0xd9, 0x96, 0x6c, 0xe2, 0xe6, 0x8a, 0x77, 0xdd, 0x5c, 0xc1, 0x97, 0xe0, 0x65, 0x98,
+	0x9c, 0xa9, 0x39, 0xb5, 0x65, 0x5b, 0x5a, 0xc8, 0x0f, 0xdc, 0x5d, 0xa5, 0x73, 0xac, 0x0d, 0x00,
+	0x6b, 0x91, 0xee, 0x3e, 0x9b, 0x99, 0x32, 0x1e, 0xe1, 0x59, 0x51, 0xbf, 0x18, 0xc4, 0x3f, 0x86,
+	0xae, 0x9b, 0x29, 0x59, 0x79, 0x44, 0x7d, 0x17, 0xca, 0x19, 0x65, 0xd5, 0x91, 0x7f, 0xb6, 0x39,
+	0xe6, 0x83, 0x0e, 0x65, 0x11, 0xd4, 0x98, 0x57, 0xec, 0x72, 0xf3, 0x59, 0x78, 0x50, 0x79, 0x77,
+	0x02, 0xa0, 0xc5, 0x37, 0xdd, 0xe2, 0xb5, 0x49, 0x56, 0x5e, 0xa7, 0x8f, 0xaa, 0xa3, 0x39, 0xe8,
+	0xd2, 0x9a, 0x7e, 0xbd, 0x72, 0xc6, 0x26, 0x2b, 0x7e, 0xe2, 0x4f, 0x06, 0xbb, 0xe3, 0xc5, 0x52,
+	0xa7, 0x79, 0xe5, 0x4a, 0x8d, 0x93, 0xb9, 0x7a, 0x65, 0xaf, 0x14, 0x81, 0xed, 0x43, 0x1c, 0xb5,
+	0x74, 0xb5, 0xe8, 0x2a, 0xf9, 0xd2, 0x80, 0x4a, 0x39, 0xfd, 0x5a, 0x39, 0xef, 0x40, 0xc7, 0xf4,
+	0x0e, 0x9a, 0x9a, 0x64, 0x72, 0x0a, 0xf3, 0x08, 0xaf, 0xe8, 0xe1, 0x6b, 0xd1, 0xc3, 0x67, 0x21,
+	0x8e, 0x11, 0xe3, 0x46, 0xc6, 0x36, 0x19, 0x2b, 0x1a, 0xb4, 0x1f, 0x47, 0x0b, 0x95, 0xe5, 0xe1,
+	0x62, 0x89, 0xf7, 0xd2, 0x1b, 0x7a, 0xb2, 0xa2, 0x11, 0x7f, 0x33, 0xe0, 0x86, 0x23, 0x8d, 0x9d,
+	0xff, 0x8f, 0xe8, 0xf5, 0x84, 0xea, 0x69, 0xb7, 0x2e, 0xa5, 0x7d, 0x1b, 0x76, 0x28, 0x1f, 0x9b,
+	0x72, 0x89, 0x70, 0x4a, 0xb9, 0x19, 0x69, 0xf8, 0x32, 0x59, 0x55, 0x71, 0x01, 0xbd, 0xca, 0x80,
+	0xc6, 0xee, 0xc2, 0xd8, 0x35, 0x9d, 0x98, 0x41, 0xff, 0x38, 0x0d, 0x93, 0x2c, 0x0e, 0x73, 0x85,
+	0xdb, 0xbd, 0x09, 0xeb, 0x2d, 0x3f, 0x2a, 0xf1, 0x3e, 0xdc, 0xda, 0x88, 0xeb, 0x66, 0x11, 0x96,
+	0xc1, 0xa3, 0x32, 0xa0, 0x28, 0xa6, 0x70, 0x73, 0xed, 0x3a, 0x1e, 0xbd, 0x51, 0x06, 0x97, 0x83,
+	0x7e, 0x50, 0xe1, 0x45, 0x41, 0xcb, 0xed, 0xb7, 0xe5, 0x7a, 0x08, 0x41, 0xd9, 0xdb, 0xe6, 0x3b,
+	0x57, 0x66, 0x30, 0x8b, 0xd4, 0x0a, 0xfd, 0x27, 0xe1, 0x42, 0x95, 0x49, 0x90, 0x8c, 0x3a, 0x9a,
+	0xc5, 0x0d, 0xfa, 0x04, 0x92, 0x2c, 0x7e, 0x64, 0xd0, 0xdf, 0x16, 0x84, 0x9e, 0xfc, 0x58, 0x85,
+	0x66, 0xfa, 0xb6, 0xa5, 0x01, 0xfc, 0x11, 0x34, 0xbf, 0x8f, 0xd4, 0xca, 0x4e, 0x5f, 0xe1, 0x2e,
+	0xe0, 0x55, 0x99, 0x48, 0xb3, 0x00, 0xdb, 0xe1, 0xf1, 0x69, 0x1e, 0xe9, 0xc4, 0x7e, 0x6c, 0x0c,
+	0xc2, 0x7d, 0x0e, 0x63, 0x7d, 0xfa, 0x2d, 0x0d, 0x39, 0x5f, 0x1a, 0x20, 0x7e, 0x66, 0x96, 0x5b,
+	0xe5, 0xf9, 0xfa, 0xcf, 0x0a, 0x9b, 0x1e, 0x2e, 0xff, 0x28, 0xa6, 0x87, 0x03, 0xf3, 0x06, 0xbb,
+	0xaf, 0x86, 0x85, 0xf8, 0xee, 0xa3, 0x38, 0x0b, 0x63, 0x73, 0x91, 0x3b, 0x72, 0x8d, 0xaf, 0xef,
+	0xfc, 0xc3, 0x1b, 0xbf, 0x5d, 0xec, 0xb3, 0xdf, 0x2f, 0xf6, 0xd9, 0x1f, 0x17, 0xfb, 0xec, 0xa7,
+	0xbf, 0xf6, 0xdf, 0x3a, 0xd9, 0xa1, 0xdf, 0xfc, 0x87, 0xff, 0x06, 0x00, 0x00, 0xff, 0xff, 0xa3,
+	0x51, 0x0d, 0x2e, 0xdd, 0x0b, 0x00, 0x00,
 }
 
 func (m *Row) Marshal() (dAtA []byte, err error) {
@@ -3190,6 +3208,18 @@ func (m *ImportRoaringRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.Block != 0 {
+		i = encodeVarintPublic(dAtA, i, uint64(m.Block))
+		i--
+		dAtA[i] = 0x20
+	}
+	if len(m.Action) > 0 {
+		i -= len(m.Action)
+		copy(dAtA[i:], m.Action)
+		i = encodeVarintPublic(dAtA, i, uint64(len(m.Action)))
+		i--
+		dAtA[i] = 0x1a
+	}
 	if len(m.Views) > 0 {
 		for iNdEx := len(m.Views) - 1; iNdEx >= 0; iNdEx-- {
 			{
@@ -3950,6 +3980,13 @@ func (m *ImportRoaringRequest) Size() (n int) {
 			n += 1 + l + sovPublic(uint64(l))
 		}
 	}
+	l = len(m.Action)
+	if l > 0 {
+		n += 1 + l + sovPublic(uint64(l))
+	}
+	if m.Block != 0 {
+		n += 1 + sovPublic(uint64(m.Block))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -8127,6 +8164,57 @@ func (m *ImportRoaringRequest) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Action", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPublic
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthPublic
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthPublic
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Action = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 4:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Block", wireType)
+			}
+			m.Block = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowPublic
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Block |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipPublic(dAtA[iNdEx:])
diff --git a/internal/public.proto b/internal/public.proto
index c87cd3357..872eadaf2 100644
--- a/internal/public.proto
+++ b/internal/public.proto
@@ -152,6 +152,8 @@ message ImportRoaringRequestView {
 message ImportRoaringRequest {
 	bool Clear = 1;
 	repeated ImportRoaringRequestView views = 2;
+	string Action = 3;
+	uint64 Block = 4;
 }
 
 message ImportColumnAttrsRequest {
diff --git a/pql/pql.peg.go b/pql/pql.peg.go
index 8ee80fd37..934dee1b1 100644
--- a/pql/pql.peg.go
+++ b/pql/pql.peg.go
@@ -1,10 +1,11 @@
 package pql
 
-//go:generate peg -inline pql.peg
+// Code generated by peg -inline pql.peg DO NOT EDIT.
 
 import (
 	"fmt"
-	"math"
+	"io"
+	"os"
 	"sort"
 	"strconv"
 )
@@ -218,19 +219,19 @@ type node32 struct {
 	up, next *node32
 }
 
-func (node *node32) print(pretty bool, buffer string) {
+func (node *node32) print(w io.Writer, pretty bool, buffer string) {
 	var print func(node *node32, depth int)
 	print = func(node *node32, depth int) {
 		for node != nil {
 			for c := 0; c < depth; c++ {
-				fmt.Printf(" ")
+				fmt.Fprintf(w, " ")
 			}
 			rule := rul3s[node.pegRule]
 			quote := strconv.Quote(string(([]rune(buffer)[node.begin:node.end])))
 			if !pretty {
-				fmt.Printf("%v %v\n", rule, quote)
+				fmt.Fprintf(w, "%v %v\n", rule, quote)
 			} else {
-				fmt.Printf("\x1B[34m%v\x1B[m %v\n", rule, quote)
+				fmt.Fprintf(w, "\x1B[34m%v\x1B[m %v\n", rule, quote)
 			}
 			if node.up != nil {
 				print(node.up, depth+1)
@@ -241,12 +242,12 @@ func (node *node32) print(pretty bool, buffer string) {
 	print(node, 0)
 }
 
-func (node *node32) Print(buffer string) {
-	node.print(false, buffer)
+func (node *node32) Print(w io.Writer, buffer string) {
+	node.print(w, false, buffer)
 }
 
-func (node *node32) PrettyPrint(buffer string) {
-	node.print(true, buffer)
+func (node *node32) PrettyPrint(w io.Writer, buffer string) {
+	node.print(w, true, buffer)
 }
 
 type tokens32 struct {
@@ -289,24 +290,24 @@ func (t *tokens32) AST() *node32 {
 }
 
 func (t *tokens32) PrintSyntaxTree(buffer string) {
-	t.AST().Print(buffer)
+	t.AST().Print(os.Stdout, buffer)
+}
+
+func (t *tokens32) WriteSyntaxTree(w io.Writer, buffer string) {
+	t.AST().Print(w, buffer)
 }
 
 func (t *tokens32) PrettyPrintSyntaxTree(buffer string) {
-	t.AST().PrettyPrint(buffer)
+	t.AST().PrettyPrint(os.Stdout, buffer)
 }
 
 func (t *tokens32) Add(rule pegRule, begin, end, index uint32) {
-	if tree := t.tree; int(index) >= len(tree) {
-		expanded := make([]token32, 2*len(tree))
-		copy(expanded, tree)
-		t.tree = expanded
-	}
-	t.tree[index] = token32{
-		pegRule: rule,
-		begin:   begin,
-		end:     end,
+	tree, i := t.tree, int(index)
+	if i >= len(tree) {
+		t.tree = append(tree, token32{pegRule: rule, begin: begin, end: end})
+		return
 	}
+	tree[i] = token32{pegRule: rule, begin: begin, end: end}
 }
 
 func (t *tokens32) Tokens() []token32 {
@@ -370,7 +371,7 @@ type parseError struct {
 }
 
 func (e *parseError) Error() string {
-	tokens, error := []token32{e.max}, "\n"
+	tokens, err := []token32{e.max}, "\n"
 	positions, p := make([]int, 2*len(tokens)), 0
 	for _, token := range tokens {
 		positions[p], p = int(token.begin), p+1
@@ -383,14 +384,14 @@ func (e *parseError) Error() string {
 	}
 	for _, token := range tokens {
 		begin, end := int(token.begin), int(token.end)
-		error += fmt.Sprintf(format,
+		err += fmt.Sprintf(format,
 			rul3s[token.pegRule],
 			translations[begin].line, translations[begin].symbol,
 			translations[end].line, translations[end].symbol,
 			strconv.Quote(string(e.p.buffer[begin:end])))
 	}
 
-	return error
+	return err
 }
 
 func (p *PQL) PrintSyntaxTree() {
@@ -401,6 +402,10 @@ func (p *PQL) PrintSyntaxTree() {
 	}
 }
 
+func (p *PQL) WriteSyntaxTree(w io.Writer) {
+	p.tokens32.WriteSyntaxTree(w, p.Buffer)
+}
+
 func (p *PQL) Execute() {
 	buffer, _buffer, text, begin, end := p.Buffer, p.buffer, "", 0, 0
 	for _, token := range p.Tokens() {
@@ -533,12 +538,31 @@ func (p *PQL) Execute() {
 	_, _, _, _, _ = buffer, _buffer, text, begin, end
 }
 
-func (p *PQL) Init() {
+func Pretty(pretty bool) func(*PQL) error {
+	return func(p *PQL) error {
+		p.Pretty = pretty
+		return nil
+	}
+}
+
+func Size(size int) func(*PQL) error {
+	return func(p *PQL) error {
+		p.tokens32 = tokens32{tree: make([]token32, 0, size)}
+		return nil
+	}
+}
+func (p *PQL) Init(options ...func(*PQL) error) error {
 	var (
 		max                  token32
 		position, tokenIndex uint32
 		buffer               []rune
 	)
+	for _, option := range options {
+		err := option(p)
+		if err != nil {
+			return err
+		}
+	}
 	p.reset = func() {
 		max = token32{}
 		position, tokenIndex = 0, 0
@@ -552,7 +576,7 @@ func (p *PQL) Init() {
 	p.reset()
 
 	_rules := p.rules
-	tree := tokens32{tree: make([]token32, math.MaxInt16)}
+	tree := p.tokens32
 	p.parse = func(rule ...int) error {
 		r := 1
 		if len(rule) > 0 {
@@ -3178,4 +3202,5 @@ func (p *PQL) Init() {
 		nil,
 	}
 	p.rules = _rules
+	return nil
 }
diff --git a/server/handler_test.go b/server/handler_test.go
index 003c9d10b..e561430fa 100644
--- a/server/handler_test.go
+++ b/server/handler_test.go
@@ -232,20 +232,20 @@ func TestHandler_Endpoints(t *testing.T) {
 		}) {
 			t.Fatalf("Unexpected result %v", resp.Results[0])
 		}
-
 	})
 
-	t.Run("ImportRoaringFieldTypeFail", func(t *testing.T) {
-		// Roaring import into a non-set field should fail.
-		if _, err := i0.CreateFieldIfNotExists("int-field", pilosa.OptFieldTypeInt(0, 1)); err != nil {
+	t.Run("ImportRoaringOverwrite", func(t *testing.T) {
+		if _, err := i0.CreateFieldIfNotExists("int-field", pilosa.OptFieldTypeInt(0, 10)); err != nil {
 			t.Fatal(err)
 		}
 		w := httptest.NewRecorder()
-		roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100")
+		roaringData, _ := hex.DecodeString("3C30000002000000000000000000000001000000200000000000000001000000280000002A00000001000100")
+
 		msg := pilosa.ImportRoaringRequest{
-			Clear: false,
+			Action: pilosa.RequestActionOverwrite,
+			Block:  0,
 			Views: map[string][]byte{
-				"": roaringData,
+				"bsig_int-field": roaringData,
 			},
 		}
 		ser := proto.Serializer{}
@@ -256,11 +256,15 @@ func TestHandler_Endpoints(t *testing.T) {
 		httpReq := test.MustNewHTTPRequest("POST", "/index/i0/field/int-field/import-roaring/0", bytes.NewBuffer(data))
 		httpReq.Header.Set("Content-Type", "application/x-protobuf")
 		httpReq.Header.Set("Accept", "application/x-protobuf")
-		h.ServeHTTP(w, httpReq)
-		if w.Code != gohttp.StatusBadRequest {
-			t.Fatalf("unexpected status code: %d", w.Code)
-		}
 
+		h.ServeHTTP(w, httpReq)
+		resp, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i0", Query: "Row(int-field>0)"})
+		if err != nil {
+			t.Fatalf("querying: %v", err)
+		}
+		if row := resp.Results[0].(*pilosa.Row); !reflect.DeepEqual(row.Columns(), []uint64{1}) {
+			t.Fatalf("Unexpected result %v", row.Columns())
+		}
 	})
 
 	t.Run("Status", func(t *testing.T) {
diff --git a/test/holder.go b/test/holder.go
index 4a45f9b91..3d73a0bf0 100644
--- a/test/holder.go
+++ b/test/holder.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/pilosa/pilosa/v2"
 	"github.com/pilosa/pilosa/v2/boltdb"
+	"github.com/pilosa/pilosa/v2/pql"
 )
 
 // Holder is a test wrapper for pilosa.Holder.
@@ -167,7 +168,7 @@ func (h *Holder) MustSetBits(index, field string, rowID uint64, columnIDs ...uin
 	}
 }
 
-// SetValue sets an value on the given field.
+// SetValue sets an integer value on the given field.
 func (h *Holder) SetValue(index, field string, columnID uint64, value int64) {
 	idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{})
 	f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
@@ -179,3 +180,32 @@ func (h *Holder) SetValue(index, field string, columnID uint64, value int64) {
 		panic(err)
 	}
 }
+
+// Value returns the integer value for a given column.
+func (h *Holder) Value(index, field string, columnID uint64) (int64, bool) {
+	idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{})
+	f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
+	if err != nil {
+		panic(err)
+	}
+	val, exists, err := f.Value(columnID)
+	if err != nil {
+		panic(err)
+	}
+	return val, exists
+}
+
+// Range returns a Row (of column IDs) for a field based
+// on the given range.
+func (h *Holder) Range(index, field string, op pql.Token, predicate int64) *pilosa.Row {
+	idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{})
+	f, err := idx.CreateFieldIfNotExists(field, pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
+	if err != nil {
+		panic(err)
+	}
+	row, err := f.Range(field, op, predicate)
+	if err != nil {
+		panic(err)
+	}
+	return row
+}