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: Implements the behaviour of SemiAdditiveJoinNode.
  
  This node will get the build a data set row filtered by the aggregate function on the
  specified dimension that is non-additive. Then that dataset would be joined with the input data
  on that dimension along with grouping by entities that are also passed in.
docstr_tok: ["Implements","the","behaviour","of","SemiAdditiveJoinNode",".","This","node","will","get","the","build","a","data","set","row","filtered","by","the","aggregate","function","on","the","specified","dimension","that","is","non-additive",".","Then","that","dataset","would","be","joined","with","the","input","data","on","that","dimension","along","with","grouping","by","entities","that","are","also","passed","in","."]
code: def visit_semi_additive_join_node(
      self, node: SemiAdditiveJoinNode
  ) -> SqlDataSet:
      """Implements the behaviour of SemiAdditiveJoinNode.
  
      This node will get the build a data set row filtered by the aggregate function on the
      specified dimension that is non-additive. Then that dataset would be joined with the input data
      on that dimension along with grouping by entities that are also passed in.
      """
      from_data_set: SqlDataSet = node.parent_node.accept(self)
  
      from_data_set_alias = self._next_unique_table_alias()
  
      # Get the output_instance_set of the parent_node
      output_instance_set = from_data_set.instance_set
      output_instance_set = output_instance_set.transform(
          ChangeAssociatedColumns(self._column_association_resolver)
      )
  
      # Build the JoinDescriptions to handle the row base filtering on the output_data_set
      inner_join_data_set_alias = self._next_unique_table_alias()
  
      column_equality_descriptions: List[ColumnEqualityDescription] = []
  
      # Build Time Dimension SqlSelectColumn
      time_dimension_column_name = (
          self.column_association_resolver.resolve_spec(
              node.time_dimension_spec
          ).column_name
      )
      join_time_dimension_column_name = (
          self.column_association_resolver.resolve_spec(
              node.time_dimension_spec.with_aggregation_state(
                  AggregationState.COMPLETE
              ),
          ).column_name
      )
      time_dimension_select_column = SqlSelectColumn(
          expr=SqlFunctionExpression.build_expression_from_aggregation_type(
              aggregation_type=node.agg_by_function,
              sql_column_expression=SqlColumnReferenceExpression(
                  SqlColumnReference(
                      table_alias=inner_join_data_set_alias,
                      column_name=time_dimension_column_name,
                  ),
              ),
          ),
          column_alias=join_time_dimension_column_name,
      )
      column_equality_descriptions.append(
          ColumnEqualityDescription(
              left_column_alias=time_dimension_column_name,
              right_column_alias=join_time_dimension_column_name,
          )
      )
  
      # Build optional window grouping SqlSelectColumn
      entity_select_columns: List[SqlSelectColumn] = []
      for entity_spec in node.entity_specs:
          entity_column_name = (
              self.column_association_resolver.resolve_spec(
                  entity_spec
              ).column_name
          )
          entity_select_columns.append(
              SqlSelectColumn(
                  expr=SqlColumnReferenceExpression(
                      SqlColumnReference(
                          table_alias=inner_join_data_set_alias,
                          column_name=entity_column_name,
                      ),
                  ),
                  column_alias=entity_column_name,
              )
          )
          column_equality_descriptions.append(
              ColumnEqualityDescription(
                  left_column_alias=entity_column_name,
                  right_column_alias=entity_column_name,
              )
          )
  
      # Propogate additional group by during query time of the non-additive time dimension
      queried_time_dimension_select_column: Optional[
          SqlSelectColumn
      ] = None
      if node.queried_time_dimension_spec:
          query_time_dimension_column_name = (
              self.column_association_resolver.resolve_spec(
                  node.queried_time_dimension_spec
              ).column_name
          )
          queried_time_dimension_select_column = SqlSelectColumn(
              expr=SqlColumnReferenceExpression(
                  SqlColumnReference(
                      table_alias=inner_join_data_set_alias,
                      column_name=query_time_dimension_column_name,
                  ),
              ),
              column_alias=query_time_dimension_column_name,
          )
  
      row_filter_group_bys = tuple(entity_select_columns)
      if queried_time_dimension_select_column:
          row_filter_group_bys += (
              queried_time_dimension_select_column,
          )
      # Construct SelectNode for Row filtering
      row_filter_sql_select_node = SqlSelectStatementNode(
          description=f"Filter row on {node.agg_by_function.name}({time_dimension_column_name})",
          select_columns=row_filter_group_bys
          + (time_dimension_select_column,),
          from_source=from_data_set.sql_select_node,
          from_source_alias=inner_join_data_set_alias,
          joins_descs=(),
          group_bys=row_filter_group_bys,
          where=None,
          order_bys=(),
      )
  
      join_data_set_alias = self._next_unique_table_alias()
      sql_join_desc = SqlQueryPlanJoinBuilder.make_column_equality_sql_join_description(
          right_source_node=row_filter_sql_select_node,
          left_source_alias=from_data_set_alias,
          right_source_alias=join_data_set_alias,
          column_equality_descriptions=column_equality_descriptions,
          join_type=SqlJoinType.INNER,
      )
      return SqlDataSet(
          instance_set=output_instance_set,
          sql_select_node=SqlSelectStatementNode(
              description=node.description,
              select_columns=output_instance_set.transform(
                  CreateSelectColumnsForInstances(
                      from_data_set_alias,
                      self._column_association_resolver,
                  )
              ).as_tuple(),
              from_source=from_data_set.sql_select_node,
              from_source_alias=from_data_set_alias,
              joins_descs=(sql_join_desc,),
              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 20 for calling the DataflowToSqlQueryPlanConverter obj's underlying member method code with example usage: obj.visit_semi_additive_join_node(node) and returns: SqlDataSet