code: def align_diff_frames(
      resolve_func: Callable[
          ["DataFrame", List[Label], List[Label]],
          Iterator[Tuple["Series", Label]],
      ],
      this: "DataFrame",
      that: "DataFrame",
      fillna: bool = True,
      how: str = "full",
      preserve_order_column: bool = False,
  ) -> "DataFrame":
      """
      This method aligns two different DataFrames with a given `func`. Columns are resolved and
      handled within the given `func`.
      To use this, `compute.ops_on_diff_frames` should be True, for now.
  
      :param resolve_func: Takes aligned (joined) DataFrame, the column of the current DataFrame, and
          the column of another DataFrame. It returns an iterable that produces Series.
  
          >>> from pyspark.pandas.config import set_option, reset_option
          >>>
          >>> set_option("compute.ops_on_diff_frames", True)
          >>>
          >>> psdf1 = ps.DataFrame({'a': [9, 8, 7, 6, 5, 4, 3, 2, 1]})
          >>> psdf2 = ps.DataFrame({'a': [9, 8, 7, 6, 5, 4, 3, 2, 1]})
          >>>
          >>> def func(psdf, this_column_labels, that_column_labels):
          ...    psdf  # conceptually this is A + B.
          ...
          ...    # Within this function, Series from A or B can be performed against `psdf`.
          ...    this_label = this_column_labels[0]  # this is ('a',) from psdf1.
          ...    that_label = that_column_labels[0]  # this is ('a',) from psdf2.
          ...    new_series = (psdf[this_label] - psdf[that_label]).rename(str(this_label))
          ...
          ...    # This new series will be placed in new DataFrame.
          ...    yield (new_series, this_label)
          >>>
          >>>
          >>> align_diff_frames(func, psdf1, psdf2).sort_index()
             a
          0  0
          1  0
          2  0
          3  0
          4  0
          5  0
          6  0
          7  0
          8  0
          >>> reset_option("compute.ops_on_diff_frames")
  
      :param this: a DataFrame to align
      :param that: another DataFrame to align
      :param fillna: If True, it fills missing values in non-common columns in both `this` and `that`.
          Otherwise, it returns as are.
      :param how: join way. In addition, it affects how `resolve_func` resolves the column conflict.
          - full: `resolve_func` should resolve only common columns from 'this' and 'that' DataFrames.
              For instance, if 'this' has columns A, B, C and that has B, C, D, `this_columns` and
              'that_columns' in this function are B, C and B, C.
          - left: `resolve_func` should resolve columns including `that` column.
              For instance, if 'this' has columns A, B, C and that has B, C, D, `this_columns` is
              B, C but `that_columns` are B, C, D.
          - inner: Same as 'full' mode; however, internally performs inner join instead.
      :return: Aligned DataFrame
      """
      from pyspark.pandas.frame import DataFrame
  
      assert how == "full" or how == "left" or how == "inner"
  
      this_column_labels = this._internal.column_labels
      that_column_labels = that._internal.column_labels
      common_column_labels = set(this_column_labels).intersection(
          that_column_labels
      )
  
      # 1. Perform the join given two dataframes.
      combined = combine_frames(
          this,
          that,
          how=how,
          preserve_order_column=preserve_order_column,
      )
  
      # 2. Apply the given function to transform the columns in a batch and keep the new columns.
      combined_column_labels = combined._internal.column_labels
  
      that_columns_to_apply: List[Label] = []
      this_columns_to_apply: List[Label] = []
      additional_that_columns: List[Label] = []
      if is_remote():
          from pyspark.sql.connect.column import Column as ConnectColumn
  
          Column = ConnectColumn
      columns_to_keep: List[Union[Series, Column]] = []  # type: ignore[valid-type]
      column_labels_to_keep: List[Label] = []
  
      for combined_label in combined_column_labels:
          for common_label in common_column_labels:
              if combined_label == tuple(["this", *common_label]):
                  this_columns_to_apply.append(combined_label)
                  break
              elif combined_label == tuple(["that", *common_label]):
                  that_columns_to_apply.append(combined_label)
                  break
          else:
              if how == "left" and combined_label in [
                  tuple(["that", *label])
                  for label in that_column_labels
              ]:
                  # In this case, we will drop `that_columns` in `columns_to_keep` but passes
                  # it later to `func`. `func` should resolve it.
                  # Note that adding this into a separate list (`additional_that_columns`)
                  # is intentional so that `this_columns` and `that_columns` can be paired.
                  additional_that_columns.append(combined_label)
              elif fillna:
                  columns_to_keep.append(
                      F.lit(None)
                      .cast(DoubleType())
                      .alias(str(combined_label))
                  )
                  column_labels_to_keep.append(combined_label)
              else:
                  columns_to_keep.append(
                      combined._psser_for(combined_label)
                  )
                  column_labels_to_keep.append(combined_label)
  
      that_columns_to_apply += additional_that_columns
  
      # Should extract columns to apply and do it in a batch in case
      # it adds new columns for example.
      columns_applied: List[Union[Series, Column]]  # type: ignore[valid-type]
      column_labels_applied: List[Label]
      if (
          len(this_columns_to_apply) > 0
          or len(that_columns_to_apply) > 0
      ):
          psser_set, column_labels_set = zip(
              *resolve_func(
                  combined, this_columns_to_apply, that_columns_to_apply
              )
          )
          columns_applied = list(psser_set)
          column_labels_applied = list(column_labels_set)
      else:
          columns_applied = []
          column_labels_applied = []
  
      applied: DataFrame = DataFrame(
          combined._internal.with_new_columns(
              columns_applied + columns_to_keep,
              column_labels=column_labels_applied
              + column_labels_to_keep,
          )
      )
  
      # 3. Restore the names back and deduplicate columns.
      this_labels: Dict[Label, Label] = {}
      # Add columns in an order of its original frame.
      for this_label in this_column_labels:
          for new_label in applied._internal.column_labels:
              if (
                  new_label[1:] not in this_labels
                  and this_label == new_label[1:]
              ):
                  this_labels[new_label[1:]] = new_label
  
      # After that, we will add the rest columns.
      other_labels: Dict[Label, Label] = {}
      for new_label in applied._internal.column_labels:
          if new_label[1:] not in this_labels:
              other_labels[new_label[1:]] = new_label
  
      psdf = applied[
          list(this_labels.values()) + list(other_labels.values())
      ]
      psdf.columns = psdf.columns.droplevel()
      return psdf
  
imports: import functools
  from contextlib import contextmanager
  import os
  from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, Union, TYPE_CHECKING, cast, no_type_check, overload
  import warnings
  import pandas
  from pandas.api.types import is_list_like
  from pyspark.sql import functions, Column, DataFrame, SparkSession
  from pyspark.sql.types import DoubleType
  from pyspark.sql.utils import is_remote, get_dataframe_class
  from pyspark.errors import PySparkTypeError
  from pyspark import pandas
  from pyspark.pandas._typing import Axis, Label, Name, DataFrameOrSeries
  from pyspark.pandas.typedef.typehints import as_spark_type
label_desc: Use image node_id 3 for calling a global function with example usage: align_diff_frames(resolve_func, this, that, fillna, how, preserve_order_column) and returns: psdf