update codec to reflect changes to timstamp range

also address a few minor feedback comments from review

api.go

@@ -2006,11 +2006,7 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
 				return fmt.Errorf("adding decimal field to codec: %w", err)
 			}
 		case "timestamp":
-			// TODO: codec will likely need to be updated to reflect changes (increase) to timestamp range
+			if err = codec.AddTimestampField(field.name, field.options.TimeUnit, field.options.Base); err != nil {
-			// i.e. values are not all being converted to Nanos and being stored at the granularity
-			// specified by user and taking advantage of the range that provides.
-			nanos := TimeUnitNanos(field.options.TimeUnit)
-			if err = codec.AddTimestampField(field.name, time.Duration(nanos), field.options.Base); err != nil {
 				return fmt.Errorf("adding timestamp field to codec: %w", err)
 			}
 		default:

executor.go

@@ -1590,8 +1590,6 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str
 				return nil, errors.Wrap(err, "translating value to timestamp")
 			}
 			results[i] = t.Format(time.RFC3339Nano)
-
-			// results[i] = FormatTimestampNano(int64(val), bsig.Base, field.options.TimeUnit)
 		}
 		result = DistinctTimestamp{Name: fieldName, Values: results}
 		return result, nil
@@ -7919,7 +7917,7 @@ func (vc *ValCount) smaller(other ValCount) ValCount {
 		return vc.decimalSmaller(other)
 	} else if vc.FloatVal != 0 || other.FloatVal != 0 {
 		return vc.floatSmaller(other)
-	} else if !vc.TimestampVal.Equal(time.Time{}) || !other.TimestampVal.Equal(time.Time{}) {
+	} else if !vc.TimestampVal.IsZero() || !other.TimestampVal.IsZero() {
 		return vc.timestampSmaller(other)
 	}
 	if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) {
@@ -7938,6 +7936,8 @@ func (vc *ValCount) smaller(other ValCount) ValCount {
 	}
 }
 
+// timestampSmaller returns the smaller of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) timestampSmaller(other ValCount) ValCount {
 	if other.TimestampVal.Equal(time.Time{}) {
 		return *vc
@@ -7956,6 +7956,8 @@ func (vc *ValCount) timestampSmaller(other ValCount) ValCount {
 	}
 }
 
+// decimalSmaller returns the smaller of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) decimalSmaller(other ValCount) ValCount {
 	if other.DecimalVal == nil {
 		return *vc
@@ -7973,6 +7975,8 @@ func (vc *ValCount) decimalSmaller(other ValCount) ValCount {
 	}
 }
 
+// floatSmaller returns the smaller of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) floatSmaller(other ValCount) ValCount {
 	if vc.Count == 0 || (other.FloatVal < vc.FloatVal && other.Count > 0) {
 		return other
@@ -8012,6 +8016,8 @@ func (vc *ValCount) larger(other ValCount) ValCount {
 	}
 }
 
+// timestampLarger returns the larger of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) timestampLarger(other ValCount) ValCount {
 	if other.TimestampVal.Equal(time.Time{}) {
 		return *vc
@@ -8030,6 +8036,8 @@ func (vc *ValCount) timestampLarger(other ValCount) ValCount {
 	}
 }
 
+// decimalLarger returns the larger of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) decimalLarger(other ValCount) ValCount {
 	if other.DecimalVal == nil {
 		return *vc
@@ -8047,6 +8055,8 @@ func (vc *ValCount) decimalLarger(other ValCount) ValCount {
 	}
 }
 
+// floatLarger returns the larger of the two (vc or other), while merging the count
+// if they are equal.
 func (vc *ValCount) floatLarger(other ValCount) ValCount {
 	if vc.Count == 0 || (other.FloatVal > vc.FloatVal && other.Count > 0) {
 		return other

executor_test.go

@@ -7493,7 +7493,7 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) {
 	}
 }
 
