featurebase/ingest/op.go

// Copyright 2021 Molecula Corp.
//
// Licensed 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 ingest

import (
	"errors"
	"fmt"
	"math/bits"
	"sort"

	"github.com/molecula/featurebase/v2/shardwidth"
)

type OpType uint8

const (
	OpNone = OpType(iota)
	OpSet
	OpRemove
	OpClear
	OpWrite
	OpDelete
)

var opNames = []string{
	"none",
	"set",
	"remove",
	"clear",
	"write",
	"delete",
}

type FieldType string

const (
	FieldTypeSet         = "set"
	FieldTypeInt         = "int"
	FieldTypeTimeQuantum = "time"
	FieldTypeTimeStamp   = "timestamp"
	FieldTypeDecimal     = "decimal"
	FieldTypeMutex       = "mutex"
	FieldTypeBool        = "bool"
)

var fieldTypeSorts = map[FieldType]func(*FieldOperation){
	FieldTypeSet:         (*FieldOperation).SortByValues,
	FieldTypeInt:         (*FieldOperation).SortByRecords,
	FieldTypeTimeQuantum: (*FieldOperation).SortByValues,
	FieldTypeDecimal:     (*FieldOperation).SortByRecords,
	FieldTypeMutex:       (*FieldOperation).SortByValues,
	FieldTypeTimeStamp:   (*FieldOperation).SortByRecords,
	FieldTypeBool:        (*FieldOperation).SortByValues,
}

func (o OpType) String() string {
	if int(o) < len(opNames) {
		return opNames[o]
	}
	return fmt.Sprintf("invalid-optype-%d", o)
}

func ParseOpType(s string) (OpType, error) {
	for i, v := range opNames[1:] {
		if s == v {
			return OpType(i + 1), nil
		}
	}
	return 0, fmt.Errorf("unknown operation type %q", s)
}

// Operation represents a single set of changes to make to
// the stored data, which means some combination of clearing
// columns, clearing individual bits, or setting bits or values.
// The same data structure can be used whether this represents the
// whole database operation or a single shard's values.
//
// Operations can specify individual per-field operations, which
// have maps of record IDs to values. They can also have a set of
// record IDs and fields to clear. A Clear operation will have only
// record IDs and fields, a Set or Remove will have only FieldOps,
// and a Write will have both -- populating the record IDs and fields
// from the fieldops.
//
// When we parse an operation, we assign each op a sequential ID
// within the overall request. We keep these IDs associated with ops
// when splitting them up across shards, so we can reverse this even
// if some shards don't get some ops.
type Operation struct {
	OpType         OpType
	Seq            int // sequence position within a chain of ops
	ClearRecordIDs []uint64
	ClearFields    []string
	FieldOps       map[string]*FieldOperation
}

// Compare reports whether two Operations seem to be the same. While
// a FieldOperation can be "empty" and compare-equal-to a nil FieldOperation,
// no Operation is considered empty even if it has no FieldOps.
func (got *Operation) Compare(expected *Operation) error {
	if got == nil && expected == nil {
		return nil
	}
	if got == nil {
		return fmt.Errorf("expected %q op, got nil", expected.OpType)
	}
	if expected == nil {
		return fmt.Errorf("expected no op, got %q", got.OpType)
	}
	if got.OpType != expected.OpType {
		return fmt.Errorf("operation type mismatch: expected %q, got %q", expected.OpType, got.OpType)
	}
	if len(got.ClearRecordIDs) != len(expected.ClearRecordIDs) {
		return fmt.Errorf("clear record counts differ: expected %d, got %d", len(expected.ClearRecordIDs), len(got.ClearRecordIDs))
	}
	for i, v1 := range got.ClearRecordIDs {
		v2 := expected.ClearRecordIDs[i]
		if v1 != v2 {
			return fmt.Errorf("clear record id %d differs: expected %d, got %d", i, v2, v1)
		}
	}
	// don't assume consistent ordering for the fields, because they're
	// coming out in arbitrary hash order
	seenFields := make(map[string]struct{}, len(expected.ClearFields))
	for _, v1 := range expected.ClearFields {
		seenFields[v1] = struct{}{}
	}
	for _, v2 := range got.ClearFields {
		if _, ok := seenFields[v2]; !ok {
			return fmt.Errorf("field %q cleared unexpectedly", v2)
		}
		delete(seenFields, v2)
	}
	for v1 := range seenFields {
		return fmt.Errorf("field %q should be cleared but wasn't", v1)
	}
	// We check compare even if the op we find on one side is nil, so
	// a round-trip test will consider a missing op and an empty op to
	// be interchangeable.
	for k, fo1 := range got.FieldOps {
		fo2 := expected.FieldOps[k]
		if err := fo1.Compare(fo2); err != nil {
			return fmt.Errorf("field %q mismatch: %w", k, err)
		}
	}
	for k, fo2 := range expected.FieldOps {
		fo1 := got.FieldOps[k]
		if err := fo1.Compare(fo2); err != nil {
			return fmt.Errorf("field %q mismatch: %w", k, err)
		}
	}
	if expected.Seq != got.Seq {
		return fmt.Errorf("sequence mismatch: expected %d, got %d", expected.Seq, got.Seq)
	}
	return nil
}

