class_docstr: Generates an SQL query plan from a node in the a metric dataflow plan.
class_docstr_tok: ["Generates","an","SQL","query","plan","from","a","node","in","the","a","metric","dataflow","plan","."]
docstr: Generates the query that realizes the behavior of ComputeMetricsNode.
docstr_tok: ["Generates","the","query","that","realizes","the","behavior","of","ComputeMetricsNode","."]
code: def visit_compute_metrics_node(
      self, node: ComputeMetricsNode
  ) -> SqlDataSet:
      """Generates the query that realizes the behavior of ComputeMetricsNode."""
      from_data_set: SqlDataSet = node.parent_node.accept(self)
      from_data_set_alias = self._next_unique_table_alias()
  
      # TODO: Check that all measures for the metrics are in the input instance set
      # The desired output instance set has no measures, so create a copy with those removed.
      output_instance_set: InstanceSet = (
          from_data_set.instance_set.transform(RemoveMeasures())
      )
  
      # Also, the output columns should always follow the resolver format.
      output_instance_set = output_instance_set.transform(
          ChangeAssociatedColumns(self._column_association_resolver)
      )
      output_instance_set = output_instance_set.transform(
          RemoveMetrics()
      )
  
      non_metric_select_column_set: SelectColumnSet = (
          output_instance_set.transform(
              CreateSelectColumnsForInstances(
                  table_alias=from_data_set_alias,
                  column_resolver=self._column_association_resolver,
              )
          )
      )
  
      # Add select columns that would compute the metrics to the select columns.
      metric_select_columns = []
      metric_instances = []
      for metric_spec in node.metric_specs:
          metric = self._metric_lookup.get_metric(metric_spec.reference)
  
          metric_expr: Optional[SqlExpressionNode] = None
          input_measure: Optional[MetricInputMeasure] = None
          if metric.type is MetricType.RATIO:
              numerator = metric.type_params.numerator
              denominator = metric.type_params.denominator
              assert (
                  numerator is not None and denominator is not None
              ), "Missing numerator or denominator for ratio metric, this should have been caught in validation!"
              numerator_column_name = (
                  self._column_association_resolver.resolve_spec(
                      MetricSpec.from_reference(
                          numerator.post_aggregation_reference
                      )
                  ).column_name
              )
              denominator_column_name = (
                  self._column_association_resolver.resolve_spec(
                      MetricSpec.from_reference(
                          denominator.post_aggregation_reference
                      )
                  ).column_name
              )
  
              metric_expr = SqlRatioComputationExpression(
                  numerator=SqlColumnReferenceExpression(
                      SqlColumnReference(
                          table_alias=from_data_set_alias,
                          column_name=numerator_column_name,
                      )
                  ),
                  denominator=SqlColumnReferenceExpression(
                      SqlColumnReference(
                          table_alias=from_data_set_alias,
                          column_name=denominator_column_name,
                      )
                  ),
              )
          elif metric.type is MetricType.SIMPLE:
              if len(metric.input_measures) > 0:
                  assert (
                      len(metric.input_measures) == 1
                  ), "Simple metrics should always source from exactly 1 measure."
                  input_measure = metric.input_measures[0]
                  expr = self._column_association_resolver.resolve_spec(
                      MeasureSpec(
                          element_name=input_measure.post_aggregation_measure_reference.element_name
                      )
                  ).column_name
              else:
                  expr = metric.name
              metric_expr = self.__make_col_reference_or_coalesce_expr(
                  column_name=expr,
                  input_measure=input_measure,
                  from_data_set_alias=from_data_set_alias,
              )
          elif metric.type is MetricType.CUMULATIVE:
              assert (
                  len(metric.measure_references) == 1
              ), "Cumulative metrics should always source from exactly 1 measure."
              input_measure = metric.input_measures[0]
              expr = self._column_association_resolver.resolve_spec(
                  MeasureSpec(
                      element_name=input_measure.post_aggregation_measure_reference.element_name
                  )
              ).column_name
              metric_expr = self.__make_col_reference_or_coalesce_expr(
                  column_name=expr,
                  input_measure=input_measure,
                  from_data_set_alias=from_data_set_alias,
              )
          elif metric.type is MetricType.DERIVED:
              assert metric.type_params.expr
              metric_expr = SqlStringExpression(
                  sql_expr=metric.type_params.expr
              )
          elif metric.type == MetricType.CONVERSION:
              conversion_type_params = (
                  metric.type_params.conversion_type_params
              )
              assert (
                  conversion_type_params
              ), "A conversion metric should have type_params.conversion_type_params defined."
              base_measure = conversion_type_params.base_measure
              conversion_measure = (
                  conversion_type_params.conversion_measure
              )
              base_measure_column = self._column_association_resolver.resolve_spec(
                  MeasureSpec(
                      element_name=base_measure.post_aggregation_measure_reference.element_name
                  )
              ).column_name
              conversion_measure_column = self._column_association_resolver.resolve_spec(
                  MeasureSpec(
                      element_name=conversion_measure.post_aggregation_measure_reference.element_name
                  )
              ).column_name
  
              calculation_type = conversion_type_params.calculation
              conversion_column_reference = (
                  SqlColumnReferenceExpression(
                      SqlColumnReference(
                          table_alias=from_data_set_alias,
                          column_name=conversion_measure_column,
                      )
                  )
              )
              base_column_reference = SqlColumnReferenceExpression(
                  SqlColumnReference(
                      table_alias=from_data_set_alias,
                      column_name=base_measure_column,
                  )
              )
              if (
                  calculation_type
                  == ConversionCalculationType.CONVERSION_RATE
              ):
                  metric_expr = SqlRatioComputationExpression(
                      numerator=conversion_column_reference,
                      denominator=base_column_reference,
                  )
              elif (
                  calculation_type
                  == ConversionCalculationType.CONVERSIONS
              ):
                  metric_expr = conversion_column_reference
          else:
              assert_values_exhausted(metric.type)
  
          assert metric_expr
  
          output_column_association = (
              self._column_association_resolver.resolve_spec(
                  metric_spec
              )
          )
          metric_select_columns.append(
              SqlSelectColumn(
                  expr=metric_expr,
                  column_alias=output_column_association.column_name,
              )
          )
          metric_instances.append(
              MetricInstance(
                  associated_columns=(output_column_association,),
                  defined_from=MetricModelReference(
                      metric_name=metric_spec.element_name
                  ),
                  spec=metric_spec,
              )
          )
      output_instance_set = output_instance_set.transform(
          AddMetrics(metric_instances)
      )
  
      combined_select_column_set = non_metric_select_column_set.merge(
          SelectColumnSet(metric_columns=metric_select_columns)
      )
  
      return SqlDataSet(
          instance_set=output_instance_set,
          sql_select_node=SqlSelectStatementNode(
              description=node.description,
              select_columns=combined_select_column_set.as_tuple(),
              from_source=from_data_set.sql_select_node,
              from_source_alias=from_data_set_alias,
              joins_descs=(),
              group_bys=(),
              where=None,
              order_bys=(),
          ),
      )
  
