diff --git a/executor.go b/executor.go index ae1071b33..0c02fd6c8 100644 --- a/executor.go +++ b/executor.go @@ -7381,9 +7381,11 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children idx := holder.Index(index) var ( - fieldName string - viewName string - ok bool + fieldName string + viewName string + ok bool + views []string + isTimeField bool ) ignorePrev := false for i, call := range children { @@ -7397,9 +7399,42 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children gbi.fields[i].Field = fieldName switch field.Type() { - case FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool: + case FieldTypeSet, FieldTypeMutex, FieldTypeBool: viewName = viewStandard + case FieldTypeTime: + var ( + err error + v interface{} + ) + // Parse "from" time, if set. + var ( + hasFrom bool + fromTime time.Time + ) + if v, hasFrom = call.Args["from"]; hasFrom { + if fromTime, err = parseTime(v); err != nil { + return nil, errors.Wrap(err, "parsing from time") + } + } + + // Parse "to" time, if set. + var ( + hasTo bool + toTime time.Time + ) + if v, hasTo = call.Args["to"]; hasTo { + if toTime, err = parseTime(v); err != nil { + return nil, errors.Wrap(err, "parsing to time") + } + } + + if hasTo || hasFrom { + views = viewsByTimeRange(viewStandard, fromTime, toTime, field.TimeQuantum()) + isTimeField = true + } else { + viewName = viewStandard + } case FieldTypeInt: viewName = viewBSIGroupPrefix + fieldName @@ -7408,11 +7443,6 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children call.Name, strings.Join([]string{FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool, FieldTypeInt}, ",")) } - // Fetch fragment. - frag := holder.fragment(index, fieldName, viewName, shard) - if frag == nil { // this means this whole shard doesn't have all it needs to continue - return nil, nil - } filters := []roaring.BitmapFilter{} if len(rowIDs[i]) > 0 { filters = append(filters, roaring.NewBitmapRowsFilter(rowIDs[i])) @@ -7424,9 +7454,35 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children } defer finisher(&err0) - gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...) - if err != nil { - return nil, err + // Fetch fragment(s), get rowIterator + if isTimeField { + var fragments []*fragment + for _, viewName := range views { + fragment := holder.fragment(index, fieldName, viewName, shard) + if fragment != nil { + fragments = append(fragments, fragment) + } + } + if len(fragments) == 0 { + // whole shard doesn't have all it needs to continue ? + return nil, nil + } + + gbi.rowIters[i], err = timeFragmentsRowIterator(fragments, tx, i != 0, filters...) + if err != nil { + return nil, err + } + } else { + frag := holder.fragment(index, fieldName, viewName, shard) + if frag == nil { // this means this whole shard doesn't have all it needs to continue + return nil, nil + } + + gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...) + if err != nil { + return nil, err + } + } prev, hasPrev, err := call.UintArg("previous") diff --git a/executor_test.go b/executor_test.go index 552aed964..c0532b4f9 100644 --- a/executor_test.go +++ b/executor_test.go @@ -6863,6 +6863,30 @@ func variousQueries(t *testing.T, clusterSize int) { {"icecream", "userF"}, }) + // Create and populate "places_visited" time field. + c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM"))) + ts2019Jan01 := int64(1546300800) * 1e+9 // 2019 January 1st 0:00:00 + ts2019Aug01 := int64(1564617600) * 1e+9 // 2019 August 1st 0:00:00 + ts2020Jan01 := int64(1577836800) * 1e+9 // 2020 January 1st 0:00:00 + c.ImportTimeQuantumKey(t, "users", "places_visited", []test.TimeQuantumKey{ + // 2019 January: nairobi, paris, austin, toronto + {RowKey: "nairobi", ColKey: "userB", Ts: ts2019Jan01}, + {RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01}, + {RowKey: "austin", ColKey: "userF", Ts: ts2019Jan01}, + {RowKey: "toronto", ColKey: "userA", Ts: ts2019Jan01}, + // 2019 August: toronto only + {RowKey: "toronto", ColKey: "userB", Ts: ts2019Aug01}, + {RowKey: "toronto", ColKey: "userC", Ts: ts2019Aug01}, + // 2020: toronto, mombasa, sydney, nairobi + {RowKey: "toronto", ColKey: "userB", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userD", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userE", Ts: ts2020Jan01}, + {RowKey: "toronto", ColKey: "userF", Ts: ts2020Jan01}, + {RowKey: "mombasa", ColKey: "userA", Ts: ts2020Jan01}, + {RowKey: "sydney", ColKey: "userD", Ts: ts2020Jan01}, + {RowKey: "nairobi", ColKey: "userE", Ts: ts2020Jan01}, + }) + // Create and populate "affinity" int field with negative, positive, zero and null values. c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000)) c.ImportIntKey(t, "users", "affinity", []test.IntKey{ @@ -6900,6 +6924,66 @@ func variousQueries(t *testing.T, clusterSize int) { qrVerifier func(t *testing.T, resp pilosa.QueryResponse) csvVerifier string }{ + { // 2020 & 2019 All + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59'))`, + csvVerifier: `nairobi,2 +paris,1 +austin,1 +toronto,6 +mombasa,1 +sydney,1 +`, + }, + { // 2019 January only + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`, + csvVerifier: `nairobi,1 +paris,1 +austin,1 +toronto,1 +`, + }, + { // 2019 All + query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'))`, + csvVerifier: `nairobi,1 +paris,1 +austin,1 +toronto,3 +`, + }, + { // 2019 All, this excludes userC (who likes pangolin & icecream) from the count. + // UserC visited Paris and Toronto in 2019 + query: `GroupBy( + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))) + )`, + csvVerifier: `nairobi,1 +austin,1 +toronto,2 +`, + }, + { // After excluding UserC, this gets the sum of the networth of everyone per cities travelled + query: `GroupBy( + Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), + filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))), + aggregate=Sum(field=net_worth) + )`, + csvVerifier: `nairobi,1,10 +austin,1,100000 +toronto,2,11 +`, + }, + { // 2020 & 2019 All + query: `Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\nmombasa\nsydney\n", + }, + { // 2019 All + query: `Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\n", + }, + { // 2019 January only + query: `Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00')`, + csvVerifier: "nairobi\nparis\naustin\ntoronto\n", + }, { query: "Count(All())", qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) { diff --git a/fragment.go b/fragment.go index aa93dcc9a..8592182f4 100644 --- a/fragment.go +++ b/fragment.go @@ -3329,6 +3329,95 @@ func (f *fragment) rowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFilter return f.setRowIterator(tx, wrap, filters...) } +type timeRowIterator struct { + tx Tx + cur int + wrap bool + allRowIDs []uint64 + rowIDToFragments map[uint64][]*fragment +} + +func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters ...roaring.BitmapFilter) (rowIterator, error) { + if len(fragments) == 0 { + return nil, fmt.Errorf("there should be at least 1 fragment") + } else if len(fragments) == 1 { + return fragments[0].setRowIterator(tx, wrap, filters...) + } + + it := &timeRowIterator{ + tx: tx, + cur: 0, + wrap: wrap, + } + + // create a sort of inverted index that maps each + // rowID back to the fragments that have that rowID + rowIDToFragments := make(map[uint64][]*fragment) + for _, f := range fragments { + rowIDs, err := f.rows(context.Background(), tx, 0, filters...) + if err != nil { + return nil, err + } + for _, rowID := range rowIDs { + fs := append(rowIDToFragments[rowID], f) + rowIDToFragments[rowID] = fs + } + } + + // if len(rowIDToFragments) == 0 what to do ?? + // ie all fragments returned empty rowIDs, is this possible + // is this an error + + // collect all rowIDs from inverted index to a slice + allRowIDs := make([]uint64, len(rowIDToFragments)) + i := 0 + for rowID := range rowIDToFragments { + allRowIDs[i] = rowID + i++ + } + sort.Slice(allRowIDs, func(i, j int) bool { return allRowIDs[i] < allRowIDs[j] }) + + it.rowIDToFragments = rowIDToFragments + it.allRowIDs = allRowIDs + + return it, nil +} + +func (it *timeRowIterator) Seek(rowID uint64) { + idx := sort.Search(len(it.allRowIDs), func(i int) bool { + return it.allRowIDs[i] >= rowID + }) + it.cur = idx +} + +func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool, err error) { + if it.cur >= len(it.allRowIDs) { + if !it.wrap || len(it.allRowIDs) == 0 { + return nil, 0, nil, true, nil + } + it.Seek(0) + wrapped = true + } + + // gather rows + rowID = it.allRowIDs[it.cur] + fragments := it.rowIDToFragments[rowID] + rows := make([]*Row, 0, len(fragments)) + for _, fragment := range fragments { + row, err := fragment.row(it.tx, rowID) + if err != nil { + return row, rowID, nil, wrapped, err + } + rows = append(rows, row) + } + + // union rows + r = rows[0].Union(rows[1:]...) + + it.cur++ + return r, rowID, nil, wrapped, nil +} + type intRowIterator struct { f *fragment values int64Slice // sorted slice of int values diff --git a/test/cluster.go b/test/cluster.go index 006729ced..3af5c9b23 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -164,6 +164,36 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys } } +// TimeQuantumKey is a string key and a string+key value +type TimeQuantumKey struct { + RowKey string + ColKey string + Ts int64 +} + +// ImportTimeQuantumKey imports data into an index where the index is keyd +// and the field is a time-quantum +func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entries []TimeQuantumKey) { + t.Helper() + importRequest := &pilosa.ImportRequest{ + Index: index, + Field: field, + RowKeys: make([]string, len(entries)), + ColumnKeys: make([]string, len(entries)), + Timestamps: make([]int64, len(entries)), + } + for i, entry := range entries { + importRequest.ColumnKeys[i] = entry.ColKey + importRequest.RowKeys[i] = entry.RowKey + importRequest.Timestamps[i] = entry.Ts + + } + err := c.Nodes[0].API.Import(context.Background(), nil, importRequest) + if err != nil { + t.Fatalf("importing keykey data: %v", err) + } +} + // IntKey is a string key and a signed integer value. type IntKey struct { Val int64