Merge pull request #1772 from molecula/sup-102

[SUP-102] fix presentation of timestamps in Groupby, Distinct calls from psql client

executor.go

@@ -254,7 +254,7 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
 		}
 	}
 	// Must copy out of Tx data before Commiting, because it will become invalid afterwards.
-	respSafeNoTxData := e.safeCopy(resp)
+	respSafeNoTxData := safeCopy(resp)
 
 	// Commit transactions if writing; else let the defer grp.Abort do the rollbacks.
 	if needWriteTxn {
@@ -267,7 +267,7 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
 
 // safeCopy copies everything in resp that has Bitmap material,
 // to avoid anything coming from the mmap-ed Tx storage.
-func (e *executor) safeCopy(resp QueryResponse) (out QueryResponse) {
+func safeCopy(resp QueryResponse) (out QueryResponse) {
 	out = QueryResponse{
 		Err:     resp.Err,     //  error
 		Profile: resp.Profile, //  *tracing.Profile
@@ -322,6 +322,8 @@ func (e *executor) safeCopy(resp QueryResponse) (out QueryResponse) {
 				safe[i] = v.Clone()
 			}
 			out.Results = append(out.Results, safe)
+		case DistinctTimestamp:
+			out.Results = append(out.Results, x)
 		default:
 			panic(fmt.Sprintf("handle %T here", v))
 		}
@@ -1498,6 +1500,7 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str
 	if field == nil {
 		return nil, ErrFieldNotFound
 	}
+
 	bsig := field.bsiGroup(fieldName)
 	if bsig == nil {
 		result = &Row{
@@ -1534,9 +1537,25 @@ func (e *executor) executeDistinctShard(ctx context.Context, qcx *Qcx, index str
 	if bsig == nil {
 		return executeDistinctShardSet(ctx, qcx, idx, fieldName, shard, filterBitmap)
 	}
+	if field.Options().Type == FieldTypeTimestamp {
+		r, err := executeDistinctShardBSI(ctx, qcx, idx, fieldName, shard, bsig, filterBitmap)
+		if err != nil {
+			return nil, err
+		}
+		results := make([]string, len(r.Pos.Columns()))
+		for i, val := range r.Pos.Columns() {
+			results[i] = FormatTimestampNano(int64(val), bsig.Base, field.options.TimeUnit)
+		}
+		return DistinctTimestamp{Name: fieldName, Values: results}, nil
+	}
 	return executeDistinctShardBSI(ctx, qcx, idx, fieldName, shard, bsig, filterBitmap)
 }
 
+type DistinctTimestamp struct {
+	Values []string
+	Name   string
+}
+
 func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldName string, shard uint64, filterBitmap *roaring.Bitmap) (result *Row, err0 error) {
 	index := idx.Name()
 	tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard})
@@ -3045,10 +3064,11 @@ func applyLimitAndOffsetToGroupByResult(c *pql.Call, results []GroupCount) ([]Gr
 
 // FieldRow is used to distinguish rows in a group by result.
 type FieldRow struct {
-	Field  string `json:"field"`
-	RowID  uint64 `json:"rowID"`
-	RowKey string `json:"rowKey,omitempty"`
-	Value  *int64 `json:"value,omitempty"`
+	Field        string        `json:"field"`
+	RowID        uint64        `json:"rowID"`
+	RowKey       string        `json:"rowKey,omitempty"`
+	Value        *int64        `json:"value,omitempty"`
+	FieldOptions *FieldOptions `json:"-"`
 }
 
 func (fr *FieldRow) Clone() (clone *FieldRow) {
@@ -3062,6 +3082,11 @@ func (fr *FieldRow) Clone() (clone *FieldRow) {
 		v := *fr.Value
 		clone.Value = &v
 	}
+	if fr.FieldOptions != nil {
+		// deep copy, for Extra Safety
+		v := *fr.FieldOptions
+		clone.FieldOptions = &v
+	}
 	return
 }
 
@@ -7719,6 +7744,8 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children
 			return nil, newNotFoundError(ErrFieldNotFound, fieldName)
 		}
 		gbi.fields[i].Field = fieldName
+		options := field.Options()
+		gbi.fields[i].FieldOptions = &options
 
 		switch field.Type() {
 		case FieldTypeSet, FieldTypeMutex, FieldTypeBool:
@@ -7952,7 +7979,6 @@ func (gbi *groupByIterator) Next(ctx context.Context) (ret GroupCount, done bool
 	ret.Group = make([]FieldRow, len(gbi.rows))
 	copy(ret.Group, gbi.fields)
 	for i, r := range gbi.rows {
-
 		ret.Group[i].RowID = r.id
 		ret.Group[i].Value = r.value
 	}

executor_internal_test.go

@@ -475,3 +475,15 @@ func TestGetSorter(t *testing.T) {
 		})
 	}
 }
+
+func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) {
+	result := DistinctTimestamp{Values: []string{"test", "test"}, Name: "test"}
+	results := make([]interface{}, 1)
+	results[0] = result
+
+	response := QueryResponse{Results: results, Err: nil, Profile: nil}
+	copied := safeCopy(response)
+	if !reflect.DeepEqual(copied.Results, response.Results) {
+		t.Fatalf("Did not copy results. got %+v, want %+v", copied.Results, response.Results)
+	}
+}

server/pg.go

@@ -155,6 +155,24 @@ type PilosaQueryHandler struct {
 	sqlVersion SqlVersion
 }
 
+func pgWriteDistinctTimestamp(w pg.QueryResultWriter, val pilosa.DistinctTimestamp) error {
+	err := w.WriteHeader(pg.ColumnInfo{
+		Name: val.Name,
+		Type: pg.TypeCharoid,
+	})
+	if err != nil {
+		return errors.Wrap(err, "writing result header")
+	}
+
+	for _, k := range val.Values {
+		err = w.WriteRowText(k)
+		if err != nil {
+			return errors.Wrap(err, "writing key")
+		}
+	}
+	return nil
+}
+
 func pgWriteRow(w pg.QueryResultWriter, row *pilosa.Row) error {
 	err := w.WriteHeader(pg.ColumnInfo{
 		Name: "_id",
@@ -313,7 +331,11 @@ func pgWriteGroupCount(w pg.QueryResultWriter, counts *pilosa.GroupCounts) error
 			var v string
 			switch {
 			case g.Value != nil:
-				v = strconv.FormatInt(*g.Value, 10)
+				if g.FieldOptions.Type == pilosa.FieldTypeTimestamp {
+					v = pilosa.FormatTimestampNano(int64(*g.Value), g.FieldOptions.Base, g.FieldOptions.TimeUnit)
+				} else {
+					v = strconv.FormatInt(*g.Value, 10)
+				}
 			case g.RowKey != "":
 				v = g.RowKey
 			default:
@@ -551,7 +573,8 @@ func pgWriteResult(w pg.QueryResultWriter, result interface{}) error {
 		}
 
 		return nil
-
+	case pilosa.DistinctTimestamp:
+		return pgWriteDistinctTimestamp(w, result)
 	case nil:
 		return nil
 

server/pg_internal_test.go

@@ -0,0 +1,67 @@
+// Copyright 2021 Pilosa 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 server
+
+import (
+	"testing"
+
+	pilosa "github.com/molecula/featurebase/v2"
+	"github.com/molecula/featurebase/v2/pg"
+)
+
+// pg_internal_test.go tests unexported methods from server/pg.go
+
+// TestQueryResultWriter implements the QueryResultWriter interface for testing
+type TestQueryResultWriter struct {
+	Header  []pg.ColumnInfo
+	RowText []string
+	TagTag  string
+}
+
+func (t *TestQueryResultWriter) WriteHeader(headers ...pg.ColumnInfo) error {
+	for _, header := range headers {
+		t.Header = append(t.Header, header)
+	}
+	return nil
+}
+
+func (t *TestQueryResultWriter) WriteRowText(rowTexts ...string) error {
+	for _, rowText := range rowTexts {
+		t.RowText = append(t.RowText, rowText)
+	}
+	return nil
+}
+
+func (t *TestQueryResultWriter) Tag(tag string) {
+	t.TagTag = tag
+}
+
+func TestPgWriteDistinctTimestamp(t *testing.T) {
+	w := TestQueryResultWriter{}
+	expected := pilosa.DistinctTimestamp{Name: "test", Values: []string{"date1", "date2", "date3"}}
+	pgWriteDistinctTimestamp(&w, expected)
+	if w.Header[0].Name != expected.Name {
+		t.Fatalf("Header Name is wrong. got %v, want %v", w.Header[0], expected.Name)
+	}
+	if w.Header[0].Type != pg.TypeCharoid {
+		t.Fatalf("Header Type is wrong. got %v, want %v", w.Header[0].Type, pg.TypeCharoid)
+	}
+	for i, value := range w.RowText {
+		if value != expected.Values[i] {
+			t.Fatalf("Value not written properly. got %v, want %v", value, expected.Values[i])
+		}
+	}
+
+}

test/pilosa.go

@@ -26,7 +26,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v2"
+	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/encoding/proto"
 	"github.com/molecula/featurebase/v2/http"
@@ -313,7 +313,27 @@ func CheckGroupBy(t *testing.T, expected, results []pilosa.GroupCount) {
 		t.Fatalf("number of groupings mismatch:\n got:%+v\nwant:%+v\n", results, expected)
 	}
 	for i, result := range results {
-		if !reflect.DeepEqual(expected[i], result) {
+		// have to check each field Row individually because FieldOptions is getting set
+		for j := range expected[i].Group {
+			// Field:"ppa", RowID:0x3, RowKey:"", Value:(*int64)(nil), FieldOptions:
+			if !reflect.DeepEqual(expected[i].Group[j].Field, result.Group[j].Field) {
+				t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
+			}
+			if !reflect.DeepEqual(expected[i].Group[j].RowKey, result.Group[j].RowKey) {
+				t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
+			}
+			if !reflect.DeepEqual(expected[i].Group[j].Value, result.Group[j].Value) {
+				t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
+			}
+		}
+
+		if !reflect.DeepEqual(expected[i].Count, result.Count) {
+			t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
+		}
+		if !reflect.DeepEqual(expected[i].Agg, result.Agg) {
+			t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
+		}
+		if !reflect.DeepEqual(expected[i].DecimalAgg, result.DecimalAgg) {
 			t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
 		}
 	}

util.go

@@ -115,8 +115,9 @@ func roaringFragmentHasData(path string, index, field, view string, shard uint64
 	return
 }
 
-// GetLoopProgress returns the estimated remaining time to iterate through some items
-// as well as the loop completion percentage with the following parameters:
+// GetLoopProgress returns the estimated remaining time to iterate through some
+// items as well as the loop completion percentage with the following
+// parameters:
 // the start time, the current time, the iteration, and the number of items
 func GetLoopProgress(start time.Time, now time.Time, iteration uint, total uint) (remaining time.Duration, pctDone float64) {
 	itemsLeft := total - (iteration + 1)
@@ -124,3 +125,9 @@ func GetLoopProgress(start time.Time, now time.Time, iteration uint, total uint)
 	pctDone = (float64(iteration+1) / float64(total)) * 100
 	return time.Duration(avgItemTime * float64(itemsLeft)), pctDone
 }
+
+// FormatTimestampNano returns the string representation of a timestamp given:
+// an epoch value, base, and time unit
+func FormatTimestampNano(value, base int64, timeUnit string) string {
+	return time.Unix(0, (value+base)*TimeUnitNanos(timeUnit)).UTC().Format(time.RFC3339Nano)
+}

util_test.go

@@ -85,3 +85,21 @@ func TestGetLoopProgress(t *testing.T) {
 		})
 	}
 }
+
+func TestFormatTimestampNano(t *testing.T) {
+	if FormatTimestampNano(0, 69, "s") != "1970-01-01T00:01:09Z" {
+		t.Fatal("Timestamp not formatted properly")
+	}
+	if FormatTimestampNano(0, 420, "ms") != "1970-01-01T00:00:00.42Z" {
+		t.Fatal("Timestamp not formatted properly")
+	}
+	if FormatTimestampNano(420, 0, "μs") != "1970-01-01T00:00:00.00000042Z" {
+		t.Fatal("Timestamp not formatted properly")
+	}
+	if FormatTimestampNano(420, 69, "us") != "1970-01-01T00:00:00.000489Z" {
+		t.Fatal("Timestamp not formatted properly")
+	}
+	if FormatTimestampNano(69, 420, "ns") != "1970-01-01T00:00:00.000000489Z" {
+		t.Fatal("Timestamp not formatted properly")
+	}
+}