fix: update api's to repect the new changes in attributes table (#6526)

pkg/query-service/app/clickhouseReader/reader.go

@@ -4117,7 +4117,7 @@ func (r *ClickHouseReader) CheckClickHouse(ctx context.Context) error {
 	return nil
 }
 
-func (r *ClickHouseReader) GetTraceAggregateAttributesV2(ctx context.Context, req *v3.AggregateAttributeRequest) (*v3.AggregateAttributeResponse, error) {
+func (r *ClickHouseReader) GetTraceAggregateAttributes(ctx context.Context, req *v3.AggregateAttributeRequest) (*v3.AggregateAttributeResponse, error) {
 	var query string
 	var err error
 	var rows driver.Rows
@@ -4196,8 +4196,13 @@ func (r *ClickHouseReader) GetTraceAggregateAttributesV2(ctx context.Context, re
 		}
 	}
 
+	fields := constants.NewStaticFieldsTraces
+	if !r.useTraceNewSchema {
+		fields = constants.DeprecatedStaticFieldsTraces
+	}
+
 	// add the new static fields
-	for _, field := range constants.NewStaticFieldsTraces {
+	for _, field := range fields {
 		if (!stringAllowed && field.DataType == v3.AttributeKeyDataTypeString) || (v3.AttributeKey{} == field) {
 			continue
 		} else if len(req.SearchText) == 0 || strings.Contains(field.Key, req.SearchText) {
@@ -4208,79 +4213,7 @@ func (r *ClickHouseReader) GetTraceAggregateAttributesV2(ctx context.Context, re
 	return &response, nil
 }
 
-func (r *ClickHouseReader) GetTraceAggregateAttributes(ctx context.Context, req *v3.AggregateAttributeRequest) (*v3.AggregateAttributeResponse, error) {
+func (r *ClickHouseReader) GetTraceAttributeKeys(ctx context.Context, req *v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
-	if r.useTraceNewSchema {
-		return r.GetTraceAggregateAttributesV2(ctx, req)
-	}
-
-	var query string
-	var err error
-	var rows driver.Rows
-	var response v3.AggregateAttributeResponse
-	where := ""
-	switch req.Operator {
-	case
-		v3.AggregateOperatorCountDistinct,
-		v3.AggregateOperatorCount:
-		where = "tagKey ILIKE $1"
-	case
-		v3.AggregateOperatorRateSum,
-		v3.AggregateOperatorRateMax,
-		v3.AggregateOperatorRateAvg,
-		v3.AggregateOperatorRate,
-		v3.AggregateOperatorRateMin,
-		v3.AggregateOperatorP05,
-		v3.AggregateOperatorP10,
-		v3.AggregateOperatorP20,
-		v3.AggregateOperatorP25,
-		v3.AggregateOperatorP50,
-		v3.AggregateOperatorP75,
-		v3.AggregateOperatorP90,
-		v3.AggregateOperatorP95,
-		v3.AggregateOperatorP99,
-		v3.AggregateOperatorAvg,
-		v3.AggregateOperatorSum,
-		v3.AggregateOperatorMin,
-		v3.AggregateOperatorMax:
-		where = "tagKey ILIKE $1 AND dataType='float64'"
-	case
-		v3.AggregateOperatorNoOp:
-		return &v3.AggregateAttributeResponse{}, nil
-	default:
-		return nil, fmt.Errorf("unsupported aggregate operator")
-	}
-	query = fmt.Sprintf("SELECT DISTINCT(tagKey), tagType, dataType, isColumn FROM %s.%s WHERE %s", r.TraceDB, r.spanAttributeTable, where)
-	if req.Limit != 0 {
-		query = query + fmt.Sprintf(" LIMIT %d;", req.Limit)
-	}
-	rows, err = r.db.Query(ctx, query, fmt.Sprintf("%%%s%%", req.SearchText))
-
-	if err != nil {
-		zap.L().Error("Error while executing query", zap.Error(err))
-		return nil, fmt.Errorf("error while executing query: %s", err.Error())
-	}
-	defer rows.Close()
-
-	var tagKey string
-	var dataType string
-	var tagType string
-	var isColumn bool
-	for rows.Next() {
-		if err := rows.Scan(&tagKey, &tagType, &dataType, &isColumn); err != nil {
-			return nil, fmt.Errorf("error while scanning rows: %s", err.Error())
-		}
-		key := v3.AttributeKey{
-			Key:      tagKey,
-			DataType: v3.AttributeKeyDataType(dataType),
-			Type:     v3.AttributeKeyType(tagType),
-			IsColumn: isColumn,
-		}
-		response.AttributeKeys = append(response.AttributeKeys, key)
-	}
-	return &response, nil
-}
-
-func (r *ClickHouseReader) GetTraceAttributeKeysV2(ctx context.Context, req *v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
 
 	var query string
 	var err error
@@ -4326,8 +4259,14 @@ func (r *ClickHouseReader) GetTraceAttributeKeysV2(ctx context.Context, req *v3.
 		}
 	}
 
+	// remove this later just to have NewStaticFieldsTraces in the response
+	fields := constants.NewStaticFieldsTraces
+	if !r.useTraceNewSchema {
+		fields = constants.DeprecatedStaticFieldsTraces
+	}
+
 	// add the new static fields
-	for _, f := range constants.NewStaticFieldsTraces {
+	for _, f := range fields {
 		if (v3.AttributeKey{} == f) {
 			continue
 		}
@@ -4339,49 +4278,7 @@ func (r *ClickHouseReader) GetTraceAttributeKeysV2(ctx context.Context, req *v3.
 	return &response, nil
 }
 
-func (r *ClickHouseReader) GetTraceAttributeKeys(ctx context.Context, req *v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
+func (r *ClickHouseReader) GetTraceAttributeValues(ctx context.Context, req *v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
-	if r.useTraceNewSchema {
-		return r.GetTraceAttributeKeysV2(ctx, req)
-	}
-
-	var query string
-	var err error
-	var rows driver.Rows
-	var response v3.FilterAttributeKeyResponse
-
-	query = fmt.Sprintf("SELECT DISTINCT(tagKey), tagType, dataType, isColumn FROM %s.%s WHERE tagKey ILIKE $1", r.TraceDB, r.spanAttributeTable)
-
-	if req.Limit != 0 {
-		query = query + fmt.Sprintf(" LIMIT %d;", req.Limit)
-	}
-	rows, err = r.db.Query(ctx, query, fmt.Sprintf("%%%s%%", req.SearchText))
-
-	if err != nil {
-		zap.L().Error("Error while executing query", zap.Error(err))
-		return nil, fmt.Errorf("error while executing query: %s", err.Error())
-	}
-	defer rows.Close()
-
-	var tagKey string
-	var dataType string
-	var tagType string
-	var isColumn bool
-	for rows.Next() {
-		if err := rows.Scan(&tagKey, &tagType, &dataType, &isColumn); err != nil {
-			return nil, fmt.Errorf("error while scanning rows: %s", err.Error())
-		}
-		key := v3.AttributeKey{
-			Key:      tagKey,
-			DataType: v3.AttributeKeyDataType(dataType),
-			Type:     v3.AttributeKeyType(tagType),
-			IsColumn: isColumn,
-		}
-		response.AttributeKeys = append(response.AttributeKeys, key)
-	}
-	return &response, nil
-}
-
-func (r *ClickHouseReader) GetTraceAttributeValuesV2(ctx context.Context, req *v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
 	var query string
 	var filterValueColumn string
 	var err error
@@ -4427,7 +4324,11 @@ func (r *ClickHouseReader) GetTraceAttributeValuesV2(ctx context.Context, req *v
 		}
 
 		// TODO(nitya): remove 24 hour limit in future after checking the perf/resource implications
-		query = fmt.Sprintf("select distinct %s from %s.%s where ts_bucket_start >= toUInt64(toUnixTimestamp(now() - INTERVAL 48 HOUR)) AND timestamp >= toDateTime64(now() - INTERVAL 48 HOUR, 9) and %s ILIKE $1 limit $2", selectKey, r.TraceDB, r.traceTableName, filterValueColumnWhere)
+		where := "timestamp >= toDateTime64(now() - INTERVAL 48 HOUR, 9)"
+		if r.useTraceNewSchema {
+			where += " AND ts_bucket_start >= toUInt64(toUnixTimestamp(now() - INTERVAL 48 HOUR))"
+		}
+		query = fmt.Sprintf("select distinct %s from %s.%s where %s and %s ILIKE $1 limit $2", selectKey, r.TraceDB, r.traceTableName, where, filterValueColumnWhere)
 		rows, err = r.db.Query(ctx, query, searchText, req.Limit)
 	} else {
 		filterValueColumnWhere := filterValueColumn
@@ -4466,63 +4367,6 @@ func (r *ClickHouseReader) GetTraceAttributeValuesV2(ctx context.Context, req *v
 	return &attributeValues, nil
 }
 
-func (r *ClickHouseReader) GetTraceAttributeValues(ctx context.Context, req *v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
-
-	if r.useTraceNewSchema {
-		return r.GetTraceAttributeValuesV2(ctx, req)
-	}
-	var query string
-	var err error
-	var rows driver.Rows
-	var attributeValues v3.FilterAttributeValueResponse
-	// if dataType or tagType is not present return empty response
-	if len(req.FilterAttributeKeyDataType) == 0 || len(req.TagType) == 0 || req.FilterAttributeKey == "body" {
-		return &v3.FilterAttributeValueResponse{}, nil
-	}
-	switch req.FilterAttributeKeyDataType {
-	case v3.AttributeKeyDataTypeString:
-		query = fmt.Sprintf("SELECT DISTINCT stringTagValue from %s.%s WHERE tagKey = $1 AND stringTagValue ILIKE $2 AND tagType=$3 limit $4", r.TraceDB, r.spanAttributeTable)
-		rows, err = r.db.Query(ctx, query, req.FilterAttributeKey, fmt.Sprintf("%%%s%%", req.SearchText), req.TagType, req.Limit)
-		if err != nil {
-			zap.L().Error("Error while executing query", zap.Error(err))
-			return nil, fmt.Errorf("error while executing query: %s", err.Error())
-		}
-		defer rows.Close()
-
-		var strAttributeValue string
-		for rows.Next() {
-			if err := rows.Scan(&strAttributeValue); err != nil {
-				return nil, fmt.Errorf("error while scanning rows: %s", err.Error())
-			}
-			attributeValues.StringAttributeValues = append(attributeValues.StringAttributeValues, strAttributeValue)
-		}
-	case v3.AttributeKeyDataTypeFloat64, v3.AttributeKeyDataTypeInt64:
-		query = fmt.Sprintf("SELECT DISTINCT float64TagValue from %s.%s where tagKey = $1 AND toString(float64TagValue) ILIKE $2 AND tagType=$3 limit $4", r.TraceDB, r.spanAttributeTable)
-		rows, err = r.db.Query(ctx, query, req.FilterAttributeKey, fmt.Sprintf("%%%s%%", req.SearchText), req.TagType, req.Limit)
-		if err != nil {
-			zap.L().Error("Error while executing query", zap.Error(err))
-			return nil, fmt.Errorf("error while executing query: %s", err.Error())
-		}
-		defer rows.Close()
-
-		var numberAttributeValue sql.NullFloat64
-		for rows.Next() {
-			if err := rows.Scan(&numberAttributeValue); err != nil {
-				return nil, fmt.Errorf("error while scanning rows: %s", err.Error())
-			}
-			if numberAttributeValue.Valid {
-				attributeValues.NumberAttributeValues = append(attributeValues.NumberAttributeValues, numberAttributeValue.Float64)
-			}
-		}
-	case v3.AttributeKeyDataTypeBool:
-		attributeValues.BoolAttributeValues = []bool{true, false}
-	default:
-		return nil, fmt.Errorf("invalid data type")
-	}
-
-	return &attributeValues, nil
-}
-
 func (r *ClickHouseReader) GetSpanAttributeKeysV2(ctx context.Context) (map[string]v3.AttributeKey, error) {
 	var query string
 	var err error
@@ -4606,6 +4450,12 @@ func (r *ClickHouseReader) GetSpanAttributeKeys(ctx context.Context) (map[string
 		}
 		response[tagKey] = key
 	}
+
+	// add the deprecated static fields as they are not present in spanAttributeKeysTable
+	for _, f := range constants.DeprecatedStaticFieldsTraces {
+		response[f.Key] = f
+	}
+
 	return response, nil
 }
 

pkg/query-service/app/logs/v3/enrich_query.go

@@ -18,7 +18,7 @@ func EnrichmentRequired(params *v3.QueryRangeParamsV3) bool {
 
 	// Build queries for each builder query
 	for queryName, query := range compositeQuery.BuilderQueries {
-		if query.Expression != queryName && query.DataSource != v3.DataSourceLogs {
+		if query.Expression != queryName || query.DataSource != v3.DataSourceLogs {
 			continue
 		}
 

pkg/query-service/constants/constants.go

@@ -582,6 +582,21 @@ var DeprecatedStaticFieldsTraces = map[string]v3.AttributeKey{
 		DataType: v3.AttributeKeyDataTypeString,
 		IsColumn: true,
 	},
+	"flags": {
+		Key:      "flags",
+		DataType: v3.AttributeKeyDataTypeInt64,
+		IsColumn: true,
+	},
+	"name": {
+		Key:      "name",
+		DataType: v3.AttributeKeyDataTypeString,
+		IsColumn: true,
+	},
+	"kind": {
+		Key:      "kind",
+		DataType: v3.AttributeKeyDataTypeString,
+		IsColumn: true,
+	},
 	"spanKind": {
 		Key:      "spanKind",
 		DataType: v3.AttributeKeyDataTypeString,