class_docstr: Visits the SQL query plan to simplify sub-queries. On each visit, return a simplified node.
  
  Unlike SqlSubQueryReducerVisitor, this will re-write expressions to realize more reductions.
docstr: Reduces nodes with joins if the join source is simple to reduce.
  
  Converts this:
  
  SELECT
    SUM(bookings_src.bookings) AS bookings
    listings_src.country_latest AS listing__country_latest
    bookings_src.ds AS ds
  FROM (
    SELECT
      fct_bookings_src.booking AS bookings
      , fct_bookings_src.ds AS ds
      , fct_bookings_src.listing_id AS listing
    FROM demo.fct_bookings fct_bookings_src
  ) bookings_src
  JOIN (
    SELECT
      dim_listings_src.country_latest AS country_latest
      , listing_id AS listing
    FROM demo.dim_listings dim_listings_src
  ) listings_src
  ON bookings_src.listing = listings_src.listing
  GROUP BY bookings_src.ds
  
  to:
  
  SELECT
    SUM(bookings_src.booking) AS bookings
    dim_listings_src.country_latest AS listing__country_latest
    bookings_src.ds AS ds
  FROM demo.fct_bookings fct_bookings_src
  JOIN demo.dim_listings dim_listings_src
  ON bookings_src.listing_id = dim_listings_src.listing_id
  GROUP BY bookings_src.ds
