feat: support nanosecond MID (phase 2) (#291)

Author: cheb0

asyncsearcher/async_searcher.go

@@ -183,7 +183,7 @@ func newAsyncSearchInfo(r AsyncSearchRequest, list fracmanager.List) asyncSearch
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	return asyncSearchInfo{
-		Version:    infoVersion1,
+		Version:    infoVersion2,
 		Finished:   false,
 		Error:      "",
 		CanceledAt: time.Time{},
@@ -591,10 +591,10 @@ func loadAsyncRequests(dataDir string) (map[string]asyncSearchInfo, error) {
 		if info.Version == 0 {
 			info.Version = infoVersion1
 		}
-		if info.Version == infoVersion2 {
-			info.Request.Params.From = seq.NanosToMID(uint64(info.Request.Params.From))
-			info.Request.Params.To = seq.NanosToMID(uint64(info.Request.Params.To))
-			info.Version = infoVersion1
+		if info.Version == infoVersion1 {
+			info.Request.Params.From = seq.MillisToMID(uint64(info.Request.Params.From))
+			info.Request.Params.To = seq.MillisToMID(uint64(info.Request.Params.To))
+			info.Version = infoVersion2
 		}
 
 		info.merged.Store(areQPRsMerged[requestID])

asyncsearcher/encoding.go

@@ -27,7 +27,7 @@ var availableVersions = map[qprBinVersion]struct{}{
 }
 
 func marshalQPR(q *seq.QPR, dst []byte) []byte {
-	dst = append(dst, uint8(qprBinVersion1))
+	dst = append(dst, uint8(qprBinVersion2))
 
 	blocksLenPos := len(dst)
 	dst = append(dst, make([]byte, 8)...)
@@ -238,8 +238,8 @@ func unmarshalIDsDelta(dst seq.IDSources, block []byte, version qprBinVersion) (
 		block = block[hintSize:]
 
 		var midValue seq.MID
-		if version == qprBinVersion2 {
-			midValue = seq.NanosToMID(uint64(mid))
+		if version == qprBinVersion1 {
+			midValue = seq.MillisToMID(uint64(mid))
 		} else {
 			midValue = seq.MID(mid)
 		}
@@ -296,8 +296,8 @@ func unmarshalHistogram(src []byte, version qprBinVersion) (map[seq.MID]uint64,
 		}
 
 		var midValue seq.MID
-		if version == qprBinVersion2 {
-			midValue = seq.NanosToMID(uint64(mid))
+		if version == qprBinVersion1 {
+			midValue = seq.MillisToMID(uint64(mid))
 		} else {
 			midValue = seq.MID(mid)
 		}
@@ -460,8 +460,8 @@ func unmarshalAggregatableSamples(q *seq.AggregatableSamples, src []byte, versio
 		src = tail
 
 		var midValue seq.MID
-		if version == qprBinVersion2 {
-			midValue = seq.NanosToMID(uint64(mid))
+		if version == qprBinVersion1 {
+			midValue = seq.MillisToMID(uint64(mid))
 		} else {
 			midValue = seq.MID(mid)
 		}

asyncsearcher/encoding_test.go

@@ -92,34 +92,34 @@ func TestQPRMarshalUnmarshal(t *testing.T) {
 	}
 }
 
-// TestQPRVersion2Compatibility tests that it's possible to unmarshall and read version 2 async search result encoded.
-// MIDs in IDs and a histogram must be converted to milliseconds
-func TestQPRVersion2Compatibility(t *testing.T) {
+// TestQPRVersion1Compatibility tests that it's possible to unmarshall and read version 1 async search result encoded.
+// MIDs in IDs and a histogram must be converted from millis to nanos
+func TestQPRVersion1Compatibility(t *testing.T) {
 	qpr := seq.QPR{
 		IDs: seq.IDSources{
 			{
-				ID: seq.ID{MID: 1761812502573000000, RID: 34734732392},
+				ID: seq.ID{MID: seq.MID(1761812502573), RID: 34734732392},
 			},
 		},
 		Histogram: map[seq.MID]uint64{
-			1761812502573000000: 433,
-			1761812502463000000: 743,
+			seq.MID(1761812502573): 433,
+			seq.MID(1761812502463): 743,
 		},
 		Aggs: []seq.AggregatableSamples{
 			{
 				SamplesByBin: map[seq.AggBin]*seq.SamplesContainer{
 					{Token: "_not_exists"}: {
 						Total: 1,
 					},
-					{Token: "seq-db store", MID: seq.MID(1761812502953000000)}: {
+					{Token: "seq-db store", MID: seq.MID(1761812502953)}: {
 						Min:       3,
 						Max:       5,
 						Sum:       794,
 						Total:     1,
 						NotExists: 7,
 						Samples:   []float64{324},
 					},
-					{Token: "seq-db store", MID: seq.MID(1761812502456000000)}: {
+					{Token: "seq-db store", MID: seq.MID(1761812502456)}: {
 						Min:       2,
 						Max:       6,
 						Sum:       544,
@@ -132,27 +132,30 @@ func TestQPRVersion2Compatibility(t *testing.T) {
 			},
 		},
 	}
+
 	rawQPR := marshalQPR(&qpr, nil)
-	rawQPR[0] = uint8(qprBinVersion2)
-	var outQpr seq.QPR
-	tail, err := unmarshalQPR(&outQpr, rawQPR, math.MaxInt)
+	rawQPR[0] = uint8(qprBinVersion1)
+
+	var outQPR seq.QPR
+	tail, err := unmarshalQPR(&outQPR, rawQPR, math.MaxInt)
 	require.NoError(t, err)
 	require.Equal(t, 0, len(tail))
-	require.Equal(t, seq.MID(1761812502573), outQpr.IDs[0].ID.MID, "mid doesn't match, should convert to milliseconds")
 
-	require.Equal(t, 2, len(outQpr.Histogram))
-	require.Equal(t, uint64(433), outQpr.Histogram[seq.MID(1761812502573)], "histogram bucket doesn't match")
-	require.Equal(t, uint64(743), outQpr.Histogram[seq.MID(1761812502463)], "histogram bucket doesn't match")
+	require.Equal(t, seq.MID(1761812502573000000), outQPR.IDs[0].ID.MID, "mid doesn't match, should convert to nanoseconds")
+
+	require.Len(t, outQPR.Histogram, 2)
+	require.Equal(t, uint64(433), outQPR.Histogram[seq.MID(1761812502573000000)], "histogram bucket doesn't match")
+	require.Equal(t, uint64(743), outQPR.Histogram[seq.MID(1761812502463000000)], "histogram bucket doesn't match")
 
-	require.Equal(t, 1, len(outQpr.Aggs), "should have one AggregatableSamples")
-	agg := outQpr.Aggs[0]
-	require.Equal(t, 3, len(agg.SamplesByBin), "should have 3 samples in bin")
+	require.Len(t, outQPR.Aggs, 1, "should have one AggregatableSamples")
+	agg := outQPR.Aggs[0]
+	require.Len(t, agg.SamplesByBin, 3, "should have 3 samples in bin")
 
 	notExistsBin := seq.AggBin{Token: "_not_exists"}
 	require.Equal(t, int64(1), agg.SamplesByBin[notExistsBin].Total, "bucket doesn't match")
-	bin1 := seq.AggBin{Token: "seq-db store", MID: seq.MID(1761812502953)}
+	bin1 := seq.AggBin{Token: "seq-db store", MID: seq.MID(1761812502953000000)}
 	require.Equal(t, int64(1), agg.SamplesByBin[bin1].Total, "bucket doesn't match")
-	bin2 := seq.AggBin{Token: "seq-db store", MID: seq.MID(1761812502456)}
+	bin2 := seq.AggBin{Token: "seq-db store", MID: seq.MID(1761812502456000000)}
 	require.Equal(t, int64(2), agg.SamplesByBin[bin2].Total, "bucket doesn't match")
 }
 

cmd/distribution/main.go

@@ -183,8 +183,8 @@ func main() {
 				zap.String("name", info.Name()),
 				zap.String("ver", info.Ver),
 				zap.Uint32("docs_total", info.DocsTotal),
-				zap.String("from", util.MsTsToESFormat(uint64(info.From))),
-				zap.String("to", util.MsTsToESFormat(uint64(info.To))),
+				zap.String("from", util.NsTsToESFormat(uint64(info.From))),
+				zap.String("to", util.NsTsToESFormat(uint64(info.To))),
 				zap.String("creation_time", util.MsTsToESFormat(info.CreationTime)),
 			)
 		}

config/frac_version.go

@@ -11,4 +11,4 @@ const (
 	BinaryDataV2
 )
 
-const CurrentFracVersion = BinaryDataV1
+const CurrentFracVersion = BinaryDataV2

docs/en/internal/common.md

@@ -3,7 +3,7 @@
 ## Common
 
 * ID: MID-RID .
-* MID - milliseconds part of ID, generated (extracted from doc) by ingestor before sending to store.
+* MID - nanoseconds part of ID, generated (extracted from doc) by ingestor before sending to store.
 * RID - random part of ID, generated by ingestor before sending to store.
 * docParam - link of ID and block position, position of doc in block.
 * Only active fraction has meta file. It is used for restoring index in memory and in process of sealing fraction it is used to form index file.

docs/en/internal/search.md

@@ -26,7 +26,7 @@ Some basic overview of nodes:
 >
 > **ID** (document ID) - full id of a document, that you can use on proxy to find this specific doc. Consists of two parts: mid and rid.
 >
-> **MID** (milliseconds ID) - timestamp of a document. This is a timestamp, that log was written into stdout on the machine, not when it came into seq-db, meaning that it can be quite old relative to the time on the seq-db machine.
+> **MID** (nanoseconds ID) - timestamp of a document. This is a timestamp, that log was written into stdout on the machine, not when it came into seq-db, meaning that it can be quite old relative to the time on the seq-db machine.
 >
 > **RID** (random ID) - random part of an id
 

docs/ru/internal/common.md

@@ -3,7 +3,7 @@
 ## Common
 
 * ID: MID-RID .
-* MID - milliseconds part of ID, generated (extracted from doc) by ingestor before sending to store.
+* MID - nanoseconds part of ID, generated (extracted from doc) by ingestor before sending to store.
 * RID - random part of ID, generated by ingestor before sending to store.
 * docParam - link of ID and block position, position of doc in block.
 * Only active fraction has meta file. It is used for restoring index in memory and in process of sealing fraction it is used to form index file.

docs/ru/internal/search.md

@@ -26,7 +26,7 @@ Some basic overview of nodes:
 >
 > **ID** (document ID) - full id of a document, that you can use on proxy to find this specific doc. Consists of two parts: mid and rid.
 >
-> **MID** (milliseconds ID) - timestamp of a document. This is a timestamp, that log was written into stdout on the machine, not when it came into seq-db, meaning that it can be quite old relative to the time on the seq-db machine.
+> **MID** (nanoseconds ID) - timestamp of a document. This is a timestamp, that log was written into stdout on the machine, not when it came into seq-db, meaning that it can be quite old relative to the time on the seq-db machine.
 >
 > **RID** (random ID) - random part of an id
 

frac/common/info.go

@@ -1,6 +1,7 @@
 package common
 
 import (
+	"encoding/json"
 	"fmt"
 	"math"
 	"path"
@@ -80,7 +81,7 @@ func (s *Info) BuildDistribution(mids []uint64) {
 }
 
 func (s *Info) InitEmptyDistribution() bool {
-	from := time.UnixMilli(int64(s.From))
+	from := s.From.Time()
 	creationTime := time.UnixMilli(int64(s.CreationTime))
 	if creationTime.Sub(from) < DistributionSpreadThreshold { // no big spread in past
 		return false
@@ -117,3 +118,35 @@ func (s *Info) IsIntersecting(from, to seq.MID) bool {
 	// check with distribution
 	return s.Distribution.IsIntersecting(from, to)
 }
+
+// MarshalJSON implements custom JSON marshaling to always store From and To in milliseconds
+func (s *Info) MarshalJSON() ([]byte, error) {
+	type TmpInfo Info // type alias to avoid infinite recursion
+
+	tmp := TmpInfo(*s)
+
+	// We convert "from" and "to" to milliseconds in order to guarantee we can rollback on deploy.
+	// When converting nanos to millis we must round "from" down (floor) and round "to" up (ceiling).
+	// This guarantees that a fraction time range (checked on search with Contains and IsIntersecting methods) is not narrowed down,
+	// and we do not lose messages on search.
+	tmp.From = seq.MID(seq.MIDToMillis(s.From))
+	tmp.To = seq.MID(seq.MIDToCeilingMillis(s.To))
+
+	return json.Marshal(tmp)
+}
+
+// UnmarshalJSON implements custom JSON unmarshaling to convert From and To from milliseconds to nanoseconds
+func (s *Info) UnmarshalJSON(data []byte) error {
+	type TmpInfo Info // type alias to avoid infinite recursion
+	var tmp TmpInfo
+
+	err := json.Unmarshal(data, &tmp)
+	if err != nil {
+		return err
+	}
+
+	*s = Info(tmp)
+	s.From = seq.MillisToMID(uint64(tmp.From))
+	s.To = seq.MillisToMID(uint64(tmp.To))
+	return nil
+}