From 0611ef3418643eabebfca3264b266dbc59953be9 Mon Sep 17 00:00:00 2001
From: Jaden Weiss <jaden@jadendw.dev>
Date: Tue, 7 Jul 2020 15:32:34 -0400
Subject: [PATCH] initial implementation of Rows like

---
 executor.go | 13 +++++++++++++
 fragment.go | 16 ++++++++++++++++
 pql/ast.go  | 14 ++++++++++++++
 3 files changed, 43 insertions(+)

diff --git a/executor.go b/executor.go
index dcfaf39ed..2c736badd 100644
--- a/executor.go
+++ b/executor.go
@@ -2431,6 +2431,14 @@ func (e *executor) executeRowsShard(ctx context.Context, tx Tx, index string, fi
 		limit = int(lim)
 	}
 
+	var likeErr chan error
+	if like, hasLike, err := c.StringArg("like"); err != nil {
+		return nil, errors.Wrap(err, "getting like pattern")
+	} else if hasLike {
+		likeErr = make(chan error, 1)
+		filters = append(filters, filterLike(like, f.TranslateStore(), likeErr))
+	}
+
 	for _, view := range views {
 		if err := ctx.Err(); err != nil {
 			return nil, err
@@ -2444,6 +2452,11 @@ func (e *executor) executeRowsShard(ctx context.Context, tx Tx, index string, fi
 		if err != nil {
 			return nil, err
 		}
+		select {
+		case err = <-likeErr:
+			return nil, err
+		default:
+		}
 		rowIDs = rowIDs.merge(viewRows, limit)
 	}
 
diff --git a/fragment.go b/fragment.go
index b31e8d8ff..d052a3a3c 100644
--- a/fragment.go
+++ b/fragment.go
@@ -2802,6 +2802,22 @@ func filterColumn(col uint64) rowFilter {
 	}
 }
 
+func filterLike(like string, t TranslateStore, e chan error) rowFilter {
+	plan := planLike(like)
+
+	return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
+		keyStr, err := t.TranslateID(rowID)
+		if err != nil {
+			select {
+			case e <- err:
+			default:
+			}
+			return false, true
+		}
+		return matchLike(keyStr, plan...), false
+	}
+}
+
 // TODO: this works, but it would be more performant if the fragment could seek
 // to the next row in the rows list rather than asking the filter for each
 // container serially. The container iterator would need to expose a seek
diff --git a/pql/ast.go b/pql/ast.go
index e93a2aaf5..4195e66bf 100644
--- a/pql/ast.go
+++ b/pql/ast.go
@@ -384,6 +384,7 @@ var callInfoByFunc = map[string]callInfo{
 			"previous": nil,
 			"from":     nil,
 			"to":       nil,
+			"like":     "",
 		},
 	},
 	"Shift": {allowUnknown: false,
@@ -662,6 +663,19 @@ func (c *Call) UintSliceArg(key string) ([]uint64, bool, error) {
 	}
 }
 
+func (c *Call) StringArg(key string) (string, bool, error) {
+	val, ok := c.Args[key]
+	if !ok {
+		return "", false, nil
+	}
+	switch tval := val.(type) {
+	case string:
+		return tval, true, nil
+	default:
+		return "", true, fmt.Errorf("unexpected type %T in StringArg, val %v", tval, tval)
+	}
+}
+
 // CallArg is for reading the value at key from call.Args as a Call. If the
 // key is not in Call.Args, the value of the returned value will be nil, and
 // the error will be nil. An error is returned if the value is not a Call.