// FieldOperation is the specific set of changes to make to a given
// field.
//
// For a Clear operation, values can be an empty array. For Set or Remove
// operations on sets, RecordIDs can contain duplicates. Times should
// be empty except for time-quantum fields.
type FieldOperation struct {
	RecordIDs []uint64
	Values    []uint64
	// For int/timestamp/decimal, this is the value
	// For time-quantum, this is the timestamp
	// No field has both signed values and timestamps.
	// This is not a place of honor.
	Signed []int64
}

// Sort sorts the clear record IDs and field list.
func (o *Operation) Sort() {
	// I am aware that this is a crime, but it avoids rewriting
	// the code and justifies FieldOperation handling the "only record
	// IDs" case.
	f := FieldOperation{RecordIDs: o.ClearRecordIDs}
	f.SortByRecords()
	sort.Strings(o.ClearFields)
}

// clone makes a duplicate of the operation without shared storage
func (o *Operation) clone() *Operation {
	o2 := &Operation{
		OpType:         o.OpType,
		Seq:            o.Seq,
		ClearRecordIDs: append([]uint64{}, o.ClearRecordIDs...),
		ClearFields:    append([]string{}, o.ClearFields...),
		FieldOps:       make(map[string]*FieldOperation, len(o.FieldOps)),
	}
	for k, v := range o.FieldOps {
		o2.FieldOps[k] = v.clone()
	}
	return o2
}

// merge merges the fieldops of the provided operation into this operation.
func (o *Operation) merge(o2 *Operation) {
	o.ClearRecordIDs = append(o.ClearRecordIDs, o2.ClearRecordIDs...)
	o.ClearFields = append(o.ClearFields, o2.ClearFields...)
	for f, op := range o2.FieldOps {
		dst := o.FieldOps[f]
		if dst == nil {
			o.FieldOps[f] = op
			continue
		}
		// otherwise append any contents. dst is a pointer-to, so this
		// updates the thing the map points to, we don't have to write back
		// into the map.
		dst.RecordIDs = append(dst.RecordIDs, op.RecordIDs...)
		dst.Values = append(dst.Values, op.Values...)
		dst.Signed = append(dst.Signed, op.Signed...)
	}
}

type ShardedFieldOperation map[uint64]*FieldOperation

// ByShard() divides the FieldOperation's values up into corresponding chunks
// based on the shards of record IDs. Does not further sort IDs within those
// chunks.
func (f *FieldOperation) ByShard() ShardedFieldOperation {
	if len(f.RecordIDs) == 0 {
		return nil
	}
	return f.SortToShards()
}

// clone makes a duplicate of the operation without shared storage
func (f *FieldOperation) clone() *FieldOperation {
	f2 := &FieldOperation{
		RecordIDs: append([]uint64{}, f.RecordIDs...),
		Values:    append([]uint64{}, f.Values...),
		Signed:    append([]int64{}, f.Signed...),
	}
	return f2
}

