func (t *table) skip(offset wal.Offset, source int) {
	t.rowStore.insert(&insert{nil, nil, nil, offset, source})
}
Skip informs the table of a new offset so that we can store it
package zenodb
import (
	"fmt"
	"hash"
	"reflect"
	"strings"
	"time"
	"github.com/dustin/go-humanize"
	"github.com/getlantern/bytemap"
	"github.com/getlantern/errors"
	"github.com/getlantern/wal"
	"github.com/getlantern/zenodb/encoding"
)
func (db *DB) Insert(stream string, ts time.Time, dims map[string]interface{}, vals map[string]interface{}) error {
	return db.InsertRaw(stream, ts, bytemap.New(dims), bytemap.New(vals))
}
func (db *DB) InsertRaw(stream string, ts time.Time, dims bytemap.ByteMap, vals bytemap.ByteMap) error {
	if db.opts.Follow != nil {
		return errors.New("Declining to insert data directly to follower")
	}
	stream = strings.TrimSpace(strings.ToLower(stream))
	db.tablesMutex.Lock()
	w := db.streams[stream]
	db.tablesMutex.Unlock()
	if w == nil {
		return fmt.Errorf("No wal found for stream %v", stream)
	}
	if len(db.opts.WhitelistedDimensions) > 0 {
		if db.log.IsTraceEnabled() {
			db.log.Tracef("Whitelist Dims Original dims: %v", dims.AsMap())
			db.log.Tracef("Whitelist Dims Slicing dims with whitelist: %v", db.opts.WhitelistedDimensions)
		}
		dims = dims.Slice(db.opts.WhitelistedDimensions)
		if db.log.IsTraceEnabled() {
			db.log.Tracef("Whitelist Dims Sliced dims: %v", dims.AsMap())
		}
	}
	tsd := make([]byte, encoding.Width64bits)
	encoding.EncodeTime(tsd, ts)
	dimsLen := make([]byte, encoding.Width32bits)
	encoding.WriteInt32(dimsLen, len(dims))
	valsLen := make([]byte, encoding.Width32bits)
	encoding.WriteInt32(valsLen, len(vals))
	if db.log.IsTraceEnabled() {
		db.log.Tracef("Writing to wal with dims length %d: %v", len(dims), bytemap.ByteMap(dims).AsMap())
	}
	err := w.Write(tsd, dimsLen, dims, valsLen, vals)
	if err != nil {
		db.log.Error(err)
	}
	return err
}
type walRead struct {
	data   []byte
	offset wal.Offset
	source int
}
func (t *table) processWALInserts() {
	in := make(chan *walRead)
	t.db.Go(func(stop <-chan interface{}) {
		t.processInserts(in, stop)
	})
	for {
		data, err := t.wal.Read()
		if err != nil {
			t.db.Panic(fmt.Errorf("Unable to read from WAL: %v", err))
		}
		in <- &walRead{data, t.wal.Offset(), 0}
	}
}
func (t *table) processInserts(in chan *walRead, stop <-chan interface{}) {
	isFollower := t.db.opts.Follow != nil
	start := time.Now()
	inserted := 0
	skipped := 0
	bytesRead := 0
	h := partitionHash()
loop:
	for {
		select {
		case <-stop:
			return
		case read := <-in:
			if read.data == nil {
				continue loop
			}
			bytesRead += len(read.data)
			if t.insert(read.data, isFollower, h, read.offset, read.source) {
				inserted++
			} else {
				t.skip(read.offset, read.source)
				skipped++
			}
			t.db.walBuffers.Put(read.data)
			delta := time.Now().Sub(start)
			if delta > 1*time.Minute {
				t.log.Debugf("Read %v at %v per second", humanize.Bytes(uint64(bytesRead)), humanize.Bytes(uint64(float64(bytesRead)/delta.Seconds())))
				t.log.Debugf("Inserted %v points at %v per second", humanize.Comma(int64(inserted)), humanize.Commaf(float64(inserted)/delta.Seconds()))
				t.log.Debugf("Skipped %v points at %v per second", humanize.Comma(int64(skipped)), humanize.Commaf(float64(skipped)/delta.Seconds()))
				inserted = 0
				skipped = 0
				bytesRead = 0
				start = time.Now()
			}
		}
	}
}
func (t *table) insert(data []byte, isFollower bool, h hash.Hash32, offset wal.Offset, source int) bool {
	defer func() {
		p := recover()
		if p != nil {
			t.log.Errorf("Panic in inserting: %v", p)
		}
	}()
	tsd, remain := encoding.Read(data, encoding.Width64bits)
	ts := encoding.TimeFromBytes(tsd)
	if ts.Before(t.truncateBefore()) {
		return false
	}
	dimsLen, remain := encoding.ReadInt32(remain)
	dims, remain := encoding.Read(remain, dimsLen)
	if isFollower && !t.db.inPartition(h, dims, t.PartitionBy, t.db.opts.Partition) {
		if t.log.IsTraceEnabled() {
			t.log.Tracef("Not in partition: %v", bytemap.ByteMap(dims).AsMap())
		}
		return false
	}
	valsLen, remain := encoding.ReadInt32(remain)
	vals, _ := encoding.Read(remain, valsLen)
	dimsBM := make(bytemap.ByteMap, len(dims))
	valsBM := make(bytemap.ByteMap, len(vals))
	copy(dimsBM, dims)
	copy(valsBM, vals)
	if t.log.IsTraceEnabled() {
		t.log.Tracef("Length of data: %d", len(data))
		if len(vals) > 0 {
			t.log.Tracef("Length of vals: %d", len(valsBM))
			t.log.Tracef("Vals are %v", valsBM.AsMap())
		}
		if len(dims) > 0 {
			t.log.Tracef("Length of dims: %d", len(dimsBM))
			t.log.Tracef("Dims are %v", dimsBM.AsMap())
		}
	}
	return t.doInsert(ts, dimsBM, valsBM, offset, source)
}