Repository: lf-edge/ekuiper
Pull request #912 (2021-07-29 05:56:29)

feat(sql): support index expression for array


Patch:
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index de841aac..535c158e 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -26,7 +26,7 @@ basic:
   # The URL where hosts all of pre-build plugins. By default it's at packages.emqx.io
   # There could be several hosts (host can be separated with comma), if same package could be found in the several hosts,
   # then the package in the 1st host will have the highest priority.
-  pluginHosts: https://www.emqx.io/downloads
+  pluginHosts: https://packages.emqx.net
 
 # The default options for all rules. Each rule can override this setting by defining its own option
 rule:
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 01fe0357..7792006c 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -285,7 +285,6 @@ func (p *Parser) parseJoins() (ast.Joins, error) {
 			return nil, nil
 		}
 	}
-	return joins, nil
 }
 
 func (p *Parser) ParseJoin(joinType ast.JoinType) (*ast.Join, error) {
@@ -492,8 +491,6 @@ func (p *Parser) ParseExpr() (ast.Expr, error) {
 			node = r
 		}
 	}
-
-	return nil, nil
 }
 
 func (p *Parser) parseUnaryExpr(isSubField bool) (ast.Expr, error) {
@@ -572,7 +569,7 @@ func (p *Parser) parseBracketExpr() (ast.Expr, error) {
 	tok2, lit2 := p.scanIgnoreWhiteSpaceWithNegativeNum()
 	if tok2 == ast.RBRACKET {
 		//field[]
-		return &ast.ColonExpr{Start: 0, End: math.MinInt32}, nil
+		return &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
 	} else if tok2 == ast.INTEGER {
 		start, err := strconv.Atoi(lit2)
 		if err != nil {
@@ -580,19 +577,32 @@ func (p *Parser) parseBracketExpr() (ast.Expr, error) {
 		}
 		if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
 			//Such as field[2]
-			return &ast.IndexExpr{Index: start}, nil
+			return &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: start}}, nil
 		} else if tok3 == ast.COLON {
 			//Such as field[2:] or field[2:4]
-			return p.parseColonExpr(start)
+			return p.parseColonExpr(&ast.IntegerLiteral{Val: start})
 		}
 	} else if tok2 == ast.COLON {
 		//Such as field[:3] or [:]
-		return p.parseColonExpr(0)
+		return p.parseColonExpr(&ast.IntegerLiteral{Val: 0})
+	} else {
+		p.unscan()
+		start, err := p.ParseExpr()
+		if err != nil {
+			return nil, fmt.Errorf("The start index %s is invalid in bracket expression.", lit2)
+		}
+		if tok3, _ := p.scanIgnoreWhitespace(); tok3 == ast.RBRACKET {
+			//Such as field[2]
+			return &ast.IndexExpr{Index: start}, nil
+		} else if tok3 == ast.COLON {
+			//Such as field[2:] or field[2:4]
+			return p.parseColonExpr(start)
+		}
 	}
 	return nil, fmt.Errorf("Unexpected token %q. when parsing bracket expressions.", lit2)
 }
 
-func (p *Parser) parseColonExpr(start int) (ast.Expr, error) {
+func (p *Parser) parseColonExpr(start ast.Expr) (ast.Expr, error) {
 	tok, lit := p.scanIgnoreWhiteSpaceWithNegativeNum()
 	if tok == ast.INTEGER {
 		end, err := strconv.Atoi(lit)
@@ -601,12 +611,12 @@ func (p *Parser) parseColonExpr(start int) (ast.Expr, error) {
 		}
 
 		if tok1, lit1 := p.scanIgnoreWhitespace(); tok1 == ast.RBRACKET {
-			return &ast.ColonExpr{Start: start, End: end}, nil
+			return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: end}}, nil
 		} else {
 			return nil, fmt.Errorf("Found %q, expected right bracket.", lit1)
 		}
 	} else if tok == ast.RBRACKET {
-		return &ast.ColonExpr{Start: start, End: math.MinInt32}, nil
+		return &ast.ColonExpr{Start: start, End: &ast.IntegerLiteral{Val: math.MinInt32}}, nil
 	}
 	return nil, fmt.Errorf("Found %q, expected right bracket.", lit)
 }