func ShardIDs(ids []uint64) (out map[uint64][]uint64) {
	shards, ends := shardwidth.FindShards(ids)
	prev := 0
	out = make(map[uint64][]uint64, len(shards))
	for i, shard := range shards {
		endIndex := ends[i]
		out[shard] = ids[prev:endIndex]
		prev = endIndex
	}
	return out
}

// SortToShards() uses a pseudo-radix-sort to divide inputs into
// shards; the individual shards are not sorted.
func (f *FieldOperation) SortToShards() ShardedFieldOperation {
	if len(f.RecordIDs) == 0 {
		return nil
	}
	diffMask := uint64(0)
	prev := f.RecordIDs[0]
	for _, r := range f.RecordIDs[1:] {
		diffMask |= r ^ prev
		prev = r
	}
	bitsRemaining := bits.Len64(diffMask)
	if bitsRemaining <= shardwidth.Exponent {
		return map[uint64]*FieldOperation{f.RecordIDs[0] >> shardwidth.Exponent: f}
	}
	output := make(ShardedFieldOperation)
	sortToShardsInto(f, bitsRemaining-8, output)
	return output
}

// sortToShardsInto puts the shards it finds into the given map, so that
// as we split off buckets, they can be inserted into the same map.
func sortToShardsInto(f *FieldOperation, shift int, into ShardedFieldOperation) {
	if shift < shardwidth.Exponent {
		shift = shardwidth.Exponent
	}
	nextShift := shift - 8
	if nextShift < shardwidth.Exponent {
		nextShift = shardwidth.Exponent
	}
	// count things that belong in each of the 256 buckets
	var buckets [256]int
	var starts [256]int

	// compute the buckets ourselves
	for _, r := range f.RecordIDs {
		b := (r >> shift) & 0xFF
		buckets[b]++
	}
	total := 0
	// compute starting points of each bucket, converting the
	// bucket counts into ends
	for i := range buckets {
		starts[i] = total
		total += buckets[i]
		buckets[i] = total
	}
	// starts[n] is the index of the first thing that should
	// go in that bucket, buckets[n] is the index of the first
	// thing that shouldn't
	var bucketOp FieldOperation
	for bucket, start := range starts {
		end := buckets[bucket]
		if end <= start {
			continue
		}
		for j := start; j < end; j++ {
			want := int((f.RecordIDs[j] >> shift) & 0xFF)
			for want != bucket {
				// move this to the beginning of the
				// bucket it wants to be in, swapping
				// the thing there here
				dst := starts[want]
				f.RecordIDs[j], f.RecordIDs[dst] = f.RecordIDs[dst], f.RecordIDs[j]
				if f.Values != nil {
					f.Values[j], f.Values[dst] = f.Values[dst], f.Values[j]
				}
				if f.Signed != nil {
					f.Signed[j], f.Signed[dst] = f.Signed[dst], f.Signed[j]
				}
				starts[want]++
				want = int((f.RecordIDs[j] >> shift) & 0xFF)
			}
		}
		// If shift == shardwidth.Exponent, then this is a completed
		// shard and can go into the sharded output. otherwise, we
		// can subdivide it.
		bucketOp.RecordIDs = f.RecordIDs[start:end]
		if f.Values != nil {
			bucketOp.Values = f.Values[start:end]
		}
		if f.Signed != nil {
			bucketOp.Signed = f.Signed[start:end]
		}
		if shift == shardwidth.Exponent {
			x := bucketOp
			into[f.RecordIDs[start]>>shardwidth.Exponent] = &x
		} else {
			sortToShardsInto(&bucketOp, nextShift, into)
		}
	}
}

const shardMask = ((uint64(1) << shardwidth.Exponent) - 1)

// SortByValues sorts the operation by values first, then by record
// ID within each value. This is the best ordering for set/mutex fields,
// where we'll want to generate positions in that order. For these
// purposes, a time quantum or bool counts as a kind of a set.
func (f *FieldOperation) SortByValues() {
	keys := make([]uint64, len(f.RecordIDs))
	for i, v := range f.RecordIDs {
		keys[i] = (f.Values[i] << shardwidth.Exponent) | (v & shardMask)
	}
	f.SortByKeys(keys)
}

