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: Builds a resulting data set with all valid conversion events.
  
  This node takes the conversion and base data set and joins them against an entity and
  a valid time range to get successful conversions. It then deduplicates opportunities
  via the window function `first_value` to take the closest opportunity to the
  corresponding conversion. Then it returns a data set with each row representing a
  successful conversion. Duplication may exist in the result due to a single base event
  being able to link to multiple conversion events.
docstr_tok: ["Builds","a","resulting","data","set","with","all","valid","conversion","events",".","This","node","takes","the","conversion","and","base","data","set","and","joins","them","against","an","entity","and","a","valid","time","range","to","get","successful","conversions",".","It","then","deduplicates","opportunities","via","the","window","function","`","first_value","`","to","take","the","closest","opportunity","to","the","corresponding","conversion",".","Then","it","returns","a","data","set","with","each","row","representing","a","successful","conversion",".","Duplication","may","exist","in","the","result","due","to","a","single","base","event","being","able","to","link","to","multiple","conversion","events","."]
code: def visit_join_conversion_events_node(
      self, node: JoinConversionEventsNode
  ) -> SqlDataSet:
      """Builds a resulting data set with all valid conversion events.
  
      This node takes the conversion and base data set and joins them against an entity and
      a valid time range to get successful conversions. It then deduplicates opportunities
      via the window function `first_value` to take the closest opportunity to the
      corresponding conversion. Then it returns a data set with each row representing a
      successful conversion. Duplication may exist in the result due to a single base event
      being able to link to multiple conversion events.
      """
      base_data_set: SqlDataSet = node.base_node.accept(self)
      base_data_set_alias = self._next_unique_table_alias()
  
      conversion_data_set: SqlDataSet = node.conversion_node.accept(
          self
      )
      conversion_data_set_alias = self._next_unique_table_alias()
  
      base_time_dimension_column_name = (
          self._column_association_resolver.resolve_spec(
              node.base_time_dimension_spec
          ).column_name
      )
      conversion_time_dimension_column_name = (
          self._column_association_resolver.resolve_spec(
              node.conversion_time_dimension_spec
          ).column_name
      )
      entity_column_name = (
          self._column_association_resolver.resolve_spec(
              node.entity_spec
          ).column_name
      )
  
      constant_property_column_names: List[Tuple[str, str]] = []
      for constant_property in node.constant_properties or []:
          base_property_col_name = (
              self._column_association_resolver.resolve_spec(
                  constant_property.base_spec
              ).column_name
          )
          conversion_property_col_name = (
              self._column_association_resolver.resolve_spec(
                  constant_property.conversion_spec
              ).column_name
          )
          constant_property_column_names.append(
              (base_property_col_name, conversion_property_col_name)
          )
  
      # Builds the join conditions that is required for a successful conversion
      sql_join_description = SqlQueryPlanJoinBuilder.make_join_conversion_join_description(
          node=node,
          base_data_set=AnnotatedSqlDataSet(
              data_set=base_data_set,
              alias=base_data_set_alias,
              _metric_time_column_name=base_time_dimension_column_name,
          ),
          conversion_data_set=AnnotatedSqlDataSet(
              data_set=conversion_data_set,
              alias=conversion_data_set_alias,
              _metric_time_column_name=conversion_time_dimension_column_name,
          ),
          column_equality_descriptions=(
              ColumnEqualityDescription(
                  left_column_alias=entity_column_name,
                  right_column_alias=entity_column_name,
              ),
          )
          + tuple(
              ColumnEqualityDescription(
                  left_column_alias=base_col,
                  right_column_alias=conversion_col,
              )
              for base_col, conversion_col in constant_property_column_names
          ),
      )
  
      # Builds the first_value window function columns
      base_sql_column_references = base_data_set.instance_set.transform(
          CreateSqlColumnReferencesForInstances(
              base_data_set_alias, self._column_association_resolver
          )
      )
  
      unique_conversion_col_names = tuple(
          self._column_association_resolver.resolve_spec(
              spec
          ).column_name
          for spec in node.unique_identifier_keys
      )
      partition_by_columns: Tuple[str, ...] = (
          entity_column_name,
          conversion_time_dimension_column_name,
      ) + unique_conversion_col_names
      if node.constant_properties:
          partition_by_columns += tuple(
              conversion_column_name
              for _, conversion_column_name in constant_property_column_names
          )
      base_sql_select_columns = tuple(
          SqlSelectColumn(
              expr=SqlWindowFunctionExpression(
                  sql_function=SqlWindowFunction.FIRST_VALUE,
                  sql_function_args=[
                      SqlColumnReferenceExpression(
                          SqlColumnReference(
                              table_alias=base_data_set_alias,
                              column_name=base_sql_column_reference.col_ref.column_name,
                          ),
                      )
                  ],
                  partition_by_args=[
                      SqlColumnReferenceExpression(
                          SqlColumnReference(
                              table_alias=conversion_data_set_alias,
                              column_name=column,
                          ),
                      )
                      for column in partition_by_columns
                  ],
                  order_by_args=[
                      SqlWindowOrderByArgument(
                          expr=SqlColumnReferenceExpression(
                              SqlColumnReference(
                                  table_alias=base_data_set_alias,
                                  column_name=base_time_dimension_column_name,
                              ),
                          ),
                          descending=True,
                      )
                  ],
              ),
              column_alias=base_sql_column_reference.col_ref.column_name,
          )
          for base_sql_column_reference in base_sql_column_references
      )
  
      conversion_data_set_output_instance_set = (
          conversion_data_set.instance_set.transform(
              FilterElements(
                  include_specs=InstanceSpecSet(
                      measure_specs=(node.conversion_measure_spec,)
                  )
              )
          )
      )
  
      # Deduplicate the fanout results
      conversion_unique_key_select_columns = tuple(
          SqlSelectColumn(
              expr=SqlColumnReferenceExpression(
                  SqlColumnReference(
                      table_alias=conversion_data_set_alias,
                      column_name=column_name,
                  ),
              ),
              column_alias=column_name,
          )
          for column_name in unique_conversion_col_names
      )
      additional_conversion_select_columns = (
          conversion_data_set_output_instance_set.transform(
              CreateSelectColumnsForInstances(
                  conversion_data_set_alias,
                  self._column_association_resolver,
              )
          ).as_tuple()
      )
      deduped_sql_select_node = SqlSelectStatementNode(
          description=f"Dedupe the fanout with {','.join(spec.qualified_name for spec in node.unique_identifier_keys)} in the conversion data set",
          select_columns=base_sql_select_columns
          + conversion_unique_key_select_columns
          + additional_conversion_select_columns,
          from_source=base_data_set.sql_select_node,
          from_source_alias=base_data_set_alias,
          joins_descs=(sql_join_description,),
          group_bys=(),
          where=None,
          order_bys=(),
          distinct=True,
      )
  
      # Returns the original dataset with all the successful conversion
      output_data_set_alias = self._next_unique_table_alias()
      output_instance_set = ChangeAssociatedColumns(
          self._column_association_resolver
      ).transform(
          InstanceSet.merge(
              [
                  conversion_data_set_output_instance_set,
                  base_data_set.instance_set,
              ]
          )
      )
      return SqlDataSet(
          instance_set=output_instance_set,
          sql_select_node=SqlSelectStatementNode(
              description=node.description,
              select_columns=output_instance_set.transform(
                  CreateSelectColumnsForInstances(
                      output_data_set_alias,
                      self._column_association_resolver,
                  )
              ).as_tuple(),
              from_source=deduped_sql_select_node,
              from_source_alias=output_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 24 for calling the DataflowToSqlQueryPlanConverter obj's underlying member method code with example usage: obj.visit_join_conversion_events_node(node) and returns: SqlDataSet