@@ -677,7 +687,7 @@ func (p *Parser) parseCall(name string) (ast.Expr, error) {
 	if tok, lit := p.scanIgnoreWhitespace(); tok != ast.RPAREN {
 		return nil, fmt.Errorf("found function call %q, expected ), but with %q.", name, lit)
 	}
-	if wt, error := validateWindows(name, args); wt == ast.NOT_WINDOW {
+	if wt, err := validateWindows(name, args); wt == ast.NOT_WINDOW {
 		if valErr := validateFuncs(name, args); valErr != nil {
 			return nil, valErr
 		}
@@ -687,12 +697,12 @@ func (p *Parser) parseCall(name string) (ast.Expr, error) {
 		}
 		return &ast.Call{Name: name, Args: args}, nil
 	} else {
-		if error != nil {
-			return nil, error
+		if err != nil {
+			return nil, err
 		}
 		win, err := p.ConvertToWindows(wt, args)
 		if err != nil {
-			return nil, error
+			return nil, err
 		}
 		// parse filter clause
 		f, err := p.parseFilter()
@@ -1230,7 +1240,7 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 							if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
 								return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, tok1)
 							} else {
-								opts.STRICT_VALIDATION = (val == "TRUE")
+								opts.STRICT_VALIDATION = val == "TRUE"
 							}
 						case ast.RETAIN_SIZE:
 							if val, err := strconv.Atoi(lit3); err != nil {
@@ -1242,7 +1252,7 @@ func (p *Parser) parseStreamOptions() (*ast.Options, error) {
 							if val := strings.ToUpper(lit3); (val != "TRUE") && (val != "FALSE") {
 								return nil, fmt.Errorf("found %q, expect TRUE/FALSE value in %s option.", lit3, tok1)
 							} else {
-								opts.SHARED = (val == "TRUE")
+								opts.SHARED = val == "TRUE"
 							}
 						default:
 							f := v.Elem().FieldByName(lit1)
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index 93722edd..822aaf55 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -212,7 +212,6 @@ func validate(t string, v interface{}) error {
 	default:
 		return fmt.Errorf("incompatible types for comparison: %s and %s", t, vt)
 	}
-	return nil
 }
 
 type EvalResultMessage struct {
@@ -354,9 +353,23 @@ func (v *ValuerEval) Eval(expr ast.Expr) interface{} {
 	case *ast.BooleanLiteral:
 		return expr.Val
 	case *ast.ColonExpr:
-		return &BracketEvalResult{Start: expr.Start, End: expr.End}
+		s, e := v.Eval(expr.Start), v.Eval(expr.End)
+		si, err := cast.ToInt(s, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return fmt.Errorf("colon start %v is not int: %v", expr.Start, err)
+		}
+		ei, err := cast.ToInt(e, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return fmt.Errorf("colon end %v is not int: %v", expr.End, err)
+		}
+		return &BracketEvalResult{Start: si, End: ei}
 	case *ast.IndexExpr:
-		return &BracketEvalResult{Start: expr.Index, End: expr.Index}
+		i := v.Eval(expr.Index)
+		ii, err := cast.ToInt(i, cast.CONVERT_SAMEKIND)
+		if err != nil {
+			return fmt.Errorf("index %v is not int: %v", expr.Index, err)
+		}
+		return &BracketEvalResult{Start: ii, End: ii}
 	case *ast.Call:
 		if valuer, ok := v.Valuer.(CallValuer); ok {
 			switch expr.Name {
@@ -972,8 +985,6 @@ func (v *ValuerEval) simpleDataEval(lhs, rhs interface{}, op ast.Token) interfac
 	default:
 		return invalidOpError(lhs, op, rhs)
 	}
-
-	return invalidOpError(lhs, op, rhs)
 }
 
 func invalidOpError(lhs interface{}, op ast.Token, rhs interface{}) error {
diff --git a/pkg/ast/expr.go b/pkg/ast/expr.go
index 8b5a014c..6b96549f 100644
--- a/pkg/ast/expr.go
+++ b/pkg/ast/expr.go
@@ -50,12 +50,12 @@ type BracketExpr struct {
 }
 
 type ColonExpr struct {
-	Start int
-	End   int
+	Start Expr
+	End   Expr
 }
 
 type IndexExpr struct {
-	Index int
+	Index Expr
 }
 
 type BooleanLiteral struct {


Test patch:
diff --git a/internal/topo/operator/misc_func_test.go b/internal/topo/operator/misc_func_test.go
index 4078b830..1b15e602 100644
--- a/internal/topo/operator/misc_func_test.go
+++ b/internal/topo/operator/misc_func_test.go
@@ -765,6 +765,26 @@ func TestJsonPathFunc_Apply1(t *testing.T) {
 			result: []map[string]interface{}{{
 				"a": "Shield of faith",
 			}},
+		}, {
+			sql: `SELECT all[poi[-1] + 1]->ts as powerOnTs FROM test`,
+			data: &xsql.Tuple{
+				Emitter: "test",
+				Message: xsql.Message{
+					"all": []map[string]interface{}{
+						{"SystemPowerMode": 0, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": float64(1), "ts": 0},
+						{"SystemPowerMode": 0, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": float64(4), "ts": 500},
+						{"SystemPowerMode": 2, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": 0, "ts": 1000},
+						{"SystemPowerMode": 2, "VehicleSpeed": 10, "FLWdwPosition": 20, "FrontWiperSwitchStatus": 0, "ts": 60000},
+						{"SystemPowerMode": 2, "VehicleSpeed": 10, "FLWdwPosition": 20, "FrontWiperSwitchStatus": 0, "ts": 89500},
+						{"SystemPowerMode": 2, "VehicleSpeed": 20, "FLWdwPosition": 50, "FrontWiperSwitchStatus": 5, "ts": 90000},
+						{"SystemPowerMode": 2, "VehicleSpeed": 40, "FLWdwPosition": 60, "FrontWiperSwitchStatus": 5, "ts": 121000},
+					},
+					"poi": []interface{}{0, 1},
+				},
+			},
+			result: []map[string]interface{}{{
+				"powerOnTs": float64(1000),
+			}},
 		},
 	}
 
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 0e8bbf3d..eeeaee57 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -1697,7 +1697,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.IndexExpr{Index: 0},
+							RHS: &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: 0}},
 						},
 						Name:  "",
 						AName: ""},
@@ -1715,7 +1715,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 							LHS: &ast.BinaryExpr{
 								LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 								OP:  ast.SUBSET,
-								RHS: &ast.IndexExpr{Index: 0},
+								RHS: &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: 0}},
 							},
 							OP:  ast.ARROW,
 							RHS: &ast.FieldRef{Name: "first"},
@@ -1740,7 +1740,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 								RHS: &ast.FieldRef{Name: "first"},
 							},
 							OP:  ast.SUBSET,
-							RHS: &ast.IndexExpr{Index: 2},
+							RHS: &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: 2}},
 						},
 
 						Name:  "",
