feat: querier topk supports counts

server/controller/common/utils.go

@@ -18,6 +18,8 @@ package common
 
 import (
 "reflect"
+"strconv"
+"strings"
 "sync"
 )
 
@@ -191,3 +193,23 @@ func getVarSizeInternal(val any, visited map[uintptr]bool) uint64 {
 return uint64(tp.Size())
 }
 }
+
+func CompareVersion(ver1, ver2 string) int {
+if ver1 == "" || ver2 == "" {
+log.Warningf("empty version string: ver1=%s, ver2=%s", ver1, ver2)
+return 0
+}
+
+parts1 := strings.Split(ver1, ".")
+parts2 := strings.Split(ver2, ".")
+
+v1, err1 := strconv.Atoi(parts1[0])
+v2, err2 := strconv.Atoi(parts2[0])
+
+if err1 != nil || err2 != nil {
+log.Errorf("invalid version format: ver1=%s, ver2=%s", ver1, ver2)
+return 0
+}
+
+return v1 - v2
+}

server/controller/http/router/ck_version.go

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2024 Yunshan Networks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+"github.com/gin-gonic/gin"
+
+"github.com/deepflowio/deepflow/server/controller/http/common/response"
+"github.com/deepflowio/deepflow/server/controller/http/service"
+)
+
+type CKVersion struct {
+}
+
+func NewCKVersion() *CKVersion {
+return &CKVersion{}
+}
+
+func (c *CKVersion) RegisterTo(e *gin.Engine) {
+e.GET("/v1/ck_version/", getCKVersion())
+}
+
+func getCKVersion() gin.HandlerFunc {
+return gin.HandlerFunc(func(c *gin.Context) {
+data, err := service.GetCKVersion()
+response.JSON(c, response.SetData(data), response.SetError(err))
+})
+}

server/controller/http/server.go