imports: from __future__ import annotations
  import logging
  from collections import OrderedDict
  from typing import List, Optional, Sequence, Tuple, Union
  from dbt_semantic_interfaces.enum_extension import assert_values_exhausted
  from dbt_semantic_interfaces.protocols.metric import MetricInputMeasure, MetricType
  from dbt_semantic_interfaces.references import MetricModelReference
  from dbt_semantic_interfaces.type_enums.aggregation_type import AggregationType
  from dbt_semantic_interfaces.type_enums.conversion_calculation_type import ConversionCalculationType
  from dbt_semantic_interfaces.validations.unique_valid_name import MetricFlowReservedKeywords
  from metricflow.aggregation_properties import AggregationState
  from metricflow.dag.id_generation import IdGeneratorRegistry
  from metricflow.dataflow.dataflow_plan import AddGeneratedUuidColumnNode, AggregateMeasuresNode, BaseOutput, CombineAggregatedOutputsNode, ComputedMetricsOutput, ComputeMetricsNode, ConstrainTimeRangeNode, DataflowPlanNodeVisitor, FilterElementsNode, JoinConversionEventsNode, JoinOverTimeRangeNode, JoinToBaseOutputNode, JoinToTimeSpineNode, MetricTimeDimensionTransformNode, MinMaxNode, OrderByLimitNode, ReadSqlSourceNode, SemiAdditiveJoinNode, WhereConstraintNode, WriteToResultDataframeNode, WriteToResultTableNode
  from metricflow.dataset.dataset import DataSet
  from metricflow.dataset.sql_dataset import SqlDataSet
  from metricflow.filters.time_constraint import TimeRangeConstraint
  from metricflow.instances import InstanceSet, MetadataInstance, MetricInstance, TimeDimensionInstance
  from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
  from metricflow.plan_conversion.instance_converters import AddLinkToLinkableElements, AddMetadata, AddMetrics, AliasAggregatedMeasures, ChangeAssociatedColumns, ChangeMeasureAggregationState, ConvertToMetadata, CreateSelectColumnForCombineOutputNode, CreateSelectColumnsForInstances, CreateSelectColumnsWithMeasuresAggregated, CreateSqlColumnReferencesForInstances, FilterElements, FilterLinkableInstancesWithLeadingLink, RemoveMeasures, RemoveMetrics, UpdateMeasureFillNullsWith, create_select_columns_for_instance_sets
  from metricflow.plan_conversion.select_column_gen import SelectColumnSet
  from metricflow.plan_conversion.spec_transforms import CreateColumnAssociations, CreateSelectCoalescedColumnsForLinkableSpecs, SelectOnlyLinkableSpecs
  from metricflow.plan_conversion.sql_join_builder import AnnotatedSqlDataSet, ColumnEqualityDescription, SqlQueryPlanJoinBuilder
  from metricflow.plan_conversion.time_spine import TimeSpineSource
  from metricflow.protocols.sql_client import SqlEngine
  from metricflow.specs.column_assoc import ColumnAssociation, ColumnAssociationResolver, SingleColumnCorrelationKey
  from metricflow.specs.specs import InstanceSpecSet, MeasureSpec, MetadataSpec, MetricSpec, TimeDimensionSpec
  from metricflow.sql.optimizer.optimization_levels import SqlQueryOptimizationLevel, SqlQueryOptimizerConfiguration
  from metricflow.sql.sql_exprs import SqlAggregateFunctionExpression, SqlBetweenExpression, SqlColumnReference, SqlColumnReferenceExpression, SqlComparison, SqlComparisonExpression, SqlDateTruncExpression, SqlExpressionNode, SqlExtractExpression, SqlFunction, SqlFunctionExpression, SqlGenerateUuidExpression, SqlLogicalExpression, SqlLogicalOperator, SqlRatioComputationExpression, SqlStringExpression, SqlStringLiteralExpression, SqlWindowFunction, SqlWindowFunctionExpression, SqlWindowOrderByArgument
  from metricflow.sql.sql_plan import SqlJoinDescription, SqlJoinType, SqlOrderByDescription, SqlQueryPlan, SqlQueryPlanNode, SqlSelectColumn, SqlSelectStatementNode, SqlTableFromClauseNode
  from metricflow.time.time_constants import ISO8601_PYTHON_FORMAT
label_desc: Use image node_id 9 for calling the DataflowToSqlQueryPlanConverter obj's underlying member method code with example usage: obj.visit_compute_metrics_node(node) and returns: SqlDataSet