Batch Shard -> Index insertions

client/host_queue.go

@@ -286,7 +286,7 @@ func (q *queue) asyncTaggedWrite(
 	// TODO(r): Use a worker pool to avoid creating new go routines for async writes
 	go func() {
 		req := q.writeTaggedBatchRawRequestPool.Get()
-		req.NameSpace = namespace.Data().Bytes()
+		req.NameSpace = namespace.Bytes()
 		req.Elements = elems
 
 		// NB(r): Defer is slow in the hot path unfortunately
@@ -351,7 +351,7 @@ func (q *queue) asyncWrite(
 	// TODO(r): Use a worker pool to avoid creating new go routines for async writes
 	go func() {
 		req := q.writeBatchRawRequestPool.Get()
-		req.NameSpace = namespace.Data().Bytes()
+		req.NameSpace = namespace.Bytes()
 		req.Elements = elems
 
 		// NB(r): Defer is slow in the hot path unfortunately

client/session.go

@@ -976,7 +976,7 @@ func (s *session) writeAttemptWithRLock(
 		wop := s.pools.writeOperation.Get()
 		wop.namespace = nsID
 		wop.shardID = s.state.topoMap.ShardSet().Lookup(tsID)
-		wop.request.ID = tsID.Data().Bytes()
+		wop.request.ID = tsID.Bytes()
 		wop.request.Datapoint.Value = value
 		wop.request.Datapoint.Timestamp = timestamp
 		wop.request.Datapoint.TimestampTimeType = timeType
@@ -986,7 +986,7 @@ func (s *session) writeAttemptWithRLock(
 		wop := s.pools.writeTaggedOperation.Get()
 		wop.namespace = nsID
 		wop.shardID = s.state.topoMap.ShardSet().Lookup(tsID)
-		wop.request.ID = tsID.Data().Bytes()
+		wop.request.ID = tsID.Bytes()
 		encodedTagBytes, ok := tagEncoder.Data()
 		if !ok {
 			return nil, 0, 0, errUnableToEncodeTags
@@ -1445,7 +1445,7 @@ func (s *session) fetchIDsAttempt(
 			}
 
 			// Append IDWithNamespace to this request
-			f.append(namespace.Data().Bytes(), tsID.Data().Bytes(), completionFn)
+			f.append(namespace.Bytes(), tsID.Bytes(), completionFn)
 		}); err != nil {
 			routeErr = err
 			break
@@ -1576,7 +1576,7 @@ func (s *session) Truncate(namespace ident.ID) (int64, error) {
 	)
 
 	t := &truncateOp{}
-	t.request.NameSpace = namespace.Data().Bytes()
+	t.request.NameSpace = namespace.Bytes()
 	t.completionFn = func(result interface{}, err error) {
 		if err != nil {
 			resultErrLock.Lock()
@@ -2044,7 +2044,7 @@ func (s *session) streamBlocksMetadataFromPeer(
 	attemptFn := func(client rpc.TChanNode) error {
 		tctx, _ := thrift.NewContext(s.streamBlocksMetadataBatchTimeout)
 		req := rpc.NewFetchBlocksMetadataRawRequest()
-		req.NameSpace = namespace.Data().Bytes()
+		req.NameSpace = namespace.Bytes()
 		req.Shard = int32(shard)
 		req.RangeStart = start.UnixNano()
 		req.RangeEnd = end.UnixNano()
@@ -2209,7 +2209,7 @@ func (s *session) streamBlocksMetadataFromPeerV2(
 	attemptFn := func(client rpc.TChanNode) error {
 		tctx, _ := thrift.NewContext(s.streamBlocksMetadataBatchTimeout)
 		req := rpc.NewFetchBlocksMetadataRawV2Request()
-		req.NameSpace = namespace.Data().Bytes()
+		req.NameSpace = namespace.Bytes()
 		req.Shard = int32(shard)
 		req.RangeStart = start.UnixNano()
 		req.RangeEnd = end.UnixNano()
@@ -2808,7 +2808,7 @@ func (s *session) streamBlocksBatchFromPeer(
 		retention          = ropts.RetentionPeriod()
 		earliestBlockStart = nowFn().Add(-retention).Truncate(ropts.BlockSize())
 	)
-	req.NameSpace = namespaceMetadata.ID().Data().Bytes()
+	req.NameSpace = namespaceMetadata.ID().Bytes()
 	req.Shard = int32(shard)
 	req.Elements = make([]*rpc.FetchBlocksRawRequestElement, 0, len(batch))
 	for i := range batch {
@@ -2817,7 +2817,7 @@ func (s *session) streamBlocksBatchFromPeer(
 			continue // Fell out of retention while we were streaming blocks
 		}
 		req.Elements = append(req.Elements, &rpc.FetchBlocksRawRequestElement{
-			ID:     batch[i].id.Data().Bytes(),
+			ID:     batch[i].id.Bytes(),
 			Starts: []int64{blockStart.UnixNano()},
 		})
 		reqBlocksLen++
@@ -2871,7 +2871,7 @@ func (s *session) streamBlocksBatchFromPeer(
 		}
 
 		id := batch[i].id
-		if !bytes.Equal(id.Data().Bytes(), result.Elements[i].ID) {
+		if !bytes.Equal(id.Bytes(), result.Elements[i].ID) {
 			blocksErr := fmt.Errorf(
 				"stream blocks mismatched ID: expectedID=%s, actualID=%s, indexID=%d, peer=%s",
 				batch[i].id.String(), id.String(), i, peer.Host().String(),

client/session_fetch_bulk_blocks_test.go

@@ -24,6 +24,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"sort"
 	"sync"
@@ -714,7 +715,7 @@ func assertFetchBlocksFromPeersResult(
 		seg, err := stream.Segment()
 		require.NoError(t, err)
 
-		actualData := append(seg.Head.Bytes(), seg.Tail.Bytes()...)
+		actualData := append(bytesFor(seg.Head), bytesFor(seg.Tail)...)
 
 		// compare actual v expected data
 		if len(expectedData) != len(actualData) {
@@ -1716,8 +1717,9 @@ func TestBlocksResultAddBlockFromPeerReadMerged(t *testing.T) {
 	require.NoError(t, err)
 
 	// Assert block has data
-	assert.Equal(t, []byte{1, 2}, seg.Head.Bytes())
-	assert.Equal(t, []byte{3}, seg.Tail.Bytes())
+	data, err := ioutil.ReadAll(xio.NewSegmentReader(seg))
+	require.NoError(t, err)
+	assert.Equal(t, []byte{1, 2, 3}, data)
 }
 
 func TestBlocksResultAddBlockFromPeerReadUnmerged(t *testing.T) {
@@ -2107,7 +2109,7 @@ func expectFetchMetadataAndReturn(
 		)
 		for j := beginIdx; j < len(result) && j < beginIdx+batchSize; j++ {
 			elem := &rpc.BlocksMetadata{}
-			elem.ID = result[j].id.Data().Bytes()
+			elem.ID = result[j].id.Bytes()
 			for k := 0; k < len(result[j].blocks); k++ {
 				bl := &rpc.BlockMetadata{}
 				bl.Start = result[j].blocks[k].start.UnixNano()
@@ -2159,7 +2161,7 @@ func expectFetchMetadataAndReturnV2(
 			beginIdx = i * batchSize
 		)
 		for j := beginIdx; j < len(result) && j < beginIdx+batchSize; j++ {
-			id := result[j].id.Data().Bytes()
+			id := result[j].id.Bytes()
 			for k := 0; k < len(result[j].blocks); k++ {
 				bl := &rpc.BlockMetadataV2{}
 				bl.ID = id
@@ -2311,7 +2313,7 @@ func expectFetchBlocksAndReturn(
 		ret := &rpc.FetchBlocksRawResult_{}
 		for _, res := range result[i] {
 			blocks := &rpc.Blocks{}
-			blocks.ID = res.id.Data().Bytes()
+			blocks.ID = res.id.Bytes()
 			for j := range res.blocks {
 				bl := &rpc.Block{}
 				bl.Start = res.blocks[j].start.UnixNano()
@@ -2468,7 +2470,7 @@ func assertFetchBootstrapBlocksResult(
 				require.NoError(t, err)
 				seg, err := stream.Segment()
 				require.NoError(t, err)
-				actualData := append(seg.Head.Bytes(), seg.Tail.Bytes()...)
+				actualData := append(bytesFor(seg.Head), bytesFor(seg.Tail)...)
 				assert.Equal(t, expectedData, actualData)
 			} else if block.segments.unmerged != nil {
 				assert.Fail(t, "unmerged comparison not supported")
@@ -2477,6 +2479,13 @@ func assertFetchBootstrapBlocksResult(
 	}
 }
 
+func bytesFor(data checked.Bytes) []byte {
+	if data == nil {
+		return nil
+	}
+	return data.Bytes()
+}
+
 func assertEnqueueChannel(
 	t *testing.T,
 	expected []receivedBlockMetadata,

glide.yaml

@@ -1,7 +1,7 @@
 package: github.com/m3db/m3db
 import:
 - package: github.com/m3db/m3x
-  version: df82cd7aacfd8439586821ecd6e073da35bfbb02
+  version: 80455550b18244f71637a568f097ae929748e5be
   vcs: git
   subpackages:
   - checked
@@ -26,7 +26,7 @@ import:
   version: ed532baee45a440f0b08b6893c816634c6978d4d
 
 - package: github.com/m3db/m3ninx
-  version: 12b8ac4f173f9d539b0e94f3bca475318ab1a8db
+  version: 2a492ea6d91d3e2b5b9deb1b5cab343c80de373b
 
 - package: github.com/m3db/bitset
   version: 07973db6b78acb62ac207d0538055e874b49d90d

integration/client.go

@@ -59,7 +59,7 @@ func tchannelClientWriteBatch(client rpc.TChanNode, timeout time.Duration, names
 	for _, series := range seriesList {
 		for _, dp := range series.Data {
 			elem := &rpc.WriteBatchRawRequestElement{
-				ID: series.ID.Data().Bytes(),
+				ID: series.ID.Bytes(),
 				Datapoint: &rpc.Datapoint{
 					Timestamp:         xtime.ToNormalizedTime(dp.Timestamp, time.Second),
 					Value:             dp.Value,
@@ -72,7 +72,7 @@ func tchannelClientWriteBatch(client rpc.TChanNode, timeout time.Duration, names
 
 	ctx, _ := thrift.NewContext(timeout)
 	batchReq := &rpc.WriteBatchRawRequest{
-		NameSpace: namespace.Data().Bytes(),
+		NameSpace: namespace.Bytes(),
 		Elements:  elems,
 	}
 	return client.WriteBatchRaw(ctx, batchReq)

integration/generate/generate.go

@@ -36,7 +36,7 @@ import (
 func (l SeriesBlock) Len() int      { return len(l) }
 func (l SeriesBlock) Swap(i, j int) { l[i], l[j] = l[j], l[i] }
 func (l SeriesBlock) Less(i, j int) bool {
-	return bytes.Compare(l[i].ID.Data().Bytes(), l[j].ID.Data().Bytes()) < 0
+	return bytes.Compare(l[i].ID.Bytes(), l[j].ID.Bytes()) < 0
 }
 
 // Block generates a SeriesBlock based on provided config
@@ -120,5 +120,5 @@ func (l SeriesDataPointsByTime) Less(i, j int) bool {
 	if !l[i].Timestamp.Equal(l[j].Timestamp) {
 		return l[i].Timestamp.Before(l[j].Timestamp)
 	}
-	return bytes.Compare(l[i].ID.Data().Bytes(), l[j].ID.Data().Bytes()) < 0
+	return bytes.Compare(l[i].ID.Bytes(), l[j].ID.Bytes()) < 0
 }

integration/integration_data_verify.go

@@ -263,7 +263,7 @@ func compareSeriesList(
 	require.Equal(t, len(expected), len(actual))
 
 	for i := range expected {
-		require.Equal(t, expected[i].ID.Data().Bytes(), actual[i].ID.Data().Bytes())
+		require.Equal(t, expected[i].ID.Bytes(), actual[i].ID.Bytes())
 		require.Equal(t, expected[i].Data, expected[i].Data)
 	}
 }

integration/truncate_namespace_test.go

@@ -106,7 +106,7 @@ func TestTruncateNamespace(t *testing.T) {
 
 	log.Debugf("truncate namespace %s", testNamespaces[0])
 	truncateReq := rpc.NewTruncateRequest()
-	truncateReq.NameSpace = testNamespaces[0].Data().Bytes()
+	truncateReq.NameSpace = testNamespaces[0].Bytes()
 	truncated, err := testSetup.truncate(truncateReq)
 	require.NoError(t, err)
 	require.Equal(t, int64(1), truncated)

network/server/tchannelthrift/node/service.go

@@ -144,7 +144,7 @@ func NewService(db storage.Database, opts tchannelthrift.Options) rpc.TChanNode
 	scope := iopts.
 		MetricsScope().
 		SubScope("service").
-		Tagged(map[string]string{"serviceName": "node"})
+		Tagged(map[string]string{"service-name": "node"})
 
 	wrapperPoolOpts := pool.NewObjectPoolOptions().
 		SetSize(checkedBytesPoolSize).
@@ -527,7 +527,7 @@ func (s *service) FetchBlocksMetadataRaw(tctx thrift.Context, req *rpc.FetchBloc
 
 	for _, fetchedMetadata := range fetchedResults {
 		blocksMetadata := s.pools.blocksMetadata.Get()
-		blocksMetadata.ID = fetchedMetadata.ID.Data().Bytes()
+		blocksMetadata.ID = fetchedMetadata.ID.Bytes()
 		fetchedMetadataBlocks := fetchedMetadata.Blocks.Results()
 		blocksMetadata.Blocks = s.pools.blockMetadataSlice.Get()
 

persist/fs/commitlog/writer.go

@@ -200,8 +200,8 @@ func (w *writer) Write(
 		// If "idx" likely hasn't been written to commit log
 		// yet we need to include series metadata
 		var metadata schema.LogMetadata
-		metadata.ID = series.ID.Data().Bytes()
-		metadata.Namespace = series.Namespace.Data().Bytes()
+		metadata.ID = series.ID.Bytes()
+		metadata.Namespace = series.Namespace.Bytes()
 		metadata.Shard = series.Shard
 		metadata.EncodedTags = encodedTags
 		w.metadataEncoder.Reset()

persist/fs/index_lookup.go

@@ -85,7 +85,7 @@ func (il *nearestIndexOffsetLookup) concurrentClone() (*nearestIndexOffsetLookup
 // In other words, the returned offset can always be used as a starting point to
 // begin scanning the index file for the desired series.
 func (il *nearestIndexOffsetLookup) getNearestIndexFileOffset(id ident.ID) (int64, error) {
-	idBytes := id.Data().Bytes()
+	idBytes := id.Bytes()
 
 	min := 0
 	max := len(il.summaryIDsOffsets) - 1

persist/fs/index_lookup_prop_test.go

@@ -49,7 +49,7 @@ func TestIndexLookupWriteRead(t *testing.T) {
 		writes := []generatedWrite{}
 		unique := map[string]struct{}{}
 		for _, write := range input.realWrites {
-			s := string(write.id.Data().Bytes())
+			s := string(write.id.Bytes())
 			if _, ok := unique[s]; ok {
 				continue
 			}

persist/fs/read_write_test.go

@@ -167,7 +167,7 @@ func readTestData(t *testing.T, r DataFileSetReader, shard uint32, timestamp tim
 
 				// Verify that the bloomFilter was bootstrapped properly by making sure it
 				// at least contains every ID
-				assert.True(t, bloomFilter.Test(id.Data().Bytes()))
+				assert.True(t, bloomFilter.Test(id.Bytes()))
 
 				id.Finalize()
 				tags.Close()
@@ -191,7 +191,7 @@ func readTestData(t *testing.T, r DataFileSetReader, shard uint32, timestamp tim
 
 				// Verify that the bloomFilter was bootstrapped properly by making sure it
 				// at least contains every ID
-				assert.True(t, bloomFilter.Test(id.Data().Bytes()))
+				assert.True(t, bloomFilter.Test(id.Bytes()))
 
 				id.Finalize()
 				tags.Close()

persist/fs/retriever.go

@@ -400,7 +400,7 @@ func (r *blockRetriever) Stream(
 
 	// If the ID is not in the seeker's bloom filter, then it's definitely not on
 	// disk and we can return immediately
-	if !bloomFilter.Test(id.Data().Bytes()) {
+	if !bloomFilter.Test(id.Bytes()) {
 		// No need to call req.onRetrieve.OnRetrieveBlock if there is no data
 		req.onRetrieved(ts.Segment{})
 		return req.toBlock(), nil

persist/fs/seek.go

@@ -414,7 +414,7 @@ func (s *seeker) SeekIndexEntry(id ident.ID) (IndexEntry, error) {
 	stream := msgpack.NewDecoderStream(s.indexMmap[offset:])
 	s.decoder.Reset(stream)
 
-	idBytes := id.Data().Bytes()
+	idBytes := id.Bytes()
 	// Prevent panic's when we're scanning to the end of the buffer
 	for stream.Remaining() != 0 {
 		entry, err := s.decoder.DecodeIndexEntry()

persist/fs/write.go

@@ -97,7 +97,7 @@ func (e indexEntries) Len() int {
 }
 
 func (e indexEntries) Less(i, j int) bool {
-	return bytes.Compare(e[i].id.Data().Bytes(), e[j].id.Data().Bytes()) < 0
+	return bytes.Compare(e[i].id.Bytes(), e[j].id.Bytes()) < 0
 }
 
 func (e indexEntries) Swap(i, j int) {
@@ -421,7 +421,7 @@ func (w *writer) writeIndexFileContents(
 	)
 	defer tagsEncoder.Finalize()
 	for i := range w.indexEntries {
-		id := w.indexEntries[i].id.Data().Bytes()
+		id := w.indexEntries[i].id.Bytes()
 		// Need to check if i > 0 or we can never write an empty string ID
 		if i > 0 && bytes.Equal(id, prevID) {
 			// Should never happen, Write() should only be called once per ID
@@ -491,7 +491,7 @@ func (w *writer) writeSummariesFileContents(
 
 		summary := schema.IndexSummary{
 			Index:            w.indexEntries[i].index,
-			ID:               w.indexEntries[i].id.Data().Bytes(),
+			ID:               w.indexEntries[i].id.Bytes(),
 			IndexEntryOffset: w.indexEntries[i].indexFileOffset,
 		}