// SortByRecords sorts the operation by record ID, and not by value at
// all. This makes the most sense for int fields and the like.
func (f *FieldOperation) SortByRecords() {
	f.SortByKeys(f.RecordIDs)
}

// SortByKeys reorganizes the record IDs and values of f according to the
// corresponding members of keys.
func (f *FieldOperation) SortByKeys(keys []uint64) {
	if len(f.RecordIDs) < 2 {
		return
	}
	diffMask := uint64(0)
	prev := keys[0]
	for _, r := range keys[1:] {
		diffMask |= r ^ prev
		prev = r
	}
	bitsRemaining := bits.Len64(diffMask)
	sortPartialByKeys(f, keys, bitsRemaining-8)
}

// simpleSort sorts a FieldOperation by external keys, or record IDs. It's a
// horribly naive bubble sort because N is small and a more complex algorithm
// doesn't help as much as you'd hope. This beats using stdlib sort by about
// a factor of two for those small N, for larger N we're using the radix sort
// that calls this.
//
// External keys exist only when we are sorting by value, which is to say,
// when we're using row-oriented formats (set, mutex, time quantum).
// For int/decimal/timestamp fields, we're sorting by record only.
// So, if keys is the same as f.RecordIDs, we're looking at an int field
// or equivalent, so Signed exists and Values doesn't exist.
// Otherwise, we might be looking at a time quantum field (both exist)
// or set/mutex (only Values exist).
func simpleSort(f *FieldOperation, keys []uint64) {
	// keys might actually just point to record IDs, in which case, we don't
	// want to shuffle the corresponding RecordIDs too, because that would just
	// reverse our swaps. If they're different, we actually need to swap them
	// both.
	if &keys[0] != &f.RecordIDs[0] {
		// sorting by record IDs
		if f.Values != nil && f.Signed != nil { // time quantum field
			for i := 1; i < len(keys); i++ {
				for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
					keys[j-1], keys[j] = keys[j], keys[j-1]
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
					f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
					f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]

				}
			}
		} else if f.Values != nil { // set/mutex/bool
			for i := 1; i < len(keys); i++ {
				for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
					keys[j-1], keys[j] = keys[j], keys[j-1]
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]

					f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
				}
			}
		} else if f.Signed != nil { // can't-happen, we think
			for i := 1; i < len(keys); i++ {
				for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
					keys[j-1], keys[j] = keys[j], keys[j-1]
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
					f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
				}
			}
		} else { // can't-happen, we think
			for i := 1; i < len(keys); i++ {
				for j := i; j > 0 && keys[j-1] > keys[j]; j-- {
					keys[j-1], keys[j] = keys[j], keys[j-1]
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
				}
			}
		}
	} else {
		if f.Values == nil && f.Signed != nil {
			for i := 1; i < len(f.RecordIDs); i++ {
				for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
					f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
				}
			}
		} else if f.Values != nil && f.Signed != nil { // can't happen, we think
			for i := 1; i < len(f.RecordIDs); i++ {
				for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
					f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
					f.Signed[j-1], f.Signed[j] = f.Signed[j], f.Signed[j-1]
				}
			}
		} else if f.Values != nil { // only happens during testing
			for i := 1; i < len(f.RecordIDs); i++ {
				for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
					f.Values[j-1], f.Values[j] = f.Values[j], f.Values[j-1]
				}
			}
		} else { // should definitely not happen
			for i := 1; i < len(f.RecordIDs); i++ {
				for j := i; j > 0 && f.RecordIDs[j-1] > f.RecordIDs[j]; j-- {
					f.RecordIDs[j-1], f.RecordIDs[j] = f.RecordIDs[j], f.RecordIDs[j-1]
				}
			}
		}
	}
}

