Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

use metrics_view valid spec instead of spec #5779

Merged
merged 2 commits into from
Sep 30, 2024
Merged

use metrics_view valid spec instead of spec #5779

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
00454c8
use metrics_view valid spec instead of spec
k-anshul Sep 25, 2024
4ea9f94
Merge remote-tracking branch 'origin/main' into use_mv_validspec
k-anshul Sep 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions runtime/pkg/metricssql/parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,9 @@ type query struct {
priority int

// fields available after parsing FROM clause
metricsView *runtimev1.MetricsViewV2
dims map[string]any
measures map[string]any
metricsViewSpec *runtimev1.MetricsViewSpec
dims map[string]any
measures map[string]any
}

// Rewrite parses a metrics SQL query and compiles it to a metricview.Query.
Expand Down Expand Up @@ -155,11 +155,11 @@ func (q *query) parseFrom(ctx context.Context, node *ast.TableRefsClause) error
}

q.q.MetricsView = tblName.Name.String()
q.metricsView = mv.GetMetricsView()
spec := mv.GetMetricsView().State.ValidSpec
if spec == nil {
return fmt.Errorf("metrics view %q is not valid: (status: %q, error: %q)", mv.Meta.GetName(), mv.Meta.ReconcileStatus, mv.Meta.ReconcileError)
}
q.metricsViewSpec = spec
q.measures = make(map[string]any, len(spec.Measures))
for _, measure := range spec.Measures {
q.measures[measure.Name] = nil
Expand Down Expand Up @@ -270,7 +270,7 @@ func (q *query) parseColumnNameExpr(in ast.Node) (string, string, error) {
return col, "DIMENSION", nil
} else if _, ok := q.measures[col]; ok {
return col, "MEASURE", nil
} else if q.metricsView.Spec.TimeDimension == col {
} else if q.metricsViewSpec.TimeDimension == col {
return col, "DIMENSION", nil
}
return "", "", fmt.Errorf("metrics sql: selected column `%s` not found in dimensions/measures in metrics view", col)
Expand Down
13 changes: 6 additions & 7 deletions runtime/pkg/metricssql/time_range_parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,21 +68,20 @@ func (q *query) parseTimeRangeEnd(ctx context.Context, node *ast.FuncCallExpr) (
}

func (q *query) getWatermark(ctx context.Context, colName string) (watermark time.Time, err error) {
olap, release, err := q.controller.AcquireOLAP(ctx, q.metricsView.Spec.Connector)
olap, release, err := q.controller.AcquireOLAP(ctx, q.metricsViewSpec.Connector)
if err != nil {
return watermark, err
}
defer release()

spec := q.metricsView.Spec
var sql string
if colName != "" {
sql = fmt.Sprintf("SELECT MAX(%s) FROM %s ", olap.Dialect().EscapeIdentifier(colName), olap.Dialect().EscapeTable(spec.Database, spec.DatabaseSchema, spec.Table))
} else if q.metricsView.Spec.WatermarkExpression != "" {
sql = fmt.Sprintf("SELECT %s FROM %s", q.metricsView.Spec.WatermarkExpression, olap.Dialect().EscapeTable(spec.Database, spec.DatabaseSchema, spec.Table))
sql = fmt.Sprintf("SELECT MAX(%s) FROM %s ", olap.Dialect().EscapeIdentifier(colName), olap.Dialect().EscapeTable(q.metricsViewSpec.Database, q.metricsViewSpec.DatabaseSchema, q.metricsViewSpec.Table))
} else if q.metricsViewSpec.WatermarkExpression != "" {
sql = fmt.Sprintf("SELECT %s FROM %s", q.metricsViewSpec.WatermarkExpression, olap.Dialect().EscapeTable(q.metricsViewSpec.Database, q.metricsViewSpec.DatabaseSchema, q.metricsViewSpec.Table))
// todo how to handle column name here
} else if spec.TimeDimension != "" {
sql = fmt.Sprintf("SELECT MAX(%s) FROM %s", olap.Dialect().EscapeIdentifier(spec.TimeDimension), olap.Dialect().EscapeTable(spec.Database, spec.DatabaseSchema, spec.Table))
} else if q.metricsViewSpec.TimeDimension != "" {
sql = fmt.Sprintf("SELECT MAX(%s) FROM %s", olap.Dialect().EscapeIdentifier(q.metricsViewSpec.TimeDimension), olap.Dialect().EscapeTable(q.metricsViewSpec.Database, q.metricsViewSpec.DatabaseSchema, q.metricsViewSpec.Table))
} else {
return watermark, fmt.Errorf("metrics sql: no watermark or time dimension found in metrics view")
}
Expand Down
9 changes: 6 additions & 3 deletions runtime/queries/metricsview_timeseries.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,10 @@ func (q *MetricsViewTimeSeries) Export(ctx context.Context, rt *runtime.Runtime,
return err
}

mv := r.GetMetricsView().Spec
spec := r.GetMetricsView().State.ValidSpec
if spec == nil {
return fmt.Errorf("metrics view spec is not valid")
}

if opts.PreWriteHook != nil {
err = opts.PreWriteHook(q.generateFilename())
Expand All @@ -129,10 +132,10 @@ func (q *MetricsViewTimeSeries) Export(ctx context.Context, rt *runtime.Runtime,

tmp := make([]*structpb.Struct, 0, len(q.Result.Data))
meta := append([]*runtimev1.MetricsViewColumn{{
Name: mv.TimeDimension,
Name: spec.TimeDimension,
}}, q.Result.Meta...)
for _, dt := range q.Result.Data {
dt.Records.Fields[mv.TimeDimension] = structpb.NewStringValue(dt.Ts.AsTime().Format(time.RFC3339Nano))
dt.Records.Fields[spec.TimeDimension] = structpb.NewStringValue(dt.Ts.AsTime().Format(time.RFC3339Nano))
tmp = append(tmp, dt.Records)
}

Expand Down
5 changes: 4 additions & 1 deletion runtime/query.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,10 @@ func (r *Runtime) Query(ctx context.Context, instanceID string, query Query, pri

// Infer OLAP connector for common resource types used in a query
if mv := res.GetMetricsView(); mv != nil {
olapConnector = mv.Spec.Connector
if mv.State.ValidSpec == nil {
return fmt.Errorf("metrics view spec is not valid")
}
olapConnector = mv.State.ValidSpec.Connector
} else if mdl := res.GetModel(); mdl != nil {
olapConnector = mdl.Spec.OutputConnector
} else if src := res.GetSource(); src != nil {
Expand Down
9 changes: 7 additions & 2 deletions runtime/resolvers/legacy_metrics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,11 @@ func (r *legacyMetricsResolver) ResolveInteractive(ctx context.Context) (runtime
return nil, err
}

spec := metricsView.GetMetricsView().State.ValidSpec
if spec == nil {
return nil, fmt.Errorf("metrics view spec is not valid")
}

err = r.runtime.Query(ctx, r.instanceID, r.query, r.args.Priority)
if err != nil {
return nil, fmt.Errorf("failed to execute query: %w", err)
Expand All @@ -132,7 +137,7 @@ func (r *legacyMetricsResolver) ResolveInteractive(ctx context.Context) (runtime
break
}
if r.args.Format {
out = append(out, r.formatMetricsViewAggregationResult(row.AsMap(), q, metricsView.GetMetricsView().Spec.Measures))
out = append(out, r.formatMetricsViewAggregationResult(row.AsMap(), q, spec.Measures))
continue
}
out = append(out, row.AsMap())
Expand All @@ -145,7 +150,7 @@ func (r *legacyMetricsResolver) ResolveInteractive(ctx context.Context) (runtime
break
}
if r.args.Format {
out = append(out, r.formatMetricsViewComparisonResult(row, q, metricsView.GetMetricsView().Spec.Measures))
out = append(out, r.formatMetricsViewComparisonResult(row, q, spec.Measures))
continue
}
r := make(map[string]any)
Expand Down
Loading