Repository: lf-edge/ekuiper
Pull request #1300 (2022-06-15 03:21:36)

fix(docs): modify kmanager.yaml link


Patch:
diff --git a/docs/en_US/tutorials/deploy/openyurt_tutorial.md b/docs/en_US/tutorials/deploy/openyurt_tutorial.md
index 9092cc55..d4d0b8aa 100644
--- a/docs/en_US/tutorials/deploy/openyurt_tutorial.md
+++ b/docs/en_US/tutorials/deploy/openyurt_tutorial.md
@@ -203,7 +203,7 @@ $ curl http://192.168.2.143:9081
 
 ## Deploy the eKuiper dashboard to cloud
 
-We will deploy ekuiper dashboard in the cloud node by kubectl tool with [kmanager.yaml](./kmanager.yaml). In the
+We will deploy ekuiper dashboard in the cloud node by kubectl tool with [kmanager.yaml](https://github.com/lf-edge/ekuiper/blob/master/docs/en_US/tutorials/deploy/kmanager.yaml). In the
 configuration file, we define a deployment and a service for eKuiper manager which is a web based UI. First, we need to
 revise the manager docker tag to the correct version that matches the eKuiper version in line 21:
 
diff --git a/docs/zh_CN/tutorials/deploy/openyurt_tutorial.md b/docs/zh_CN/tutorials/deploy/openyurt_tutorial.md
index f9686720..1d93b393 100644
--- a/docs/zh_CN/tutorials/deploy/openyurt_tutorial.md
+++ b/docs/zh_CN/tutorials/deploy/openyurt_tutorial.md
@@ -177,7 +177,7 @@ $ curl http://192.168.2.143:9081
 
 ## 将 eKuiper 仪表板部署到云端
 
-我们将使用 [kmanager.yaml](./kmanager.yaml) 和 kubectl 工具在云节点中部署 ekuiper 仪表板。 eKuiper manager 是一个基于 web
+我们将使用 [kmanager.yaml](https://github.com/lf-edge/ekuiper/blob/master/docs/zh_CN/tutorials/deploy/kmanager.yaml) 和 kubectl 工具在云节点中部署 ekuiper 仪表板。 eKuiper manager 是一个基于 web
 的用户界面。在配置文件中，我们为eKuiper manager 定义了部署和服务。
 
 首先，我们需要确保文件中使用的仪表盘版本跟 eKuiper 版本相匹配。打开并修改 kmanager.yaml 第21行，确保版本正确。
diff --git a/internal/topo/operator/order_operator.go b/internal/topo/operator/order_operator.go
index 1adcab8b..bfeedd3b 100644
--- a/internal/topo/operator/order_operator.go
+++ b/internal/topo/operator/order_operator.go
@@ -29,10 +29,10 @@ type OrderOp struct {
  *  input: *xsql.Tuple from preprocessor | xsql.WindowTuplesSet from windowOp | xsql.JoinTupleSets from joinOp
  *  output: *xsql.Tuple | xsql.WindowTuplesSet | xsql.JoinTupleSets
  */
-func (p *OrderOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, _ *xsql.AggregateFunctionValuer) interface{} {
+func (p *OrderOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) interface{} {
 	log := ctx.GetLogger()
 	log.Debugf("order plan receive %s", data)
-	sorter := xsql.OrderedBy(p.SortFields, fv)
+	sorter := xsql.OrderedBy(p.SortFields, fv, afv)
 	switch input := data.(type) {
 	case error:
 		return input
diff --git a/internal/xsql/parser.go b/internal/xsql/parser.go
index 6bddd051..7737c517 100644
--- a/internal/xsql/parser.go
+++ b/internal/xsql/parser.go
@@ -374,6 +374,13 @@ func (p *Parser) parseSorts() (ast.SortFields, error) {
 						return nil, err
 					}
 
+					p.unscan()
+					if exp, err := p.ParseExpr(); err != nil {
+						return nil, err
+					} else {
+						s.FieldExpr = exp
+					}
+
 					if t2, _ := p.scanIgnoreWhitespace(); t2 == ast.DESC {
 						s.Ascending = false
 						ss = append(ss, s)
diff --git a/internal/xsql/valuer.go b/internal/xsql/valuer.go
index ad4ac8fc..3e9f3462 100644
--- a/internal/xsql/valuer.go
+++ b/internal/xsql/valuer.go
@@ -102,17 +102,19 @@ type SortingData interface {
 // MultiSorter multiSorter implements the Sort interface, sorting the changes within.Hi
 type MultiSorter struct {
 	SortingData
-	fields ast.SortFields
-	valuer CallValuer
-	values []map[string]interface{}
+	fields    ast.SortFields
+	valuer    *FunctionValuer
+	aggValuer *AggregateFunctionValuer
+	values    []map[string]interface{}
 }
 
 // OrderedBy returns a Sorter that sorts using the less functions, in order.
 // Call its Sort method to sort the data.
-func OrderedBy(fields ast.SortFields, fv *FunctionValuer) *MultiSorter {
+func OrderedBy(fields ast.SortFields, fv *FunctionValuer, afv *AggregateFunctionValuer) *MultiSorter {
 	return &MultiSorter{
-		fields: fields,
-		valuer: fv,
+		fields:    fields,
+		valuer:    fv,
+		aggValuer: afv,
 	}
 }
 
@@ -161,15 +163,53 @@ func (ms *MultiSorter) Sort(data SortingData) error {
 	types := make([]string, len(ms.fields))
 	ms.values = make([]map[string]interface{}, data.Len())
 	//load and validate data
-	for i := 0; i < data.Len(); i++ {
-		ms.values[i] = make(map[string]interface{})
-		p := data.Index(i)
-		vep := &ValuerEval{Valuer: MultiValuer(p, ms.valuer)}
-		for j, field := range ms.fields {
-			vp, _ := vep.Valuer.Value(field.Name, string(field.StreamName))
-			if err, ok := vp.(error); ok {
-				return err
-			} else {
+	switch input := data.(type) {
+	case error:
+		return input
+	case GroupedTuplesSet:
+		for i, v := range input {
+			ms.values[i] = make(map[string]interface{})
+			ms.aggValuer.SetData(v)
+			vep := &ValuerEval{Valuer: MultiAggregateValuer(v, ms.valuer, v.Content[0], ms.valuer, ms.aggValuer, &WildcardValuer{Data: v.Content[0]})}
+			for j, field := range ms.fields {
+				vp := vep.Eval(field.FieldExpr)
+				if types[j] == "" && vp != nil {
+					types[j] = fmt.Sprintf("%T", vp)
+				}
+				if err := validate(types[j], vp); err != nil {
+					return err
+				} else {
+					ms.values[i][field.Uname] = vp
+				}
+			}
+		}
+	case WindowTuplesSet:
+		if len(input.Content) != 1 {
+			return fmt.Errorf("run Order error: input WindowTuplesSet with multiple tuples cannot be evaluated")
+		}
+		wdtps := input.Content[0].Tuples
+		for i, v := range wdtps {
+			ms.values[i] = make(map[string]interface{})
+			vep := &ValuerEval{Valuer: MultiValuer(ms.valuer, &v, ms.valuer, &WildcardValuer{Data: &v})}
+			for j, field := range ms.fields {
+				vp := vep.Eval(field.FieldExpr)
+				if types[j] == "" && vp != nil {
+					types[j] = fmt.Sprintf("%T", vp)
+				}
+				if err := validate(types[j], vp); err != nil {
+					return err
+				} else {
+					ms.values[i][field.Uname] = vp
+				}
+			}
+		}
+	case *JoinTupleSets:
+		joinTps := input.Content
+		for i, v := range joinTps {
+			ms.values[i] = make(map[string]interface{})
+			vep := &ValuerEval{Valuer: MultiValuer(ms.valuer, &v, ms.valuer, &WildcardValuer{Data: &v})}
+			for j, field := range ms.fields {
+				vp := vep.Eval(field.FieldExpr)
 				if types[j] == "" && vp != nil {
 					types[j] = fmt.Sprintf("%T", vp)
 				}
@@ -178,6 +218,7 @@ func (ms *MultiSorter) Sort(data SortingData) error {
 				} else {
 					ms.values[i][field.Uname] = vp
 				}
+
 			}
 		}
 	}
diff --git a/pkg/ast/statement.go b/pkg/ast/statement.go
index 4b568f21..3d4c0ed1 100644
--- a/pkg/ast/statement.go
+++ b/pkg/ast/statement.go
@@ -181,6 +181,7 @@ type SortField struct {
 	StreamName StreamName
 	Uname      string // unique name of a field
 	Ascending  bool
+	FieldExpr  Expr
 
 	Expr
 }
diff --git a/pkg/ast/visitor.go b/pkg/ast/visitor.go
index d27e47d8..245e6b87 100644
--- a/pkg/ast/visitor.go
+++ b/pkg/ast/visitor.go
@@ -80,7 +80,7 @@ func Walk(v Visitor, node Node) {
 
 	case SortFields:
 		for _, sf := range n {
-			Walk(v, &sf)
+			Walk(v, sf.FieldExpr)
 		}
 
 	//case *SortField:


Test patch:
diff --git a/internal/topo/topotest/rule_test.go b/internal/topo/topotest/rule_test.go
index 7278dcf7..72cec6b9 100644
--- a/internal/topo/topotest/rule_test.go
+++ b/internal/topo/topotest/rule_test.go
@@ -538,6 +538,34 @@ func TestSingleSQL(t *testing.T) {
 				"sink_mockSink_0_records_in_total":  int64(5),
 				"sink_mockSink_0_records_out_total": int64(5),
 
+				"source_demo_0_exceptions_total":  int64(0),
+				"source_demo_0_records_in_total":  int64(5),
+				"source_demo_0_records_out_total": int64(5),
+			},
+		}, {
+			Name: `TestAliasOrderBy14`,
+			Sql:  "SELECT color, count(*) as c FROM demo where color != \"red\" GROUP BY COUNTWINDOW(5), color Order by c DESC",
+			R: [][]map[string]interface{}{
+				{{
+					"color": "blue",
+					"c":     float64(2),
+				},
+					{
+						"color": "yellow",
+						"c":     float64(1),
+					},
+				},
+			},
+			M: map[string]interface{}{
+				"op_6_project_0_exceptions_total":   int64(0),
+				"op_6_project_0_process_latency_us": int64(0),
+				"op_6_project_0_records_in_total":   int64(1),
+				"op_6_project_0_records_out_total":  int64(1),
+
+				"sink_mockSink_0_exceptions_total":  int64(0),
+				"sink_mockSink_0_records_in_total":  int64(1),
+				"sink_mockSink_0_records_out_total": int64(1),
+
 				"source_demo_0_exceptions_total":  int64(0),
 				"source_demo_0_records_in_total":  int64(5),
 				"source_demo_0_records_out_total": int64(5),
diff --git a/internal/xsql/parser_test.go b/internal/xsql/parser_test.go
index 5f5204f2..e4bcf8ca 100644
--- a/internal/xsql/parser_test.go
+++ b/internal/xsql/parser_test.go
@@ -999,7 +999,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						},
 					},
 				},
-				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: true}},
+				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: true, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
 		},
 
@@ -1020,7 +1020,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						},
 					},
 				},
-				SortFields: []ast.SortField{{Uname: "s1\007name", Name: "name", StreamName: ast.StreamName("s1"), Ascending: true}},
+				SortFields: []ast.SortField{{Uname: "s1\007name", Name: "name", StreamName: ast.StreamName("s1"), Ascending: true, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
 		},
 
@@ -1041,7 +1041,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						},
 					},
 				},
-				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false}},
+				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
 		},
 
@@ -1055,7 +1055,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: ""},
 				},
 				Sources:    []ast.Source{&ast.Table{Name: "topic/sensor1"}},
-				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false}},
+				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}},
 			},
 		},
 
@@ -1069,7 +1069,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						AName: ""},
 				},
 				Sources:    []ast.Source{&ast.Table{Name: "topic/sensor1"}},
-				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false}, {Uname: "name2", Name: "name2", Ascending: true}},
+				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}, {Uname: "name2", Name: "name2", Ascending: true, FieldExpr: &ast.FieldRef{Name: "name2", StreamName: ast.DefaultStream}}},
 			},
 		},
 
@@ -1093,7 +1093,7 @@ func TestParser_ParseStatement(t *testing.T) {
 						},
 					},
 				},
-				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false}, {Uname: "name2", Name: "name2", Ascending: true}},
+				SortFields: []ast.SortField{{Uname: "name", Name: "name", Ascending: false, FieldExpr: &ast.FieldRef{Name: "name", StreamName: ast.DefaultStream}}, {Uname: "name2", Name: "name2", Ascending: true, FieldExpr: &ast.FieldRef{Name: "name2", StreamName: ast.DefaultStream}}},
 			},
 		},