func sortPartialByKeys(f *FieldOperation, keys []uint64, shift int) {
	if shift < 0 {
		shift = 0
	}
	externalKeys := &f.RecordIDs[0] != &keys[0]
	nextShift := shift - 8
	if nextShift < 0 {
		nextShift = 0
	}
	// count things that belong in each of the 256 buckets
	var buckets [256]int
	var starts [256]int
	// compute the buckets ourselves
	for _, r := range keys {
		b := (r >> shift) & 0xFF
		buckets[b]++
	}
	total := 0
	// compute starting points of each bucket, converting the
	// bucket counts into ends
	for i := range buckets {
		starts[i] = total
		total += buckets[i]
		buckets[i] = total
	}
	// starts[n] is the index of the first thing that should
	// go in that bucket, buckets[n] is the index of the first
	// thing that shouldn't
	// var newbuckets [256]int
	var bucketOp FieldOperation
	for bucket, start := range starts {
		end := buckets[bucket]
		if end <= start {
			continue
		}
		for j := start; j < end; j++ {
			want := int((keys[j] >> shift) & 0xFF)
			for want != bucket {
				// move this to the beginning of the
				// bucket it wants to be in, swapping
				// the thing there here
				dst := starts[want]
				keys[j], keys[dst] = keys[dst], keys[j]
				// we do this to allow you to just pass in the records as keys
				if externalKeys {
					f.RecordIDs[j], f.RecordIDs[dst] = f.RecordIDs[dst], f.RecordIDs[j]
				}
				if f.Values != nil {
					f.Values[j], f.Values[dst] = f.Values[dst], f.Values[j]
				}
				if f.Signed != nil {
					f.Signed[j], f.Signed[dst] = f.Signed[dst], f.Signed[j]
				}
				starts[want]++
				want = int((keys[j] >> shift) & 0xFF)
			}
		}
		// If shift == shardwidth.Exponent, then this is a completed
		// shard and can go into the sharded output. otherwise, we
		// can subdivide it.
		if shift > 0 {
			bucketOp.RecordIDs = f.RecordIDs[start:end]
			if f.Values != nil {
				bucketOp.Values = f.Values[start:end]
			}
			if f.Signed != nil {
				bucketOp.Signed = f.Signed[start:end]
			}
			// if there's not very many, sort naively instead
			if end-start > 32 {
				sortPartialByKeys(&bucketOp, keys[start:end], nextShift)
			} else {
				simpleSort(&bucketOp, keys[start:end])
			}
		}
	}
}

// AddPair adds a record ID/value pair where the value is unsigned, as
// when used with set/mutex/time quantum fields.
func (f *FieldOperation) AddPair(rec uint64, value uint64) {
	f.RecordIDs = append(f.RecordIDs, rec)
	f.Values = append(f.Values, value)
}

// AddSignedPair adds a record ID/value pair where the value is signed,
// as when used with int/decimal/timestamp fields.
func (f *FieldOperation) AddSignedPair(rec uint64, value int64) {
	f.RecordIDs = append(f.RecordIDs, rec)
	f.Signed = append(f.Signed, value)
}

// AddStampedPair adds a record/value pair plus a time, which is just
// a Unix time in seconds. (Note, no scaling here; timestamp fields are
// scaled int fields, this is for time quantums.)
func (f *FieldOperation) AddStampedPair(rec uint64, value uint64, stamp int64) {
	f.RecordIDs = append(f.RecordIDs, rec)
	f.Values = append(f.Values, value)
	f.Signed = append(f.Signed, stamp)
}