@@ -1763,7 +1763,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 									RHS: &ast.FieldRef{Name: "first"},
 								},
 								OP:  ast.SUBSET,
-								RHS: &ast.IndexExpr{Index: 2},
+								RHS: &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: 2}},
 							},
 							OP:  ast.ARROW,
 							RHS: &ast.FieldRef{Name: "test"},
@@ -1784,7 +1784,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.ColonExpr{Start: 0, End: 1},
+							RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: 1}},
 						},
 						Name:  "",
 						AName: ""},
@@ -1801,7 +1801,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.ColonExpr{Start: 0, End: 1},
+							RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: 1}},
 						},
 						Name:  "",
 						AName: ""},
@@ -1818,7 +1818,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.ColonExpr{Start: 0, End: math.MinInt32},
+							RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: math.MinInt32}},
 						},
 						Name:  "",
 						AName: ""},
@@ -1835,7 +1835,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.ColonExpr{Start: 2, End: math.MinInt32},
+							RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 2}, End: &ast.IntegerLiteral{Val: math.MinInt32}},
 						},
 						Name:  "",
 						AName: "c"},
@@ -1850,7 +1850,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 				Fields: []ast.Field{
 					{
 						Expr: &ast.BinaryExpr{
-							LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: 2, End: math.MinInt32}},
+							LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 2}, End: &ast.IntegerLiteral{Val: math.MinInt32}}},
 							OP:  ast.ARROW,
 							RHS: &ast.FieldRef{Name: "first"},
 						},