@@ -127,6 +127,8 @@ func (s *Server) appendRegistrant() []registrant.Registrant {
 
 // icon
 router.NewIcon(s.controllerConfig),
+// ck version
+router.NewCKVersion(),
 
 // resource
 resource.NewDomain(s.controllerConfig),

server/controller/http/service/ck_version.go

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2024 Yunshan Networks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package service
+
+import (
+"fmt"
+"net/http"
+"net/url"
+
+"github.com/deepflowio/deepflow/server/controller/common"
+"github.com/deepflowio/deepflow/server/querier/config"
+)
+
+func GetCKVersion() (string, error) {
+ckVersion := ""
+
+body := url.Values{
+"sql": {"SELECT version()"},
+}
+url := fmt.Sprintf("http://localhost:%d/v1/query?simple_sql=true", config.Cfg.ListenPort)
+resp, err := common.CURLForm(
+http.MethodPost,
+url,
+body,
+)
+if err != nil {
+log.Error(err)
+return ckVersion, err
+}
+values := resp.Get("result").Get("values").MustArray()
+if len(values) == 0 || len(values[0].([]interface{})) == 0 {
+err = fmt.Errorf("get clickhouse version failed: empty result")
+log.Error(err)
+return ckVersion, err
+}
+if row, ok := values[0].([]interface{}); ok && len(row) > 0 {
+if version, ok := row[0].(string); ok {
+ckVersion = version
+} else {
+return ckVersion, fmt.Errorf("version is not string type")
+}
+} else {
+return ckVersion, fmt.Errorf("invalid response format")
+}
+return ckVersion, err
+}

server/controller/tagrecorder/dictionary.go

@@ -386,7 +386,7 @@ func (c *Dictionary) update(clickHouseCfg *clickhouse.ClickHouseConfig) {
 log.Error(err, logger.NewORGPrefix(orgID))
 return
 }
-if versions[0] > common.CLICK_HOUSE_VERSION {
+if common.CompareVersion(versions[0], common.CLICK_HOUSE_VERSION) >= 0 {
 continue
 }
 // Get the current view in the database

server/querier/config/config.go

@@ -90,6 +90,7 @@ type Clickhouse struct {
 MaxConnection int `default:"20" yaml:"max-connection"`
 UseQueryCache bool `default:"true" yaml:"use-query-cache"`
 QueryCacheTTL string `default:"600" yaml:"query-cache-ttl"`
+Version string `default:"" yaml:"-"`
 }
 
 type AutoCustomTags struct {

server/querier/engine/clickhouse/clickhouse.go

@@ -83,6 +83,13 @@ var showPatterns = []string{
 }
 var res []*regexp.Regexp
 
+const (
+TUPLE_ELEMENT_VALUES_INDEX = 1
+TUPLE_ELEMENT_COUNTS_INDEX = 2
+TOPK_PREFIX_ARRAY = "array_"
+TOPK_PREFIX_COUNTS = "counts_"
+)
+
 type TargetLabelFilter struct {
 OriginFilter string
 TransFilter string
@@ -114,6 +121,31 @@ func init() {
 }
 }
 
+// createTopKColumn creates a column definition for TopK results
+// functionAs: the function alias (e.g., "array_TopK_10(ip_0)")
+// prefix: column prefix ("" for values, "counts_" for counts)
+// elementIndex: tuple element index (1 for values, 2 for counts)
+// argsLength: number of TopK function arguments
+func createTopKColumn(functionAs, prefix string, elementIndex, argsLength int) (string, string, error) {
+if strings.TrimSpace(functionAs) == "" {
+return "", "", fmt.Errorf("TopK function alias cannot be empty")
+}
+if elementIndex < 1 || elementIndex > 2 {
+return "", "", fmt.Errorf("invalid tuple element index: %d, must be 1 or 2", elementIndex)
+}
+columnValue := "`" + strings.Trim(functionAs, "`") + "`"
+if ctlcommon.CompareVersion(config.Cfg.Clickhouse.Version, ctlcommon.CLICK_HOUSE_VERSION) >= 0 {
+columnValue = fmt.Sprintf("tupleElement(`%s`,%d)", strings.Trim(functionAs, "`"), elementIndex)
+}
+
+// if topk has one arg, need to concat array to string
+if argsLength == 1 {
+columnValue = fmt.Sprintf("arrayStringConcat(%s,',')", columnValue)
+}
+columnAlias := strings.Replace(functionAs, TOPK_PREFIX_ARRAY, prefix, 1)
+return columnValue, columnAlias, nil
+}
+
 func (e *CHEngine) ExecuteQuery(args *common.QuerierParams) (*common.Result, map[string]interface{}, error) {
 // 解析show开头的sql
 // show metrics/tags from <table_name> 例：show metrics/tags from l4_flow_log
@@ -1597,6 +1629,37 @@ func (e *CHEngine) parseSelectAlias(item *sqlparser.AliasedExpr) error {
 functionAs = strings.ReplaceAll(chCommon.ParseAlias(item.Expr), "`", "")
 }
 }
+
+// topk add counts column
+if name == view.FUNCTION_TOPK {
+argsLength := len(args)
+if strings.HasPrefix(functionAs, "`") {
+functionAs = strings.TrimPrefix(functionAs, "`")
+functionAs = "`" + TOPK_PREFIX_ARRAY + functionAs
+} else {
+functionAs = TOPK_PREFIX_ARRAY + functionAs
+}
+e.ColumnSchemas[len(e.ColumnSchemas)-1].Name = strings.Trim(functionAs, "`")
+// create topk string and counts column
+topKStr, topKStrAs, err := createTopKColumn(functionAs, "", TUPLE_ELEMENT_VALUES_INDEX, argsLength-1)
+if err != nil {
+return err
+}
+topKCounts, topKCountsAs, err := createTopKColumn(functionAs, TOPK_PREFIX_COUNTS, TUPLE_ELEMENT_COUNTS_INDEX, argsLength-1)
+if err != nil {
+return err
+}
+// make sure topk string and counts is the first two select item
+topkStrSchema := common.NewColumnSchema(topKStrAs, topKStr, "")
+topkStrSchema.Type = common.COLUMN_SCHEMA_TYPE_METRICS
+topkCountsSchema := common.NewColumnSchema(topKCountsAs, topKCounts, "")
+topkCountsSchema.Type = common.COLUMN_SCHEMA_TYPE_METRICS
+e.Statements = append([]Statement{&SelectTag{Value: topKCounts, Alias: topKCountsAs, Flag: view.NODE_FLAG_METRICS_OUTER}}, e.Statements...)
+e.ColumnSchemas = append([]*common.ColumnSchema{topkCountsSchema}, e.ColumnSchemas...)
+e.Statements = append([]Statement{&SelectTag{Value: topKStr, Alias: topKStrAs, Flag: view.NODE_FLAG_METRICS_OUTER}}, e.Statements...)
+e.ColumnSchemas = append([]*common.ColumnSchema{topkStrSchema}, e.ColumnSchemas...)
+}
+
 function, levelFlag, unit, err := GetAggFunc(name, args, functionAs, derivativeArgs, e)
 if err != nil {
 return err

server/querier/engine/clickhouse/clickhouse_test.go

@@ -350,7 +350,7 @@ var (
 name: "topk_enum",
 db: "flow_log",
 input: "select TopK(protocol,2) from l4_flow_log limit 2",
-output: []string{"SELECT arrayStringConcat(topKIf(2)(dictGetOrDefault('flow_tag.int_enum_map', 'name_en', ('protocol',toUInt64(protocol)), protocol), dictGetOrDefault('flow_tag.int_enum_map', 'name_en', ('protocol',toUInt64(protocol)), protocol) != ''), ',') AS `TopK_2(protocol)` FROM flow_log.`l4_flow_log` LIMIT 2"},
+output: []string{"SELECT arrayStringConcat(tupleElement(`array_TopK_2(protocol)`,1),',') AS `TopK_2(protocol)`, arrayStringConcat(tupleElement(`array_TopK_2(protocol)`,2),',') AS `counts_TopK_2(protocol)`, topKIf(2, 3, 'counts')(dictGetOrDefault('flow_tag.int_enum_map', 'name_en', ('protocol',toUInt64(protocol)), protocol), dictGetOrDefault('flow_tag.int_enum_map', 'name_en', ('protocol',toUInt64(protocol)), protocol) != '') AS `array_TopK_2(protocol)` FROM flow_log.`l4_flow_log` LIMIT 2"},
 }, {
 name: "select_enum",
 db: "flow_log",
@@ -372,19 +372,19 @@ var (
 }, {
 name: "TopK_1",
 input: "select TopK(ip_0, 10) from l4_flow_log limit 1",
-output: []string{"SELECT arrayStringConcat(topKIf(10)(if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)), if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)) != ''), ',') AS `TopK_10(ip_0)` FROM flow_log.`l4_flow_log` LIMIT 1"},
+output: []string{"SELECT arrayStringConcat(tupleElement(`array_TopK_10(ip_0)`,1),',') AS `TopK_10(ip_0)`, arrayStringConcat(tupleElement(`array_TopK_10(ip_0)`,2),',') AS `counts_TopK_10(ip_0)`, topKIf(10, 3, 'counts')(if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)), if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)) != '') AS `array_TopK_10(ip_0)` FROM flow_log.`l4_flow_log` LIMIT 1"},
 }, {
 name: "TopK_2",
 input: "select TopK(ip_0, pod_0, 10) from l4_flow_log limit 1",
-output: []string{"SELECT topKIf(10)((if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)), dictGet('flow_tag.pod_map', 'name', (toUInt64(pod_id_0)))), (if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)) != '' AND dictGet('flow_tag.pod_map', 'name', (toUInt64(pod_id_0))) != '')) AS `TopK_10(ip_0, pod_0)` FROM flow_log.`l4_flow_log` LIMIT 1"},
+output: []string{"SELECT tupleElement(`array_TopK_10(ip_0, pod_0)`,1) AS `TopK_10(ip_0, pod_0)`, tupleElement(`array_TopK_10(ip_0, pod_0)`,2) AS `counts_TopK_10(ip_0, pod_0)`, topKIf(10, 3, 'counts')((if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)), dictGet('flow_tag.pod_map', 'name', (toUInt64(pod_id_0)))), (if(is_ipv4=1, IPv4NumToString(ip4_0), IPv6NumToString(ip6_0)) != '' AND dictGet('flow_tag.pod_map', 'name', (toUInt64(pod_id_0))) != '')) AS `array_TopK_10(ip_0, pod_0)` FROM flow_log.`l4_flow_log` LIMIT 1"},
 }, {
 name: "TopK_err",
 input: "select TopK(ip_0, 111) from l4_flow_log limit 1",
 wantErr: "function [TopK] argument [111] value range is incorrect, it should be within [1, 100]",
 }, {
 name: "TopK_3",
 input: "SELECT TopK(`region`,3) AS `TopK_3(区域)` FROM `vtap_app_port` WHERE (time>=1705370520 AND time<=1705371300)",
-output: []string{"SELECT arrayStringConcat(topKArray(3)(`_grouparray_dictGet(flow_tag.region_map, name, (toUInt64(region_id)))_dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != ''`), ',') AS `TopK_3(区域)` FROM (SELECT groupArrayIf(dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))), dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != '') AS `_grouparray_dictGet(flow_tag.region_map, name, (toUInt64(region_id)))_dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != ''` FROM flow_metrics.`application.1m` WHERE (`time` >= 1705370520 AND `time` <= 1705371300)) LIMIT 10000"},
+output: []string{"SELECT arrayStringConcat(tupleElement(`array_TopK_3(区域)`,1),',') AS `TopK_3(区域)`, arrayStringConcat(tupleElement(`array_TopK_3(区域)`,2),',') AS `counts_TopK_3(区域)`, topKArray(3, 3, 'counts')(`_grouparray_dictGet(flow_tag.region_map, name, (toUInt64(region_id)))_dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != ''`) AS `array_TopK_3(区域)` FROM (SELECT groupArrayIf(dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))), dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != '') AS `_grouparray_dictGet(flow_tag.region_map, name, (toUInt64(region_id)))_dictGet('flow_tag.region_map', 'name', (toUInt64(region_id))) != ''` FROM flow_metrics.`application.1m` WHERE (`time` >= 1705370520 AND `time` <= 1705371300)) LIMIT 10000"},
 db: "flow_metrics",
 datasource: "1m",
 }, {
@@ -700,6 +700,7 @@ func TestGetSql(t *testing.T) {
 func Load() error {
 ServerCfg := config.DefaultConfig()
 config.Cfg = &ServerCfg.QuerierConfig
+config.Cfg.Clickhouse.Version = "24.8"
 config.ControllerCfg = &ServerCfg.ControllerConfig
 dir := "../../db_descriptions"
 dbDescriptions, err := common.LoadDbDescriptions(dir)

server/querier/engine/clickhouse/client/client.go

@@ -63,9 +63,10 @@ type Client struct {
 DB string
 Context context.Context
 Debug *Debug
+Version string
 }
 
-func (c *Client) init(query_uuid string) error {
+func (c *Client) Init(query_uuid string) error {
 if query_uuid == "" {
 query_uuid = uuid.NewString()
 }
@@ -104,6 +105,7 @@ func (c *Client) init(query_uuid string) error {
 c.connection = connection
 if version == "" {
 version, _ = c.GetVersion()
+c.Version = version
 }
 return nil
 }
@@ -141,7 +143,7 @@ func (c *Client) DoQuery(params *QueryParams) (result *common.Result, err error)
 sqlstr = strings.ReplaceAll(sqlstr, "target_label_live_view", "target_label_map")
 }
 
-err = c.init(query_uuid)
+err = c.Init(query_uuid)
 if err != nil {
 return nil, err
 }

server/querier/engine/clickhouse/view/function.go

@@ -22,7 +22,9 @@ import (
 "strconv"
 "strings"
 
+ctlcommon "github.com/deepflowio/deepflow/server/controller/common"
 "github.com/deepflowio/deepflow/server/querier/common"
+"github.com/deepflowio/deepflow/server/querier/config"
 )
 
 const (
@@ -57,6 +59,11 @@ const (
 FUNCTION_COUNTDISTINCT = "countDistinct"
 )
 
+const (
+TOPK_COUNTS_DEFAULT_LIMIT = "3"
+TOPK_COUNTS_MODE_FLAG = "'counts'"
+)
+
 // 对外提供的算子与数据库实际算子转换
 var FUNC_NAME_MAP map[string]string = map[string]string{
 FUNCTION_SUM: "SUM",
@@ -237,10 +244,6 @@ func (f *DefaultFunction) WriteTo(buf *bytes.Buffer) {
 return
 }
 
-isSingleTagTok := f.Name == FUNCTION_TOPK && len(f.Args) == 1
-if isSingleTagTok {
-buf.WriteString("arrayStringConcat(")
-}
 buf.WriteString(dbFuncName)
 
 if f.IsGroupArray {
@@ -255,6 +258,10 @@ func (f *DefaultFunction) WriteTo(buf *bytes.Buffer) {
 args := f.Args
 if f.Name == FUNCTION_TOPK {
 args = f.Args[len(f.Args)-1:]
+// topk add counts mode
+if ctlcommon.CompareVersion(config.Cfg.Clickhouse.Version, ctlcommon.CLICK_HOUSE_VERSION) >= 0 {
+args = append(args, []string{TOPK_COUNTS_DEFAULT_LIMIT, TOPK_COUNTS_MODE_FLAG}...)
+}
 } else if f.Name == FUNCTION_ANY || f.Name == FUNCTION_UNIQ || f.Name == FUNCTION_UNIQ_EXACT {
 args = nil
 }
@@ -311,9 +318,6 @@ func (f *DefaultFunction) WriteTo(buf *bytes.Buffer) {
 }
 
 buf.WriteString(")")
-if isSingleTagTok {
-buf.WriteString(", ',')")
-}
 buf.WriteString(f.Math)
 if !f.Nest && f.Alias != "" {
 buf.WriteString(" AS ")