-func toCSV(s string) string {
+func lineBreaker(s string) string {
 	return strings.Join(strings.Split(s, " "), "\n") + "\n"
 }
 
@@ -7553,44 +7553,44 @@ func variousQueriesOnTimeFields(t *testing.T, c *test.Cluster) {
 		// Rows
 		{
 			query:       `Rows(f1, from='2019-08-04T14:36', to='2019-08-04T16:00')`,
-			csvVerifier: toCSV("R4 R5"),
+			csvVerifier: lineBreaker("R4 R5"),
 		},
 		{
 			query:       `Rows(f1, from='2019-08-04T14', to='2019-08-04T17:00')`,
-			csvVerifier: toCSV("R4 R5 R6"),
+			csvVerifier: lineBreaker("R4 R5 R6"),
 		},
 		{
 			query:       `Rows(f1, from='2019-08-04', to='2019-08-05')`,
-			csvVerifier: toCSV("R3 R4 R5 R6"),
+			csvVerifier: lineBreaker("R3 R4 R5 R6"),
 		},
 		{
 			query:       `Rows(f1, from='2019-08', to='2019-12')`,
-			csvVerifier: toCSV("R2 R3 R4 R5 R6 R7"),
+			csvVerifier: lineBreaker("R2 R3 R4 R5 R6 R7"),
 		},
 		{
 			query:       `Rows(f1, from='2019', to='2020')`,
-			csvVerifier: toCSV("R1 R2 R3 R4 R5 R6 R7 R8"),
+			csvVerifier: lineBreaker("R1 R2 R3 R4 R5 R6 R7 R8"),
 		},
 		// Row
 		{
 			query:       `Row(f2='R', from='2019-08-04T14:36', to='2019-08-04T16:00')`,
-			csvVerifier: toCSV("C4 C5"),
+			csvVerifier: lineBreaker("C4 C5"),
 		},
 		{
 			query:       `Row(f2='R', from='2019-08-04T14', to='2019-08-04T17:00')`,
-			csvVerifier: toCSV("C4 C5 C6"),
+			csvVerifier: lineBreaker("C4 C5 C6"),
 		},
 		{
 			query:       `Row(f2='R', from='2019-08-04', to='2019-08-05')`,
-			csvVerifier: toCSV("C3 C4 C5 C6"),
+			csvVerifier: lineBreaker("C3 C4 C5 C6"),
 		},
 		{
 			query:       `Row(f2='R', from='2019-08', to='2019-12')`,
-			csvVerifier: toCSV("C2 C3 C4 C5 C6 C7"),
+			csvVerifier: lineBreaker("C2 C3 C4 C5 C6 C7"),
 		},
 		{
 			query:       `Row(f2='R', from='2019', to='2020')`,
-			csvVerifier: toCSV("C1 C2 C3 C4 C5 C6 C7 C8"),
+			csvVerifier: lineBreaker("C1 C2 C3 C4 C5 C6 C7 C8"),
 		},
 	}
 
@@ -7862,39 +7862,39 @@ userB
 		},
 		{
 			query:       `Row(unix_sec="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_sec="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Row(unix_milli="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_milli="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Row(unix_micro="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_micro="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Row(unix_nano="1833-11-24T17:31:44Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_nano="2106-02-07T06:28:16Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Union(Row(unix_nano="2106-02-07T06:28:16Z"), Row(unix_micro="0001-01-01T00:00:01Z"))`,
-			csvVerifier: toCSV("userA\nuserB"),
+			csvVerifier: lineBreaker("userA\nuserB"),
 		},
 		{
 			query: `Set("userA", unix_milli="2000-12-31T23:59:59.999Z")`,
@@ -8155,59 +8155,59 @@ userC
 		},
 		{
 			query:       `Row(unix_sec_min="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_sec_max="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userC"),
+			csvVerifier: lineBreaker("userC"),
 		},
 		{
 			query:       `Row(unix_sec_max="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_milli_min="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_milli_min="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userC"),
+			csvVerifier: lineBreaker("userC"),
 		},
 		{
 			query:       `Row(unix_micro_max="0001-01-01T00:00:01Z")`,
-			csvVerifier: toCSV("userC"),
+			csvVerifier: lineBreaker("userC"),
 		},
 		{
 			query:       `Row(unix_micro_max="9999-12-31T23:59:59Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_nano_min="1833-11-24T17:31:44Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_nano_min="1969-12-31T23:59:59.999999999Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Row(unix_nano_min="1970-01-01T00:00:00Z")`,
-			csvVerifier: toCSV("userC"),
+			csvVerifier: lineBreaker("userC"),
 		},
 		{
 			query:       `Row(unix_nano_max="2106-02-07T06:28:16Z")`,
-			csvVerifier: toCSV("userA"),
+			csvVerifier: lineBreaker("userA"),
 		},
 		{
 			query:       `Row(unix_nano_max="1970-01-01T00:00:00.000000001Z")`,
-			csvVerifier: toCSV("userB"),
+			csvVerifier: lineBreaker("userB"),
 		},
 		{
 			query:       `Row(unix_nano_max="1970-01-01T00:00:00Z")`,
-			csvVerifier: toCSV("userC"),
+			csvVerifier: lineBreaker("userC"),
 		},
 		{
 			query:       `Union(Row(unix_nano_max="2106-02-07T06:28:16Z"), Row(unix_micro_max="0001-01-01T00:00:01Z"))`,
-			csvVerifier: toCSV("userA\nuserC"),
+			csvVerifier: lineBreaker("userA\nuserC"),
 		},
 		{
 			query: `Set("userA", unix_sec_min="2000-12-31T23:59:59.999Z")`,

field.go

@@ -1833,7 +1833,7 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard
 	// of timestamps are already relative to the epoch (base).
 	// So a user may set an epoch to 2022-03-01 as the start of a race
 	// and import finishing times in seconds.
-	// Timestamps ingested as timestamps are of coure absolute, but by the time
+	// Timestamps ingested as timestamps are of course absolute, but by the time
 	// we get here it would be a relative integer.
 	if f.Type() != FieldTypeTimestamp {
 		min -= bsig.Base

ingest/codec.go

@@ -44,7 +44,7 @@ type Codec interface {
 	AddBoolField(name string) error
 	AddIntField(name string, keys KeyTranslator) error
 	AddDecimalField(name string, scale int64) error
-	AddTimestampField(name string, scale time.Duration, epoch int64) error
+	AddTimestampField(name string, scale string, epoch int64) error
 
 	// Parse data from a reader into the vectors.
 	// This must only be called once on a codec.
@@ -70,14 +70,14 @@ type fieldCodec struct {
 	currentOp *FieldOperation
 	decode    jsonDecFn
 	encode    jsonEncFn
-	// For timestamp: scale-in-nanoseconds; for instance, if scaleUnit is
-	// 1,000,000,000, we are storing numbers-of-seconds since the Unix epoch.
-	// The actual value recorded in BSI will be offset by the field's
-	// epoch, but we don't need to know that.
 	// For decimal: Decimal digits of precision. So for instance, with
 	// scale 2, scaleUnit is 100, "1" is stored as 100 and "1.2" is stored as
 	// 120.
 	scaleUnit int64
+	// For timestamp: we use timeUnit to determine the scale at which
+	// to store a timestamp. For example if the timeUnit is milliseconds
+	// we store the number of milliseconds from the given epoch.
+	timeUnit  string
 	scale     int64
 	epoch     int64    // used only by Timestamp fields
 	scratch   []uint64 // reusable scratch space for sets of values
@@ -282,10 +282,10 @@ func (codec *JSONCodec) AddBoolField(name string) error {
 // passed to this function should be the offset from the Unix epoch to the
 // desired epoch, in the same scale. (So if the scale is milliseconds,
 // it should be the Unix timestamp in seconds, times 1000.)
-func (codec *JSONCodec) AddTimestampField(name string, timeScale time.Duration, epoch int64) error {
+func (codec *JSONCodec) AddTimestampField(name string, timeScale string, epoch int64) error {
 	fieldCodec := &fieldCodec{
 		fieldType: FieldTypeTimeStamp,
-		scaleUnit: int64(timeScale),
+		timeUnit:  timeScale,
 		epoch:     epoch,
 	}
 	fieldCodec.decode = fieldCodec.DecodeTimeValue
@@ -590,7 +590,7 @@ func (j *fieldCodec) DecodeTimeValue(recID uint64, dataType jsonparser.ValueType
 		if err != nil {
 			return fmt.Errorf("parsing timestamp: %w", err)
 		}
-		j.currentOp.AddSignedPair(recID, (stamp.UnixNano()/j.scaleUnit)-j.epoch)
+		j.currentOp.AddSignedPair(recID, TimestampToVal(j.timeUnit, stamp)-j.epoch)
 	case jsonparser.Number:
 		// We could in theory convert this to a time, then convert it
 		// back, by multiplying by scaleUnit, then dividing. Or... not.
@@ -609,7 +609,11 @@ func (j *fieldCodec) EncodeTimeValue(dst *jsonBuffer, values []uint64, signed []
 	if len(signed) == 0 {
 		return errors.New("encoding time value: no value provided")
 	}
-	dst.EncodeTime(time.Unix(0, (signed[0]+j.epoch)*j.scaleUnit).UTC())
+	t, err := ValToTimestamp(j.timeUnit, signed[0]+j.epoch)
+	if err != nil {
+		return errors.Wrap(err, "translating value to timestamp")
+	}
+	dst.EncodeTime(t)
 	return nil
 }
 
@@ -1132,3 +1136,35 @@ type errFieldNotFound struct {
 func (err errFieldNotFound) Error() string {
 	return fmt.Sprintf("field not found: %q", err.field)
 }
+
+// TimestampToVal takes a time unit and a time.Time and converts it to an integer value
+func TimestampToVal(unit string, ts time.Time) int64 {
+	switch unit {
+	case "s":
+		return ts.Unix()
+	case "ms":
+		return ts.UnixMilli()
+	case "us":
+		return ts.UnixMicro()
+	case "ns":
+		return ts.UnixNano()
+	}
+	return 0
+
+}
+
+// ValToTimestamp takes a timeunit and an integer value and converts it to time.Time
+func ValToTimestamp(unit string, val int64) (time.Time, error) {
+	switch unit {
+	case "s":
+		return time.Unix(val, 0).UTC(), nil
+	case "ms":
+		return time.UnixMilli(val).UTC(), nil
+	case "us", "μs":
+		return time.UnixMicro(val).UTC(), nil
+	case "ns":
+		return time.Unix(0, val).UTC(), nil
+	default:
+		return time.Time{}, errors.Errorf("Unknown time unit: '%v'", unit)
+	}
+}

ingest/codec_test.go

@@ -58,7 +58,7 @@ func TestEncode(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't parse sample epoch time: %v", err)
 	}
-	_ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000)
+	_ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000)
 	_ = codec.AddDecimalField("dec", 2)
 	_ = codec.AddBoolField("bool")
 
@@ -74,7 +74,7 @@ func TestEncode(t *testing.T) {
 	_ = codec.AddTimeQuantumField("tqkeys", newStableTranslator())
 	_ = codec.AddIntField("int", nil)
 	_ = codec.AddIntField("intkeys", newStableTranslator())
-	_ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000)
+	_ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000)
 	_ = codec.AddDecimalField("dec", 2)
 	_ = codec.AddBoolField("bool")
 
@@ -291,7 +291,7 @@ func TestCodecErrors(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't parse sample epoch time: %v", err)
 	}
-	_ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000)
+	_ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000)
 	_ = codec.AddDecimalField("dec", 2)
 	_ = codec.AddBoolField("bool")
 
@@ -475,7 +475,7 @@ func TestSimpleCodec(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't parse sample epoch time: %v", err)
 	}
-	_ = codec.AddTimestampField("ts", time.Millisecond, epoch.Unix()*1000)
+	_ = codec.AddTimestampField("ts", "ms", epoch.Unix()*1000)
 	_ = codec.AddDecimalField("dec", 2)
 	_ = codec.AddBoolField("bool")
 	var nextShard = uint64(1<<shardwidth.Exponent) + 5