hanahmily commented on code in PR #943:
URL:
https://github.com/apache/skywalking-banyandb/pull/943#discussion_r2697191629
##########
banyand/query/processor.go:
##########
@@ -161,6 +162,138 @@ type measureQueryProcessor struct {
*bus.UnImplementedHealthyListener
}
+// measureExecutionContext holds the common execution context for measure
queries.
+type measureExecutionContext struct {
+ ml *logger.Logger
+ ecc []executor.MeasureExecutionContext
+ metadata []*commonv1.Metadata
+ schemas []logical.Schema
+}
+
+// buildMeasureContext builds the execution context for a measure query.
+func buildMeasureContext(measureService measure.Service, log *logger.Logger,
queryCriteria *measurev1.QueryRequest, logPrefix string)
(*measureExecutionContext, error) {
+ var metadata []*commonv1.Metadata
+ var schemas []logical.Schema
+ var ecc []executor.MeasureExecutionContext
+ for i := range queryCriteria.Groups {
+ meta := &commonv1.Metadata{
+ Name: queryCriteria.Name,
+ Group: queryCriteria.Groups[i],
+ }
+ ec, ecErr := measureService.Measure(meta)
+ if ecErr != nil {
+ return nil, fmt.Errorf("fail to get execution context
for measure %s: %w", meta.GetName(), ecErr)
+ }
+ s, schemaErr := logical_measure.BuildSchema(ec.GetSchema(),
ec.GetIndexRules())
+ if schemaErr != nil {
+ return nil, fmt.Errorf("fail to build schema for
measure %s: %w", meta.GetName(), schemaErr)
+ }
+ ecc = append(ecc, ec)
+ schemas = append(schemas, s)
+ metadata = append(metadata, meta)
+ }
+ ml := log.Named(logPrefix, queryCriteria.Groups[0], queryCriteria.Name)
+ return &measureExecutionContext{
+ metadata: metadata,
+ schemas: schemas,
+ ecc: ecc,
+ ml: ml,
+ }, nil
+}
+
+// executeMeasurePlan executes the measure query plan and returns the iterator.
+func executeMeasurePlan(ctx context.Context, queryCriteria
*measurev1.QueryRequest, mctx *measureExecutionContext) (executor.MIterator,
logical.Plan, error) {
+ plan, planErr := logical_measure.Analyze(queryCriteria, mctx.metadata,
mctx.schemas, mctx.ecc)
+ if planErr != nil {
+ return nil, nil, fmt.Errorf("fail to analyze the query request
for measure %s: %w", queryCriteria.GetName(), planErr)
+ }
+ if e := mctx.ml.Debug(); e.Enabled() {
+ e.Str("plan", plan.String()).Msg("query plan")
+ }
+ mIterator, execErr := plan.(executor.MeasureExecutable).Execute(ctx)
+ if execErr != nil {
+ mctx.ml.Error().Err(execErr).RawJSON("req",
logger.Proto(queryCriteria)).Msg("fail to query")
+ return nil, nil, fmt.Errorf("fail to execute the query plan for
measure %s: %w", queryCriteria.GetName(), execErr)
+ }
+ return mIterator, plan, nil
+}
+
+// extractTagValuesFromInternalDataPoints extracts tag values from
InternalDataPoints for RewriteAggTopNResult.
+func extractTagValuesFromInternalDataPoints(dataPoints
[]*measurev1.InternalDataPoint, groupByTags []string)
map[string][]*modelv1.TagValue {
+ tagValueMap := make(map[string][]*modelv1.TagValue)
+ for _, idp := range dataPoints {
+ dp := idp.DataPoint
+ if dp == nil {
+ continue
+ }
+ for _, tagFamily := range dp.GetTagFamilies() {
+ for _, tag := range tagFamily.GetTags() {
+ tagName := tag.GetKey()
+ if len(groupByTags) == 0 ||
slices.Contains(groupByTags, tagName) {
+ tagValueMap[tagName] =
append(tagValueMap[tagName], tag.GetValue())
+ }
+ }
+ }
+ }
+ return tagValueMap
+}
+
+// collectInternalDataPoints collects InternalDataPoints (with ShardId) from
the iterator.
+func collectInternalDataPoints(mIterator executor.MIterator)
[]*measurev1.InternalDataPoint {
+ result := make([]*measurev1.InternalDataPoint, 0)
+ for mIterator.Next() {
+ current := mIterator.Current()
+ if len(current) > 0 {
+ dp := current[0]
+ internalDp := &measurev1.InternalDataPoint{
+ DataPoint: dp,
+ ShardId: uint32(mIterator.CurrentShardID()),
+ }
+ result = append(result, internalDp)
+ }
+ }
+ return result
+}
+
+// extractTagValuesFromDataPoints extracts tag values from DataPoints for
RewriteAggTopNResult.
+func extractTagValuesFromDataPoints(dataPoints []*measurev1.DataPoint,
groupByTags []string) map[string][]*modelv1.TagValue {
+ tagValueMap := make(map[string][]*modelv1.TagValue)
+ for _, dp := range dataPoints {
+ for _, tagFamily := range dp.GetTagFamilies() {
+ for _, tag := range tagFamily.GetTags() {
+ tagName := tag.GetKey()
+ if len(groupByTags) == 0 ||
slices.Contains(groupByTags, tagName) {
+ tagValueMap[tagName] =
append(tagValueMap[tagName], tag.GetValue())
+ }
+ }
+ }
+ }
+ return tagValueMap
+}
Review Comment:
Would you please extract a common helper?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