// Compare returns a diagnostic if the field operations do not seem
// equivalent.
func (got *FieldOperation) Compare(expected *FieldOperation) error {
	if got == nil {
		if expected == nil {
			return nil
		}
		// We don't worry about non-empty Values or Signed here, because in theory
		// RecordIDs are the Source of Truth as to what's in the op.
		if len(expected.RecordIDs) == 0 {
			return nil
		}
		return fmt.Errorf("expected field operation with %d records, got nil", len(expected.RecordIDs))
	}
	if expected == nil {
		if len(got.RecordIDs) == 0 {
			return nil
		}
		return fmt.Errorf("expected empty field operation, got %d records", len(got.RecordIDs))
	}
	if len(got.RecordIDs) != len(expected.RecordIDs) {
		return fmt.Errorf("record counts differ: expected %d, got %d", len(expected.RecordIDs), len(got.RecordIDs))
	}
	for i, v1 := range got.RecordIDs {
		v2 := expected.RecordIDs[i]
		if v1 != v2 {
			return fmt.Errorf("record id %d differs: expected %d, got %d", i, v2, v1)
		}
	}
	if len(got.Values) != len(expected.Values) {
		return fmt.Errorf("value counts differ: expected %d, got %d", len(expected.Values), len(got.Values))
	}
	for i, v1 := range got.Values {
		v2 := expected.Values[i]
		if v1 != v2 {
			return fmt.Errorf("value %d differs: expected %d, got %d", i, v2, v1)
		}
	}
	if len(got.Signed) != len(expected.Signed) {
		return fmt.Errorf("signed value counts differ: expected %d, got %d", len(expected.Signed), len(got.Signed))
	}
	for i, v1 := range got.Signed {
		v2 := expected.Signed[i]
		if v1 != v2 {
			return fmt.Errorf("signed value %d differs: expected %d, got %d", i, v2, v1)
		}
	}
	return nil
}

// ShardOperations is a set of Operations associated with a specific shard.
type ShardOperations struct {
	Shard uint64
	Ops   []*Operation
}

// Request is a complete ingest request, which may be any combination
// of operations, which may apply to multiple shards.
type Request struct {
	Ops []*Operation
}

// ShardedRequest is an ingest request, split up into individual per-shard
// operations.
type ShardedRequest struct {
	Ops map[uint64][]*Operation
}

// ByShard converts a request into the same request, only sharded.
func (r *Request) ByShard(fields map[string]FieldType) (*ShardedRequest, error) {
	if len(r.Ops) == 0 {
		return &ShardedRequest{Ops: nil}, nil
	}
	req := make(map[uint64][]*Operation)
	shards := make(map[uint64]*Operation)
	// we're getting per-field things, which we want to divide per-shard,
	// and return to per-shard sets of per-field things, so we're inverting
	// the structure.
	for _, op := range r.Ops {
		// for clear and write ops, we also need to split up the
		// ClearRecords values, which may be distinct from the set of
		// records for any given field. For Write ops, we'll then end
		// up adding in field values for some fields.
		if op.OpType == OpClear || op.OpType == OpWrite || op.OpType == OpDelete {
			sharded := ShardIDs(op.ClearRecordIDs)
			for shard, data := range sharded {
				shards[shard] = &Operation{OpType: op.OpType, Seq: op.Seq, ClearRecordIDs: data, ClearFields: op.ClearFields, FieldOps: map[string]*FieldOperation{}}
			}
		}
		for field, fieldOp := range op.FieldOps {
			sharded := fieldOp.ByShard()
			sorter := fieldTypeSorts[fields[field]]
			if sorter == nil {
				sorter = (*FieldOperation).SortByRecords
			}
			for shard, data := range sharded {
				sorter(data)
				shardOp, ok := shards[shard]
				if !ok {
					if op.OpType == OpWrite {
						return nil, fmt.Errorf("write operation has field operation data (%d items) for shard %d, but no clear data", len(data.RecordIDs), shard)
					}
					shardOp = &Operation{OpType: op.OpType, Seq: op.Seq}
					shards[shard] = shardOp
					shardOp.FieldOps = map[string]*FieldOperation{field: data}
				} else {
					shardOp.FieldOps[field] = data
				}
			}
		}
		for shard, shardOp := range shards {
			req[shard] = append(req[shard], shardOp)
		}
		for k := range shards {
			delete(shards, k)
		}

	}
	return &ShardedRequest{Ops: req}, nil
}

// merge combines the components of a sharded request back into a single
// unsharded request, processing shards in numerical order.
func (s *ShardedRequest) merge() *Request {
	req := &Request{}
	if s == nil || len(s.Ops) == 0 {
		return req
	}
	shards := make([]uint64, 0, len(s.Ops))
	for shard := range s.Ops {
		shards = append(shards, shard)
	}
	sort.Slice(shards, func(i, j int) bool { return shards[i] < shards[j] })
	for _, shard := range shards {
		ops := s.Ops[shard]
		for _, op := range ops {
			var _ *Operation
			if op.Seq >= len(req.Ops) {
				// Pad out with nil *Operations to the required length
				req.Ops = append(req.Ops, make([]*Operation, op.Seq+1-len(req.Ops))...)
			}
			if req.Ops[op.Seq] == nil {
				req.Ops[op.Seq] = op.clone()
				continue
			}
			req.Ops[op.Seq].merge(op)
		}
	}
	return req
}

