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: Implement the behavior of the MetricTimeDimensionTransformNode.
  
  This node will create an output data set that is similar to the input data set, but the measure instances it
  contains is a subset of the input data set. Only measure instances that have an aggregation time dimension
  matching the one defined in the node will be passed. In addition, an additional time dimension instance for
  "metric time" will be included. See DataSet.metric_time_dimension_reference().
docstr_tok: ["Implement","the","behavior","of","the","MetricTimeDimensionTransformNode",".","This","node","will","create","an","output","data","set","that","is","similar","to","the","input","data","set",",","but","the","measure","instances","it","contains","is","a","subset","of","the","input","data","set",".","Only","measure","instances","that","have","an","aggregation","time","dimension","matching","the","one","defined","in","the","node","will","be","passed",".","In","addition",",","an","additional","time","dimension","instance","for","''","metric","time","''","will","be","included",".","See","DataSet.metric_time_dimension_reference","(",")","."]
code: def visit_metric_time_dimension_transform_node(
      self, node: MetricTimeDimensionTransformNode
  ) -> SqlDataSet:
      """Implement the behavior of the MetricTimeDimensionTransformNode.
  
      This node will create an output data set that is similar to the input data set, but the measure instances it
      contains is a subset of the input data set. Only measure instances that have an aggregation time dimension
      matching the one defined in the node will be passed. In addition, an additional time dimension instance for
      "metric time" will be included. See DataSet.metric_time_dimension_reference().
      """
      input_data_set: SqlDataSet = node.parent_node.accept(self)
  
      # Find which measures have an aggregation time dimension that is the same as the one specified in the node.
      # Only these measures will be in the output data set.
      output_measure_instances = []
      for (
          measure_instance
      ) in input_data_set.instance_set.measure_instances:
          semantic_model = self._semantic_model_lookup.get_by_reference(
              semantic_model_reference=measure_instance.origin_semantic_model_reference.semantic_model_reference
          )
          assert semantic_model is not None, (
              f"{measure_instance} was defined from "
              f"{measure_instance.origin_semantic_model_reference.semantic_model_reference}, but that can't be found"
          )
          aggregation_time_dimension_for_measure = (
              semantic_model.checked_agg_time_dimension_for_measure(
                  measure_reference=measure_instance.spec.reference
              )
          )
          if (
              aggregation_time_dimension_for_measure
              == node.aggregation_time_dimension_reference
          ):
              output_measure_instances.append(measure_instance)
  
      if len(output_measure_instances) == 0:
          raise RuntimeError(
              f"No measure instances in the input source match the aggregation time dimension "
              f"{node.aggregation_time_dimension_reference}. Check if the dataflow plan was constructed correctly."
          )
  
      # Find time dimension instances that refer to the same dimension as the one specified in the node.
      matching_time_dimension_instances = []
      for (
          time_dimension_instance
      ) in input_data_set.instance_set.time_dimension_instances:
          # The specification for the time dimension to use for aggregation is the local one.
          if (
              len(time_dimension_instance.spec.entity_links) == 0
              and time_dimension_instance.spec.reference
              == node.aggregation_time_dimension_reference
          ):
              matching_time_dimension_instances.append(
                  time_dimension_instance
              )
  
      output_time_dimension_instances: List[TimeDimensionInstance] = []
      output_time_dimension_instances.extend(
          input_data_set.instance_set.time_dimension_instances
      )
      output_column_to_input_column: OrderedDict[
          str, str
      ] = OrderedDict()
  
      # For those matching time dimension instances, create the analog metric time dimension instances for the output.
      for (
          matching_time_dimension_instance
      ) in matching_time_dimension_instances:
          metric_time_dimension_spec = DataSet.metric_time_dimension_spec(
              time_granularity=matching_time_dimension_instance.spec.time_granularity,
              date_part=matching_time_dimension_instance.spec.date_part,
          )
          metric_time_dimension_column_association = (
              self._column_association_resolver.resolve_spec(
                  metric_time_dimension_spec
              )
          )
          output_time_dimension_instances.append(
              TimeDimensionInstance(
                  defined_from=matching_time_dimension_instance.defined_from,
                  associated_columns=(
                      self._column_association_resolver.resolve_spec(
                          metric_time_dimension_spec
                      ),
                  ),
                  spec=metric_time_dimension_spec,
              )
          )
          output_column_to_input_column[
              metric_time_dimension_column_association.column_name
          ] = (
              matching_time_dimension_instance.associated_column.column_name
          )
  
      output_instance_set = InstanceSet(
          measure_instances=tuple(output_measure_instances),
          dimension_instances=input_data_set.instance_set.dimension_instances,
          time_dimension_instances=tuple(
              output_time_dimension_instances
          ),
          entity_instances=input_data_set.instance_set.entity_instances,
          metric_instances=input_data_set.instance_set.metric_instances,
      )
      output_instance_set = ChangeAssociatedColumns(
          self._column_association_resolver
      ).transform(output_instance_set)
  
      from_data_set_alias = self._next_unique_table_alias()
  
      return SqlDataSet(
          instance_set=output_instance_set,
          sql_select_node=SqlSelectStatementNode(
              description=node.description,
              # This creates select expressions for all columns referenced in the instance set.
              select_columns=CreateSelectColumnsForInstances(
                  column_resolver=self._column_association_resolver,
                  table_alias=from_data_set_alias,
                  output_to_input_column_mapping=output_column_to_input_column,
              )
              .transform(output_instance_set)
              .as_tuple(),
              from_source=input_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 19 for calling the DataflowToSqlQueryPlanConverter obj's underlying member method code with example usage: obj.visit_metric_time_dimension_transform_node(node) and returns: SqlDataSet