feat: fixProducerAPI

pkg/query-service/app/http_handler.go

@@ -3222,16 +3222,16 @@ func (aH *APIHandler) getProducerThroughputOverview(
 	}
 
 	for _, res := range result {
-		for _, series := range res.Series {
+		for _, list := range res.List {
-			serviceName, serviceNameOk := series.Labels["service_name"]
+			serviceName, serviceNameOk := list.Data["service_name"].(*string)
-			topicName, topicNameOk := series.Labels["topic"]
+			topicName, topicNameOk := list.Data["topic"].(*string)
-			params := []string{serviceName, topicName}
+			params := []string{*serviceName, *topicName}
 			hashKey := uniqueIdentifier(params, "#")
 			_, ok := attributeCache.Hash[hashKey]
 			if topicNameOk && serviceNameOk && !ok {
 				attributeCache.Hash[hashKey] = struct{}{}
-				attributeCache.TopicName = append(attributeCache.TopicName, topicName)
+				attributeCache.TopicName = append(attributeCache.TopicName, *topicName)
-				attributeCache.ServiceName = append(attributeCache.ServiceName, serviceName)
+				attributeCache.ServiceName = append(attributeCache.ServiceName, *serviceName)
 			}
 		}
 	}
@@ -3256,25 +3256,23 @@ func (aH *APIHandler) getProducerThroughputOverview(
 	}
 
 	latencyColumn := &v3.Result{QueryName: "latency"}
-	var latencySeries []*v3.Series
+	var latencySeries []*v3.Row
 	for _, res := range resultFetchLatency {
-		for _, series := range res.Series {
+		for _, list := range res.List {
-			topic, topicOk := series.Labels["topic"]
+			topic, topicOk := list.Data["topic"].(*string)
-			serviceName, serviceNameOk := series.Labels["service_name"]
+			serviceName, serviceNameOk := list.Data["service_name"].(*string)
-			params := []string{topic, serviceName}
+			params := []string{*serviceName, *topic}
 			hashKey := uniqueIdentifier(params, "#")
 			_, ok := attributeCache.Hash[hashKey]
 			if topicOk && serviceNameOk && ok {
-				latencySeries = append(latencySeries, series)
+				latencySeries = append(latencySeries, list)
 			}
 		}
 	}
 
-	latencyColumn.Series = latencySeries
+	latencyColumn.List = latencySeries
 	result = append(result, latencyColumn)
 
-	resultFetchLatency = postprocess.TransformToTableForBuilderQueries(result, queryRangeParams)
-
 	resp := v3.QueryRangeResponse{
 		Result: resultFetchLatency,
 	}

pkg/query-service/app/integrations/messagingQueues/kafka/translator.go

@@ -284,7 +284,7 @@ func BuildQRParamsWithCache(messagingQueue *MessagingQueue, queryContext string,
 		cq = &v3.CompositeQuery{
 			QueryType:      v3.QueryTypeBuilder,
 			BuilderQueries: bhq,
-			PanelType:      v3.PanelTypeTable,
+			PanelType:      v3.PanelTypeList,
 		}
 	}
 
@@ -364,7 +364,7 @@ func BuildClickHouseQuery(messagingQueue *MessagingQueue, queueType string, quer
 
 func buildCompositeQuery(chq *v3.ClickHouseQuery, queryContext string) (*v3.CompositeQuery, error) {
 
-	if queryContext == "producer-consumer-eval" {
+	if queryContext == "producer-consumer-eval" || queryContext == "producer-throughput-overview" {
 		return &v3.CompositeQuery{
 			QueryType:         v3.QueryTypeClickHouseSQL,
 			ClickHouseQueries: map[string]*v3.ClickHouseQuery{queryContext: chq},