func (r *Request) Dump(logf func(string, ...interface{})) {
	logf("req: %#v", r)
	for _, op := range r.Ops {
		logf("op: %#v", op)
		if len(op.ClearRecordIDs) > 0 {
			if len(op.ClearRecordIDs) > 8 {
				logf("  clearRecordIDs: %d...+%d", op.ClearRecordIDs[:8], len(op.ClearRecordIDs)-8)
			} else {
				logf("  clearRecordIDs: %d", op.ClearRecordIDs)
			}
		}
		if len(op.ClearFields) > 0 {
			if len(op.ClearFields) > 8 {
				logf("  clearFields: %s...+%d", op.ClearFields[:8], len(op.ClearFields)-8)
			} else {
				logf("  clearFields: %s", op.ClearFields)
			}
		}
		for field, fieldOp := range op.FieldOps {
			if fieldOp != nil {
				logf("  field %q: op (%d/%d/%d)", field, len(fieldOp.RecordIDs), len(fieldOp.Values), len(fieldOp.Signed))
				if len(fieldOp.RecordIDs) > 0 {
					if len(fieldOp.RecordIDs) > 8 {
						logf("    records %d...+%d", fieldOp.RecordIDs[:8], len(fieldOp.RecordIDs)-8)
					} else {
						logf("    records %d", fieldOp.RecordIDs)
					}
				}
			} else {
				logf("  field %q: nil op", field)
			}
		}
	}
}

func (r *Request) Compare(other *Request) error {
	if other == nil {
		if r != nil && len(r.Ops) != 0 {
			return errors.New("non-empty sharded request can't equal empty/nil sharded request")
		}
		// empty and nil are allowed
		return nil
	}
	if r == nil {
		if other != nil && len(other.Ops) != 0 {
			return errors.New("non-empty sharded request can't equal empty/nil sharded request")
		}
		// empty and nil are allowed
		return nil
	}
	ops := r.Ops
	ops2 := other.Ops
	if len(ops2) != len(ops) {
		return fmt.Errorf("expected %d ops, got %d", len(ops), len(ops2))
	}
	for i, op := range ops {
		if err := op.Compare(ops2[i]); err != nil {
			return fmt.Errorf("op %d: %v", i, err)
		}
	}
	return nil
}

// Compare checks whether two ShardedRequest objects represent the same
// data. Empty shards shouldn't have entries in the map in the first place,
// so we don't accept a nil or 0-length slice of ops as equal to the
// shard key not existing, but we do accept nil or empty requests as
// equal to each other.
func (s *ShardedRequest) Compare(other *ShardedRequest) error {
	if other == nil {
		if s != nil && len(s.Ops) != 0 {
			return errors.New("non-empty sharded request can't equal empty/nil sharded request")
		}
		// empty and nil are allowed
		return nil
	}
	if s == nil {
		if other != nil && len(other.Ops) != 0 {
			return errors.New("non-empty sharded request can't equal empty/nil sharded request")
		}
		// empty and nil are allowed
		return nil
	}
	for shard, ops := range s.Ops {
		ops2, ok := other.Ops[shard]
		if !ok {
			return fmt.Errorf("shard %d missing in other", shard)
		}
		if len(ops2) != len(ops) {
			return fmt.Errorf("shard %d: expected %d ops, got %d", shard, len(ops), len(ops2))
		}
		for i, op := range ops {
			if err := op.Compare(ops2[i]); err != nil {
				return fmt.Errorf("shard %d, op %d: %v", shard, i, err)
			}
		}
	}
	if len(other.Ops) != len(s.Ops) {
		for shard := range other.Ops {
			if _, ok := s.Ops[shard]; !ok {
				return fmt.Errorf("shard %d missing in self", shard)
			}
		}
	}
	return nil
}