class_docstr: The implementation of InternalWindowProcessFunction for MergingWindowAssigner.
class_docstr_tok: ["The","implementation","of","InternalWindowProcessFunction","for","MergingWindowAssigner","."]
code: def _add_window(self, new_window: W):
      collector = MergeResultCollector()
      self._window_assigner.merge_windows(
          new_window, self._sorted_windows, collector
      )
  
      result_window = new_window
      is_new_window_merged = False
  
      # perform the merge
      merge_results = collector.merge_results
      for merge_result in merge_results:
          merge_windows = merge_results[merge_result]  # type: Set[W]
  
          # if our new window is in the merged windows make the merge result the result window
          try:
              merge_windows.remove(new_window)
              is_new_window_merged = True
              result_window = merge_result
          except KeyError:
              pass
  
          # if our new window is the same as a pre-existing window, nothing to do
          if not merge_windows:
              continue
  
          # pick any of the merged windows and choose that window's state window
          # as the state window for the merge result
          merged_state_namespace = self._window_mapping.get(
              iter(merge_windows).__next__()
          )
          merged_state_windows = []
  
          # figure out the state windows that we are merging
          for merged_window in merge_windows:
              res = self._window_mapping.get(merged_window)
              if res is not None:
                  self._window_mapping.remove(merged_window)
                  self._sorted_windows.remove(merged_window)
                  # don't put the target state window into the merged windows
                  if res != merged_state_namespace:
                      merged_state_windows.append(res)
  
          self._window_mapping.put(merge_result, merged_state_namespace)
          self._sorted_windows.append(merge_result)
          self._sorted_windows.sort()
  
          # don't merge the new window itself, it never had any state associated with it
          # i.e. if we are only merging one pre-existing window into itself
          # without extending the pre-existing window
          if not (
              len(merge_windows) == 1 and merge_result in merge_windows
          ):
              self._merge(
                  merge_result,
                  merge_windows,
                  merged_state_namespace,
                  merged_state_windows,
              )
  
      # the new window created a new, self-contained window without merging
      if (
          len(merge_results) == 0
          or result_window == new_window
          and not is_new_window_merged
      ):
          self._window_mapping.put(result_window, result_window)
          self._sorted_windows.append(result_window)
          self._sorted_windows.sort()
  
      return result_window
  
imports: from abc import abstractmethod, ABC
  from typing import Generic, List, Iterable, Dict, Set
  from pyflink.common import Row
  from pyflink.common.constants import MAX_LONG_VALUE
  from pyflink.datastream.state import MapState
  from pyflink.fn_execution.table.window_assigner import WindowAssigner, PanedWindowAssigner, MergingWindowAssigner
  from pyflink.fn_execution.table.window_context import Context, K, W
label_desc: Use image node_id 8 for calling the MergingWindowProcessFunction obj's underlying member method code with example usage: obj._add_window(new_window) and returns: result_window