code: def _rewrite_node_with_join(
      node: SqlSelectStatementNode,
  ) -> SqlSelectStatementNode:
      """Reduces nodes with joins if the join source is simple to reduce.
  
      Converts this:
  
      SELECT
        SUM(bookings_src.bookings) AS bookings
        listings_src.country_latest AS listing__country_latest
        bookings_src.ds AS ds
      FROM (
        SELECT
          fct_bookings_src.booking AS bookings
          , fct_bookings_src.ds AS ds
          , fct_bookings_src.listing_id AS listing
        FROM demo.fct_bookings fct_bookings_src
      ) bookings_src
      JOIN (
        SELECT
          dim_listings_src.country_latest AS country_latest
          , listing_id AS listing
        FROM demo.dim_listings dim_listings_src
      ) listings_src
      ON bookings_src.listing = listings_src.listing
      GROUP BY bookings_src.ds
  
      to:
  
      SELECT
        SUM(bookings_src.booking) AS bookings
        dim_listings_src.country_latest AS listing__country_latest
        bookings_src.ds AS ds
      FROM demo.fct_bookings fct_bookings_src
      JOIN demo.dim_listings dim_listings_src
      ON bookings_src.listing_id = dim_listings_src.listing_id
      GROUP BY bookings_src.ds
      """
      # Check that there aren't any duplicates in source aliases, or else there would be a collision when reduced.
      # This check is conservative as it checks for duplicates in this node and parent nodes, but depending on
      # on which sources get reduced, there may not be a collision.
      from_source = node.from_source
      from_source_select = from_source.as_select_node
      from_source_alias = node.from_source_alias
  
      all_source_aliases = [from_source_alias]
      source_alias_set = {from_source_alias}
  
      if from_source_select:
          all_source_aliases.append(
              from_source_select.from_source_alias
          )
          source_alias_set.add(from_source_select.from_source_alias)
  
      for join_desc in node.join_descs:
          all_source_aliases.append(join_desc.right_source_alias)
          source_alias_set.add(join_desc.right_source_alias)
          joined_node_select = join_desc.right_source.as_select_node
          if joined_node_select:
              all_source_aliases.append(
                  joined_node_select.from_source_alias
              )
              source_alias_set.add(joined_node_select.from_source_alias)
  
      # Some duplicate aliases, so safer to not reduce.
      if len(all_source_aliases) != len(source_alias_set):
          return node
  
      # See if any of the joined sources can be reduced.
      new_join_descs = []
  
      clauses_to_rewrite = RewritableSqlClauses(
          select_columns=list(node.select_columns),
          wheres=[node.where] if node.where else [],
          group_bys=list(node.group_bys),
          order_bys=list(node.order_bys),
      )
  
      # Can't re-write ambiguous expressions.
      if clauses_to_rewrite.contains_ambiguous_exprs:
          return node
  
      additional_where_clauses = []
      column_replacements_from_all_joins = []
  
      for join_desc in node.join_descs:
          join_select_node = join_desc.right_source.as_select_node
  
          # Verifying that it's simple makes it easier to reason about the logic.
          if (
              not join_select_node
              or not SqlRewritingSubQueryReducerVisitor._is_simple_source(
                  join_select_node
              )
          ):
              new_join_descs.append(join_desc)
              continue
  
          column_replacements = SqlRewritingSubQueryReducerVisitor._get_column_replacements(
              parent_node=join_select_node,
              parent_node_alias=join_desc.right_source_alias,
          )
          column_replacements_from_all_joins.append(column_replacements)
  
          new_join_descs.append(
              SqlJoinDescription(
                  right_source=join_select_node.from_source,
                  right_source_alias=join_select_node.from_source_alias,
                  on_condition=join_desc.on_condition.rewrite(
                      column_replacements
                  )
                  if join_desc.on_condition
                  else None,
                  join_type=join_desc.join_type,
              )
          )
  
          if join_select_node.where:
              additional_where_clauses.append(join_select_node.where)
  
          clauses_to_rewrite.rewrite(column_replacements)
  
      # While the above re-wrote each individual join when that join was reduced, the ON condition could reference
      # columns from other joins, so the column replacements have to be applied to all join ON conditions.
      for column_replacements in column_replacements_from_all_joins:
          new_join_descs = [
              SqlJoinDescription(
                  right_source=x.right_source,
                  right_source_alias=x.right_source_alias,
                  on_condition=x.on_condition.rewrite(
                      column_replacements=column_replacements
                  )
                  if x.on_condition
                  else None,
                  join_type=x.join_type,
              )
              for x in new_join_descs
          ]
  
      from_source_is_simple = (
          SqlRewritingSubQueryReducerVisitor._is_simple_source(
              from_source_select
          )
          if from_source_select
          else False
      )
      if from_source_select and from_source_is_simple:
          column_replacements = SqlRewritingSubQueryReducerVisitor._get_column_replacements(
              parent_node=from_source_select,
              parent_node_alias=node.from_source_alias,
          )
  
          if from_source_select.where:
              additional_where_clauses.append(from_source_select.where)
  
          clauses_to_rewrite.rewrite(
              column_replacements=column_replacements
          )
          # This was already checked in _is_simple_source().
          assert len(from_source_select.parent_nodes) == 1
          from_source = from_source_select.from_source
          from_source_alias = from_source_select.from_source_alias
  
          new_join_descs = [
              SqlJoinDescription(
                  right_source=x.right_source,
                  right_source_alias=x.right_source_alias,
                  on_condition=x.on_condition.rewrite(
                      column_replacements=column_replacements
                  )
                  if x.on_condition
                  else None,
                  join_type=x.join_type,
              )
              for x in new_join_descs
          ]
  
      return SqlSelectStatementNode(
          description=node.description,
          select_columns=tuple(clauses_to_rewrite.select_columns),
          from_source=from_source,
          from_source_alias=from_source_alias,
          joins_descs=tuple(new_join_descs),
          group_bys=tuple(clauses_to_rewrite.group_bys),
          order_bys=tuple(clauses_to_rewrite.order_bys),
          where=clauses_to_rewrite.combine_wheres(
              additional_where_clauses
          ),
          limit=node.limit,
          distinct=node.distinct,
      )
  
imports: from __future__ import annotations
  import logging
  from dataclasses import dataclass
  from typing import List, Optional, Sequence, Tuple
  from metricflow.sql.optimizer.sql_query_plan_optimizer import SqlQueryPlanOptimizer
  from metricflow.sql.sql_exprs import SqlColumnAliasReferenceExpression, SqlColumnReference, SqlColumnReplacements, SqlExpressionNode, SqlExpressionTreeLineage, SqlLogicalExpression, SqlLogicalOperator
  from metricflow.sql.sql_plan import SqlJoinDescription, SqlOrderByDescription, SqlQueryPlanNode, SqlQueryPlanNodeVisitor, SqlSelectColumn, SqlSelectQueryFromClauseNode, SqlSelectStatementNode, SqlTableFromClauseNode
label_desc: Use image node_id 13 for calling the SqlRewritingSubQueryReducerVisitor obj's underlying member method code with example usage: obj._rewrite_node_with_join(node) and returns: SqlSelectStatementNode, node, node