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","."]
code: def visit_join_to_time_spine_node(
      self, node: JoinToTimeSpineNode
  ) -> SqlDataSet:  # noqa: D
      parent_data_set = node.parent_node.accept(self)
      parent_alias = self._next_unique_table_alias()
  
      # Build time spine dataset
      metric_time_dimension_instance: Optional[
          TimeDimensionInstance
      ] = None
      for instance in parent_data_set.metric_time_dimension_instances:
          if len(instance.spec.entity_links) == 0:
              # Use the instance with the lowest granularity
              if not metric_time_dimension_instance or (
                  instance.spec.time_granularity
                  < metric_time_dimension_instance.spec.time_granularity
              ):
                  metric_time_dimension_instance = instance
      assert (
          metric_time_dimension_instance
      ), "Can't join to time spine without metric time. Validations should have prevented this."
      metric_time_dimension_column_name = (
          self.column_association_resolver.resolve_spec(
              metric_time_dimension_instance.spec
          ).column_name
      )
      time_spine_alias = self._next_unique_table_alias()
      time_spine_dataset = self._make_time_spine_data_set(
          metric_time_dimension_instance=metric_time_dimension_instance,
          metric_time_dimension_column_name=metric_time_dimension_column_name,
          time_spine_source=self._time_spine_source,
          time_range_constraint=node.time_range_constraint,
      )
  
      # Build join expression
      join_description = SqlQueryPlanJoinBuilder.make_join_to_time_spine_join_description(
          node=node,
          time_spine_alias=time_spine_alias,
          metric_time_dimension_column_name=metric_time_dimension_column_name,
          parent_sql_select_node=parent_data_set.sql_select_node,
          parent_alias=parent_alias,
      )
  
      # Use all instances EXCEPT metric_time from parent data set.
      non_metric_time_parent_instance_set = InstanceSet(
          measure_instances=parent_data_set.instance_set.measure_instances,
          dimension_instances=parent_data_set.instance_set.dimension_instances,
          time_dimension_instances=tuple(
              time_dimension_instance
              for time_dimension_instance in parent_data_set.instance_set.time_dimension_instances
              if time_dimension_instance.spec.element_name
              != DataSet.metric_time_dimension_reference().element_name
          ),
          entity_instances=parent_data_set.instance_set.entity_instances,
          metric_instances=parent_data_set.instance_set.metric_instances,
          metadata_instances=parent_data_set.instance_set.metadata_instances,
      )
      parent_select_columns = create_select_columns_for_instance_sets(
          self._column_association_resolver,
          OrderedDict(
              {parent_alias: non_metric_time_parent_instance_set}
          ),
      )
  
      # Use time instance from time spine to replace metric_time instances.
      assert (
          len(time_spine_dataset.instance_set.time_dimension_instances)
          == 1
          and len(time_spine_dataset.sql_select_node.select_columns)
          == 1
      ), "Time spine dataset not configured properly. Expected exactly one column."
      time_spine_dim_instance = (
          time_spine_dataset.instance_set.time_dimension_instances[0]
      )
      time_spine_column_select_expr: Union[
          SqlColumnReferenceExpression, SqlDateTruncExpression
      ] = SqlColumnReferenceExpression(
          SqlColumnReference(
              table_alias=time_spine_alias,
              column_name=time_spine_dim_instance.spec.qualified_name,
          )
      )
  
      # Add requested granularities (if different from time_spine) and date_parts to time spine column.
      time_spine_select_columns = []
      time_spine_dim_instances = []
      where: Optional[SqlExpressionNode] = None
      for (
          requested_time_dimension_spec
      ) in node.requested_metric_time_dimension_specs:
          # Apply granularity to time spine column select expression.
          if (
              requested_time_dimension_spec.time_granularity
              == time_spine_dim_instance.spec.time_granularity
          ):
              select_expr: SqlExpressionNode = (
                  time_spine_column_select_expr
              )
          else:
              select_expr = SqlDateTruncExpression(
                  time_granularity=requested_time_dimension_spec.time_granularity,
                  arg=time_spine_column_select_expr,
              )
              if node.offset_to_grain:
                  # Filter down to one row per granularity period
                  new_filter = SqlComparisonExpression(
                      left_expr=select_expr,
                      comparison=SqlComparison.EQUALS,
                      right_expr=time_spine_column_select_expr,
                  )
                  if not where:
                      where = new_filter
                  else:
                      where = SqlLogicalExpression(
                          operator=SqlLogicalOperator.OR,
                          args=(where, new_filter),
                      )
          # Apply date_part to time spine column select expression.
          if requested_time_dimension_spec.date_part:
              select_expr = SqlExtractExpression(
                  date_part=requested_time_dimension_spec.date_part,
                  arg=select_expr,
              )
          time_dim_spec = TimeDimensionSpec(
              element_name=time_spine_dim_instance.spec.element_name,
              entity_links=time_spine_dim_instance.spec.entity_links,
              time_granularity=requested_time_dimension_spec.time_granularity,
              date_part=requested_time_dimension_spec.date_part,
              aggregation_state=time_spine_dim_instance.spec.aggregation_state,
          )
          time_spine_dim_instance = TimeDimensionInstance(
              defined_from=time_spine_dim_instance.defined_from,
              associated_columns=(
                  self._column_association_resolver.resolve_spec(
                      time_dim_spec
                  ),
              ),
              spec=time_dim_spec,
          )
          time_spine_dim_instances.append(time_spine_dim_instance)
          time_spine_select_columns.append(
              SqlSelectColumn(
                  expr=select_expr,
                  column_alias=time_spine_dim_instance.associated_column.column_name,
              )
          )
      time_spine_instance_set = InstanceSet(
          time_dimension_instances=tuple(time_spine_dim_instances)
      )
  
      return SqlDataSet(
          instance_set=InstanceSet.merge(
              [
                  time_spine_instance_set,
                  non_metric_time_parent_instance_set,
              ]
          ),
          sql_select_node=SqlSelectStatementNode(
              description=node.description,
              select_columns=tuple(time_spine_select_columns)
              + parent_select_columns,
              from_source=time_spine_dataset.sql_select_node,
              from_source_alias=time_spine_alias,
              joins_descs=(join_description,),
              group_bys=(),
              order_bys=(),
              where=where,
          ),
      )
  
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 21 for calling the DataflowToSqlQueryPlanConverter obj's underlying member method code with example usage: obj.visit_join_to_time_spine_node(node) and returns: SqlDataSet