diff --git a/executor.go b/executor.go index f99a0f667..74e565aed 100644 --- a/executor.go +++ b/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 } diff --git a/executor_internal_test.go b/executor_internal_test.go index b752957ef..ce860c46c 100644 --- a/executor_internal_test.go +++ b/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) + } +} diff --git a/server/pg.go b/server/pg.go index c2522f286..169f2e747 100644 --- a/server/pg.go +++ b/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 diff --git a/server/pg_internal_test.go b/server/pg_internal_test.go new file mode 100644 index 000000000..75024d509 --- /dev/null +++ b/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]) + } + } + +} diff --git a/test/pilosa.go b/test/pilosa.go index d1deb491e..48659587f 100644 --- a/test/pilosa.go +++ b/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]) } } diff --git a/util.go b/util.go index 6d4b8e09e..1db0c74fc 100644 --- a/util.go +++ b/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) +} diff --git a/util_test.go b/util_test.go index c659053c5..14ed78c36 100644 --- a/util_test.go +++ b/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") + } +}