Merge pull request #1344 from nagamocha3000/bnm-fix-1080

Fix filtering on time fields in Rows() embedded within GroupBy
2026-09-11 15:21:02 +00:00 · 2021-01-22 19:56:08 +03:00 · 2021-01-22 19:56:08 +03:00 · 6c3c460c83
commit 6c3c460c83
parent b273f3ba60 b5fdc6609a
4 changed files with 271 additions and 12 deletions
--- 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")
--- 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) {
--- 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
--- 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