@@ -1880,7 +1880,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 				Fields: []ast.Field{
 					{
 						Expr: &ast.BinaryExpr{
-							LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{StreamName: ast.StreamName("demo"), Name: "children"}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: 2, End: math.MinInt32}},
+							LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{StreamName: ast.StreamName("demo"), Name: "children"}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 2}, End: &ast.IntegerLiteral{Val: math.MinInt32}}},
 							OP:  ast.ARROW,
 							RHS: &ast.FieldRef{Name: "first"},
 						},
@@ -1900,7 +1900,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 							Name: "lower",
 							Args: []ast.Expr{
 								&ast.BinaryExpr{
-									LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{StreamName: ast.StreamName("demo"), Name: "children"}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: 2, End: math.MinInt32}},
+									LHS: &ast.BinaryExpr{LHS: &ast.FieldRef{StreamName: ast.StreamName("demo"), Name: "children"}, OP: ast.SUBSET, RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 2}, End: &ast.IntegerLiteral{Val: math.MinInt32}}},
 									OP:  ast.ARROW,
 									RHS: &ast.FieldRef{Name: "first"},
 								},
@@ -1921,7 +1921,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 						Expr: &ast.BinaryExpr{
 							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
 							OP:  ast.SUBSET,
-							RHS: &ast.ColonExpr{Start: 0, End: 1},
+							RHS: &ast.ColonExpr{Start: &ast.IntegerLiteral{Val: 0}, End: &ast.IntegerLiteral{Val: 1}},
 						},
 						Name:  "",
 						AName: ""},
@@ -1931,7 +1931,7 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 					LHS: &ast.BinaryExpr{
 						LHS: &ast.FieldRef{Name: "abc", StreamName: ast.DefaultStream},
 						OP:  ast.SUBSET,
-						RHS: &ast.IndexExpr{Index: 0},
+						RHS: &ast.IndexExpr{Index: &ast.IntegerLiteral{Val: 0}},
 					},
 					OP:  ast.GT,
 					RHS: &ast.IntegerLiteral{Val: 12},
@@ -1944,6 +1944,22 @@ func TestParser_ParseJsonExpr(t *testing.T) {
 			stmt: nil,
 			err:  "Too many field names. Please use -> to reference keys in struct.\n",
 		},
+		{
+			s: `SELECT children[index] FROM demo`,
+			stmt: &ast.SelectStatement{
+				Fields: []ast.Field{
+					{
+						Expr: &ast.BinaryExpr{
+							LHS: &ast.FieldRef{Name: "children", StreamName: ast.DefaultStream},
+							OP:  ast.SUBSET,
+							RHS: &ast.IndexExpr{Index: &ast.FieldRef{Name: "index", StreamName: ast.DefaultStream}},
+						},
+						Name:  "",
+						AName: ""},
+				},
+				Sources: []ast.Source{&ast.Table{Name: "demo"}},
+			},
+		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))