code: ################################################################################
  #  Licensed to the Apache Software Foundation (ASF) under one
  #  or more contributor license agreements.  See the NOTICE file
  #  distributed with this work for additional information
  #  regarding copyright ownership.  The ASF licenses this file
  #  to you under the Apache License, Version 2.0 (the
  #  "License"); you may not use this file except in compliance
  #  with the License.  You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #  See the License for the specific language governing permissions and
  # limitations under the License.
  ################################################################################
  import abc
  import time
  from functools import reduce
  from itertools import chain
  from typing import List, Tuple, Any, Dict
  
  from apache_beam.coders import PickleCoder
  
  from pyflink.datastream.state import ValueStateDescriptor, ValueState, ListStateDescriptor, \
      ListState, MapStateDescriptor, MapState, ReducingStateDescriptor, ReducingState
  from pyflink.datastream import TimeDomain
  from pyflink.datastream.functions import RuntimeContext, TimerService, ProcessFunction, \
      KeyedProcessFunction
  from pyflink.fn_execution import flink_fn_execution_pb2, operation_utils
  from pyflink.fn_execution.aggregate import extract_data_view_specs
  from pyflink.fn_execution.beam.beam_coders import DataViewFilterCoder
  from pyflink.fn_execution.operation_utils import extract_user_defined_aggregate_function
  from pyflink.fn_execution.state_impl import RemoteKeyedStateBackend
  
  try:
      from pyflink.fn_execution.aggregate_fast import RowKeySelector, SimpleAggsHandleFunction, \
          GroupAggFunction, DistinctViewDescriptor, SimpleTableAggsHandleFunction, \
          GroupTableAggFunction
  except ImportError:
      from pyflink.fn_execution.aggregate_slow import RowKeySelector, SimpleAggsHandleFunction, \
          GroupAggFunction, DistinctViewDescriptor, SimpleTableAggsHandleFunction,\
          GroupTableAggFunction
  
  from pyflink.metrics.metricbase import GenericMetricGroup
  from pyflink.table import FunctionContext, Row
  
  
  # table operations
  SCALAR_FUNCTION_URN = "flink:transform:scalar_function:v1"
  TABLE_FUNCTION_URN = "flink:transform:table_function:v1"
  STREAM_GROUP_AGGREGATE_URN = "flink:transform:stream_group_aggregate:v1"
  STREAM_GROUP_TABLE_AGGREGATE_URN = "flink:transform:stream_group_table_aggregate:v1"
  PANDAS_AGGREGATE_FUNCTION_URN = "flink:transform:aggregate_function:arrow:v1"
  PANDAS_BATCH_OVER_WINDOW_AGGREGATE_FUNCTION_URN = \
      "flink:transform:batch_over_window_aggregate_function:arrow:v1"
  
  # datastream operations
  DATA_STREAM_STATELESS_FUNCTION_URN = "flink:transform:datastream_stateless_function:v1"
  PROCESS_FUNCTION_URN = "flink:transform:process_function:v1"
  KEYED_PROCESS_FUNCTION_URN = "flink:transform:keyed_process_function:v1"
  
  
  class Operation(abc.ABC):
      def __init__(self, spec):
          super(Operation, self).__init__()
          self.spec = spec
          self.func, self.user_defined_funcs = self.generate_func(self.spec.serialized_fn)
          if self.spec.serialized_fn.metric_enabled:
              self.base_metric_group = GenericMetricGroup(None, None)
          else:
              self.base_metric_group = None
  
      def open(self):
          for user_defined_func in self.user_defined_funcs:
              if hasattr(user_defined_func, 'open'):
                  user_defined_func.open(FunctionContext(self.base_metric_group))
  
      def close(self):
          for user_defined_func in self.user_defined_funcs:
              if hasattr(user_defined_func, 'close'):
                  user_defined_func.close()
  
      def finish(self):
          self._update_gauge(self.base_metric_group)
  
      def _update_gauge(self, base_metric_group):
          if base_metric_group is not None:
              for name in base_metric_group._flink_gauge:
                  flink_gauge = base_metric_group._flink_gauge[name]
                  beam_gauge = base_metric_group._beam_gauge[name]
                  beam_gauge.set(flink_gauge())
              for sub_group in base_metric_group._sub_groups:
                  self._update_gauge(sub_group)
  
      @abc.abstractmethod
      def generate_func(self, serialized_fn) -> Tuple:
          pass
  
  
  class ScalarFunctionOperation(Operation):
      def __init__(self, spec):
          super(ScalarFunctionOperation, self).__init__(spec)
  
      def generate_func(self, serialized_fn):
          """
          Generates a lambda function based on udfs.
          :param serialized_fn: serialized function which contains a list of the proto
                                representation of the Python :class:`ScalarFunction`
          :return: the generated lambda function
          """
          scalar_functions, variable_dict, user_defined_funcs = reduce(
              lambda x, y: (
                  ','.join([x[0], y[0]]),
                  dict(chain(x[1].items(), y[1].items())),
                  x[2] + y[2]),
              [operation_utils.extract_user_defined_function(udf) for udf in serialized_fn.udfs])
          generate_func = eval('lambda value: [%s]' % scalar_functions, variable_dict)
          return generate_func, user_defined_funcs
  
  
  class TableFunctionOperation(Operation):
      def __init__(self, spec):
          super(TableFunctionOperation, self).__init__(spec)
  
      def generate_func(self, serialized_fn):
          """
          Generates a lambda function based on udtfs.
          :param serialized_fn: serialized function which contains the proto representation of
                                the Python :class:`TableFunction`
          :return: the generated lambda function
          """
          table_function, variable_dict, user_defined_funcs = \
              operation_utils.extract_user_defined_function(serialized_fn.udfs[0])
          generate_func = eval('lambda value: %s' % table_function, variable_dict)
          return generate_func, user_defined_funcs
  
  
  class PandasAggregateFunctionOperation(Operation):
      def __init__(self, spec):
          super(PandasAggregateFunctionOperation, self).__init__(spec)
  
      def generate_func(self, serialized_fn):
          pandas_functions, variable_dict, user_defined_funcs = reduce(
              lambda x, y: (
                  ','.join([x[0], y[0]]),
                  dict(chain(x[1].items(), y[1].items())),
                  x[2] + y[2]),
              [operation_utils.extract_user_defined_function(udf, True)
               for udf in serialized_fn.udfs])
          variable_dict['wrap_pandas_result'] = operation_utils.wrap_pandas_result
          generate_func = eval('lambda value: wrap_pandas_result([%s])' %
                               pandas_functions, variable_dict)
          return generate_func, user_defined_funcs
  
  
  class PandasBatchOverWindowAggregateFunctionOperation(Operation):
      def __init__(self, spec):
          super(PandasBatchOverWindowAggregateFunctionOperation, self).__init__(spec)
          self.windows = [window for window in self.spec.serialized_fn.windows]
          # the index among all the bounded range over window
          self.bounded_range_window_index = [-1 for _ in range(len(self.windows))]
          # Whether the specified position window is a bounded range window.
          self.is_bounded_range_window = []
          window_types = flink_fn_execution_pb2.OverWindow
  
          bounded_range_window_nums = 0
          for i, window in enumerate(self.windows):
              window_type = window.window_type
              if (window_type is window_types.RANGE_UNBOUNDED_PRECEDING) or (
                      window_type is window_types.RANGE_UNBOUNDED_FOLLOWING) or (
                      window_type is window_types.RANGE_SLIDING):
                  self.bounded_range_window_index[i] = bounded_range_window_nums
                  self.is_bounded_range_window.append(True)
                  bounded_range_window_nums += 1
              else:
                  self.is_bounded_range_window.append(False)
  
      def generate_func(self, serialized_fn):
          user_defined_funcs = []
          self.window_indexes = []
          self.mapper = []
          for udf in serialized_fn.udfs:
              pandas_agg_function, variable_dict, user_defined_func, window_index = \
                  operation_utils.extract_over_window_user_defined_function(udf)
              user_defined_funcs.extend(user_defined_func)
              self.window_indexes.append(window_index)
              self.mapper.append(eval('lambda value: %s' % pandas_agg_function, variable_dict))
          return self.wrapped_over_window_function, user_defined_funcs
  
      def wrapped_over_window_function(self, boundaries_series):
          import pandas as pd
          OverWindow = flink_fn_execution_pb2.OverWindow
          input_series = boundaries_series[-1]
          # the row number of the arrow format data
          input_cnt = len(input_series[0])
          results = []
          # loop every agg func
          for i in range(len(self.window_indexes)):
              window_index = self.window_indexes[i]
              # the over window which the agg function belongs to
              window = self.windows[window_index]
              window_type = window.window_type
              func = self.mapper[i]
              result = []
              if self.is_bounded_range_window[window_index]:
                  window_boundaries = boundaries_series[
                      self.bounded_range_window_index[window_index]]
                  if window_type is OverWindow.RANGE_UNBOUNDED_PRECEDING:
                      # range unbounded preceding window
                      for j in range(input_cnt):
                          end = window_boundaries[j]
                          series_slices = [s.iloc[:end] for s in input_series]
                          result.append(func(series_slices))
                  elif window_type is OverWindow.RANGE_UNBOUNDED_FOLLOWING:
                      # range unbounded following window
                      for j in range(input_cnt):
                          start = window_boundaries[j]
                          series_slices = [s.iloc[start:] for s in input_series]
                          result.append(func(series_slices))
                  else:
                      # range sliding window
                      for j in range(input_cnt):
                          start = window_boundaries[j * 2]
                          end = window_boundaries[j * 2 + 1]
                          series_slices = [s.iloc[start:end] for s in input_series]
                          result.append(func(series_slices))
              else:
                  # unbounded range window or unbounded row window
                  if (window_type is OverWindow.RANGE_UNBOUNDED) or (
                          window_type is OverWindow.ROW_UNBOUNDED):
                      series_slices = [s.iloc[:] for s in input_series]
                      func_result = func(series_slices)
                      result = [func_result for _ in range(input_cnt)]
                  elif window_type is OverWindow.ROW_UNBOUNDED_PRECEDING:
                      # row unbounded preceding window
                      window_end = window.upper_boundary
                      for j in range(input_cnt):
                          end = min(j + window_end + 1, input_cnt)
                          series_slices = [s.iloc[: end] for s in input_series]
                          result.append(func(series_slices))
                  elif window_type is OverWindow.ROW_UNBOUNDED_FOLLOWING:
                      # row unbounded following window
                      window_start = window.lower_boundary
                      for j in range(input_cnt):
                          start = max(j + window_start, 0)
                          series_slices = [s.iloc[start: input_cnt] for s in input_series]
                          result.append(func(series_slices))
                  else:
                      # row sliding window
                      window_start = window.lower_boundary
                      window_end = window.upper_boundary
                      for j in range(input_cnt):
                          start = max(j + window_start, 0)
                          end = min(j + window_end + 1, input_cnt)
                          series_slices = [s.iloc[start: end] for s in input_series]
                          result.append(func(series_slices))
              results.append(pd.Series(result))
          return results
  
  
  class StatefulFunctionOperation(Operation):
  
      def __init__(self, spec, keyed_state_backend):
          self.keyed_state_backend = keyed_state_backend
          super(StatefulFunctionOperation, self).__init__(spec)
  
      def finish(self):
          super().finish()
          if self.keyed_state_backend:
              self.keyed_state_backend.commit()
  
  
  TRIGGER_TIMER = 1
  
  
  class AbstractStreamGroupAggregateOperation(StatefulFunctionOperation):
  
      def __init__(self, spec, keyed_state_backend):
          self.generate_update_before = spec.serialized_fn.generate_update_before
          self.grouping = [i for i in spec.serialized_fn.grouping]
          self.group_agg_function = None
          # If the upstream generates retract message, we need to add an additional count1() agg
          # to track current accumulated messages count. If all the messages are retracted, we need
          # to send a DELETE message to downstream.
          self.index_of_count_star = spec.serialized_fn.index_of_count_star
          self.count_star_inserted = spec.serialized_fn.count_star_inserted
          self.state_cache_size = spec.serialized_fn.state_cache_size
          self.state_cleaning_enabled = spec.serialized_fn.state_cleaning_enabled
          self.data_view_specs = extract_data_view_specs(spec.serialized_fn.udfs)
          super(AbstractStreamGroupAggregateOperation, self).__init__(spec, keyed_state_backend)
  
      def open(self):
          self.group_agg_function.open(FunctionContext(self.base_metric_group))
  
      def close(self):
          self.group_agg_function.close()
  
      def generate_func(self, serialized_fn):
          user_defined_aggs = []
          input_extractors = []
          filter_args = []
          # stores the indexes of the distinct views which the agg functions used
          distinct_indexes = []
          # stores the indexes of the functions which share the same distinct view
          # and the filter args of them
          distinct_info_dict = {}
          for i in range(len(serialized_fn.udfs)):
              user_defined_agg, input_extractor, filter_arg, distinct_index = \
                  extract_user_defined_aggregate_function(
                      i, serialized_fn.udfs[i], distinct_info_dict)
              user_defined_aggs.append(user_defined_agg)
              input_extractors.append(input_extractor)
              filter_args.append(filter_arg)
              distinct_indexes.append(distinct_index)
          distinct_view_descriptors = {}
          for agg_index_list, filter_arg_list in distinct_info_dict.values():
              if -1 in filter_arg_list:
                  # If there is a non-filter call, we don't need to check filter or not before
                  # writing the distinct data view.
                  filter_arg_list = []
              # use the agg index of the first function as the key of shared distinct view
              distinct_view_descriptors[agg_index_list[0]] = DistinctViewDescriptor(
                  input_extractors[agg_index_list[0]], filter_arg_list)
  
          key_selector = RowKeySelector(self.grouping)
          if len(self.data_view_specs) > 0:
              state_value_coder = DataViewFilterCoder(self.data_view_specs)
          else:
              state_value_coder = PickleCoder()
  
          self.group_agg_function = self.create_process_function(
              user_defined_aggs, input_extractors, filter_args, distinct_indexes,
              distinct_view_descriptors, key_selector, state_value_coder)
  
          return self.process_element_or_timer, []
  
      def process_element_or_timer(self, input_datas: List[Tuple[int, Row, int, Row]]):
          # the structure of the input data:
          # [element_type, element(for process_element), timestamp(for timer), key(for timer)]
          # all the fields are nullable except the "element_type"
          for input_data in input_datas:
              if input_data[0] != TRIGGER_TIMER:
                  self.group_agg_function.process_element(input_data[1])
              else:
                  self.group_agg_function.on_timer(input_data[3])
          return self.group_agg_function.finish_bundle()
  
      @abc.abstractmethod
      def create_process_function(self, user_defined_aggs, input_extractors, filter_args,
                                  distinct_indexes, distinct_view_descriptors, key_selector,
                                  state_value_coder):
          pass
  
  
  class StreamGroupAggregateOperation(AbstractStreamGroupAggregateOperation):
  
      def __init__(self, spec, keyed_state_backend):
          super(StreamGroupAggregateOperation, self).__init__(spec, keyed_state_backend)
  
      def create_process_function(self, user_defined_aggs, input_extractors, filter_args,
                                  distinct_indexes, distinct_view_descriptors, key_selector,
                                  state_value_coder):
          aggs_handler_function = SimpleAggsHandleFunction(
              user_defined_aggs,
              input_extractors,
              self.index_of_count_star,
              self.count_star_inserted,
              self.data_view_specs,
              filter_args,
              distinct_indexes,
              distinct_view_descriptors)
  
          return GroupAggFunction(
              aggs_handler_function,
              key_selector,
              self.keyed_state_backend,
              state_value_coder,
              self.generate_update_before,
              self.state_cleaning_enabled,
              self.index_of_count_star)
  
  
  class StreamGroupTableAggregateOperation(AbstractStreamGroupAggregateOperation):
      def __init__(self, spec, keyed_state_backend):
          super(StreamGroupTableAggregateOperation, self).__init__(spec, keyed_state_backend)
  
      def create_process_function(self, user_defined_aggs, input_extractors, filter_args,
                                  distinct_indexes, distinct_view_descriptors, key_selector,
                                  state_value_coder):
          aggs_handler_function = SimpleTableAggsHandleFunction(
              user_defined_aggs,
              input_extractors,
              self.data_view_specs,
              filter_args,
              distinct_indexes,
              distinct_view_descriptors)
          return GroupTableAggFunction(
              aggs_handler_function,
              key_selector,
              self.keyed_state_backend,
              state_value_coder,
              self.generate_update_before,
              self.state_cleaning_enabled,
              self.index_of_count_star)
  
  
  class DataStreamStatelessFunctionOperation(Operation):
  
      def __init__(self, spec):
          super(DataStreamStatelessFunctionOperation, self).__init__(spec)
  
      def open(self):
          for user_defined_func in self.user_defined_funcs:
              if hasattr(user_defined_func, 'open'):
                  runtime_context = RuntimeContext(
                      self.spec.serialized_fn.runtime_context.task_name,
                      self.spec.serialized_fn.runtime_context.task_name_with_subtasks,
                      self.spec.serialized_fn.runtime_context.number_of_parallel_subtasks,
                      self.spec.serialized_fn.runtime_context.max_number_of_parallel_subtasks,
                      self.spec.serialized_fn.runtime_context.index_of_this_subtask,
                      self.spec.serialized_fn.runtime_context.attempt_number,
                      {p.key: p.value for p in self.spec.serialized_fn.runtime_context.job_parameters}
                  )
                  user_defined_func.open(runtime_context)
  
      def generate_func(self, serialized_fn):
          func, user_defined_func = operation_utils.extract_data_stream_stateless_function(
              serialized_fn)
          return func, [user_defined_func]
  
  
  class InternalRuntimeContext(RuntimeContext):
  
      def __init__(self,
                   task_name: str,
                   task_name_with_subtasks: str,
                   number_of_parallel_subtasks: int,
                   max_number_of_parallel_subtasks: int,
                   index_of_this_subtask: int,
                   attempt_number: int,
                   job_parameters: Dict[str, str],
                   keyed_state_backend: RemoteKeyedStateBackend):
          super(InternalRuntimeContext, self).__init__(
              task_name, task_name_with_subtasks, number_of_parallel_subtasks,
              max_number_of_parallel_subtasks, index_of_this_subtask, attempt_number,
              job_parameters)
          self._keyed_state_backend = keyed_state_backend
  
      def get_state(self, state_descriptor: ValueStateDescriptor) -> ValueState:
          return self._keyed_state_backend.get_value_state(state_descriptor.name, PickleCoder())
  
      def get_list_state(self, state_descriptor: ListStateDescriptor) -> ListState:
          return self._keyed_state_backend.get_list_state(state_descriptor.name, PickleCoder())
  
      def get_map_state(self, state_descriptor: MapStateDescriptor) -> MapState:
          return self._keyed_state_backend.get_map_state(state_descriptor.name, PickleCoder(),
                                                         PickleCoder())
  
      def get_reducing_state(self, state_descriptor: ReducingStateDescriptor) -> ReducingState:
          return self._keyed_state_backend.get_reducing_state(
              state_descriptor.get_name(), PickleCoder(), state_descriptor.get_reduce_function())
  
  
  class ProcessFunctionOperation(DataStreamStatelessFunctionOperation):
  
      def __init__(self, spec):
          self.timer_service = ProcessFunctionOperation.InternalTimerService()
          self.function_context = ProcessFunctionOperation.InternalProcessFunctionContext(
              self.timer_service)
          super(ProcessFunctionOperation, self).__init__(spec)
  
      def generate_func(self, serialized_fn) -> tuple:
          func, proc_func = operation_utils.extract_process_function(
              serialized_fn, self.function_context)
          return func, [proc_func]
  
      class InternalProcessFunctionContext(ProcessFunction.Context):
          """
          Internal implementation of ProcessFunction.Context.
          """
  
          def __init__(self, timer_service: TimerService):
              self._timer_service = timer_service
              self._timestamp = None
  
          def timer_service(self):
              return self._timer_service
  
          def timestamp(self) -> int:
              return self._timestamp
  
          def set_timestamp(self, ts: int):
              self._timestamp = ts
  
      class InternalTimerService(TimerService):
          """
          Internal implementation of TimerService.
          """
          def __init__(self):
              self._current_watermark = None
  
          def current_processing_time(self) -> int:
              return int(time.time() * 1000)
  
          def current_watermark(self):
              return self._current_watermark
  
          def set_current_watermark(self, wm):
              self._current_watermark = wm
  
          def register_processing_time_timer(self, t: int):
              raise Exception("Register timers is only supported on a keyed stream.")
  
          def register_event_time_timer(self, t: int):
              raise Exception("Register timers is only supported on a keyed stream.")
  
  
  class KeyedProcessFunctionOperation(StatefulFunctionOperation):
  
      def __init__(self, spec, keyed_state_backend):
          self._collector = KeyedProcessFunctionOperation.InternalCollector()
          internal_timer_service = KeyedProcessFunctionOperation.InternalTimerService(
              self._collector, keyed_state_backend)
          self.function_context = KeyedProcessFunctionOperation.InternalKeyedProcessFunctionContext(
              internal_timer_service)
          self.on_timer_ctx = KeyedProcessFunctionOperation\
              .InternalKeyedProcessFunctionOnTimerContext(internal_timer_service)
          super(KeyedProcessFunctionOperation, self).__init__(spec, keyed_state_backend)
  
      def generate_func(self, serialized_fn) -> Tuple:
          func, proc_func = operation_utils.extract_keyed_process_function(
              serialized_fn, self.function_context, self.on_timer_ctx, self._collector,
              self.keyed_state_backend)
          return func, [proc_func]
  
      def open(self):
          for user_defined_func in self.user_defined_funcs:
              if hasattr(user_defined_func, 'open'):
                  runtime_context = InternalRuntimeContext(
                      self.spec.serialized_fn.runtime_context.task_name,
                      self.spec.serialized_fn.runtime_context.task_name_with_subtasks,
                      self.spec.serialized_fn.runtime_context.number_of_parallel_subtasks,
                      self.spec.serialized_fn.runtime_context.max_number_of_parallel_subtasks,
                      self.spec.serialized_fn.runtime_context.index_of_this_subtask,
                      self.spec.serialized_fn.runtime_context.attempt_number,
                      {p.key: p.value for p in
                       self.spec.serialized_fn.runtime_context.job_parameters},
                      self.keyed_state_backend)
                  user_defined_func.open(runtime_context)
  
      class InternalCollector(object):
          """
          Internal implementation of the Collector. It uses a buffer list to store data to be emitted.
          There will be a header flag for each data type. 0 means it is a proc time timer registering
          request, while 1 means it is an event time timer and 2 means it is a normal data. When
          registering a timer, it must take along with the corresponding key for it.
          """
  
          def __init__(self):
              self.buf = []
  
          def collect_reg_proc_timer(self, a: Any, key: Any):
              self.buf.append(
                  (operation_utils.KeyedProcessFunctionOutputFlag.REGISTER_PROC_TIMER.value,
                   a, key, None))
  
          def collect_reg_event_timer(self, a: Any, key: Any):
              self.buf.append(
                  (operation_utils.KeyedProcessFunctionOutputFlag.REGISTER_EVENT_TIMER.value,
                   a, key, None))
  
          def collect_del_proc_timer(self, a: Any, key: Any):
              self.buf.append(
                  (operation_utils.KeyedProcessFunctionOutputFlag.DEL_PROC_TIMER.value,
                   a, key, None))
  
          def collect_del_event_timer(self, a: Any, key: Any):
              self.buf.append(
                  (operation_utils.KeyedProcessFunctionOutputFlag.DEL_EVENT_TIMER.value,
                   a, key, None))
  
          def collect(self, a: Any):
              self.buf.append((operation_utils.KeyedProcessFunctionOutputFlag.NORMAL_DATA.value, a))
  
          def clear(self):
              self.buf.clear()
  
      class InternalKeyedProcessFunctionOnTimerContext(KeyedProcessFunction.OnTimerContext):
          """
          Internal implementation of ProcessFunction.OnTimerContext.
          """
  
          def __init__(self, timer_service: TimerService):
              self._timer_service = timer_service
              self._time_domain = None
              self._timestamp = None
              self._current_key = None
  
          def get_current_key(self):
              return self._current_key
  
          def set_current_key(self, current_key):
              self._current_key = current_key
  
          def timer_service(self) -> TimerService:
              return self._timer_service
  
          def timestamp(self) -> int:
              return self._timestamp
  
          def set_timestamp(self, ts: int):
              self._timestamp = ts
  
          def time_domain(self) -> TimeDomain:
              return self._time_domain
  
          def set_time_domain(self, td: TimeDomain):
              self._time_domain = td
  
      class InternalKeyedProcessFunctionContext(KeyedProcessFunction.Context):
          """
          Internal implementation of KeyedProcessFunction.Context.
          """
  
          def __init__(self, timer_service: TimerService):
              self._timer_service = timer_service
              self._timestamp = None
              self._current_key = None
  
          def get_current_key(self):
              return self._current_key
  
          def set_current_key(self, current_key):
              self._current_key = current_key
  
          def timer_service(self) -> TimerService:
              return self._timer_service
  
          def timestamp(self) -> int:
              return self._timestamp
  
          def set_timestamp(self, ts: int):
              self._timestamp = ts
  
      class InternalTimerService(TimerService):
          """
          Internal implementation of TimerService.
          """
  
          def __init__(self, collector, keyed_state_backend):
              self._collector = collector
              self._keyed_state_backend = keyed_state_backend
              self._current_watermark = None
  
          def current_processing_time(self) -> int:
              return int(time.time() * 1000)
  
          def current_watermark(self) -> int:
              return self._current_watermark
  
          def set_current_watermark(self, wm):
              self._current_watermark = wm
  
          def register_processing_time_timer(self, t: int):
              current_key = self._keyed_state_backend.get_current_key()
              self._collector.collect_reg_proc_timer(t, current_key)
  
          def register_event_time_timer(self, t: int):
              current_key = self._keyed_state_backend.get_current_key()
              self._collector.collect_reg_event_timer(t, current_key)
  
          def delete_processing_time_timer(self, t: int):
              current_key = self._keyed_state_backend.get_current_key()
              self._collector.collect_del_proc_timer(t, current_key)
  
          def delete_event_time_timer(self, t: int):
              current_key = self._keyed_state_backend.get_current_key()
              self._collector.collect_del_event_timer(t, current_key)
   ################################################################################
  #  Licensed to the Apache Software Foundation (ASF) under one
  #  or more contributor license agreements.  See the NOTICE file
  #  distributed with this work for additional information
  #  regarding copyright ownership.  The ASF licenses this file
  #  to you under the Apache License, Version 2.0 (the
  #  "License"); you may not use this file except in compliance
  #  with the License.  You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  #  See the License for the specific language governing permissions and
  # limitations under the License.
  ################################################################################
  import datetime
  import decimal
  
  from pandas.util.testing import assert_frame_equal
  
  from pyflink.common import Row
  from pyflink.table.types import DataTypes
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkBlinkBatchTableTestCase, \
      PyFlinkBlinkStreamTableTestCase, PyFlinkOldStreamTableTestCase
  
  
  class PandasConversionTestBase(object):
  
      @classmethod
      def setUpClass(cls):
          super(PandasConversionTestBase, cls).setUpClass()
          cls.data = [(1, 1, 1, 1, True, 1.1, 1.2, 'hello', bytearray(b"aaa"),
                       decimal.Decimal('1000000000000000000.01'), datetime.date(2014, 9, 13),
                       datetime.time(hour=1, minute=0, second=1),
                       datetime.datetime(1970, 1, 1, 0, 0, 0, 123000), ['hello', '中文'],
                       Row(a=1, b='hello', c=datetime.datetime(1970, 1, 1, 0, 0, 0, 123000),
                           d=[1, 2])),
                      (1, 2, 2, 2, False, 2.1, 2.2, 'world', bytearray(b"bbb"),
                       decimal.Decimal('1000000000000000000.02'), datetime.date(2014, 9, 13),
                       datetime.time(hour=1, minute=0, second=1),
                       datetime.datetime(1970, 1, 1, 0, 0, 0, 123000), ['hello', '中文'],
                       Row(a=1, b='hello', c=datetime.datetime(1970, 1, 1, 0, 0, 0, 123000),
                           d=[1, 2]))]
          cls.data_type = DataTypes.ROW(
              [DataTypes.FIELD("f1", DataTypes.TINYINT()),
               DataTypes.FIELD("f2", DataTypes.SMALLINT()),
               DataTypes.FIELD("f3", DataTypes.INT()),
               DataTypes.FIELD("f4", DataTypes.BIGINT()),
               DataTypes.FIELD("f5", DataTypes.BOOLEAN()),
               DataTypes.FIELD("f6", DataTypes.FLOAT()),
               DataTypes.FIELD("f7", DataTypes.DOUBLE()),
               DataTypes.FIELD("f8", DataTypes.STRING()),
               DataTypes.FIELD("f9", DataTypes.BYTES()),
               DataTypes.FIELD("f10", DataTypes.DECIMAL(38, 18)),
               DataTypes.FIELD("f11", DataTypes.DATE()),
               DataTypes.FIELD("f12", DataTypes.TIME()),
               DataTypes.FIELD("f13", DataTypes.TIMESTAMP(3)),
               DataTypes.FIELD("f14", DataTypes.ARRAY(DataTypes.STRING())),
               DataTypes.FIELD("f15", DataTypes.ROW(
                   [DataTypes.FIELD("a", DataTypes.INT()),
                    DataTypes.FIELD("b", DataTypes.STRING()),
                    DataTypes.FIELD("c", DataTypes.TIMESTAMP(3)),
                    DataTypes.FIELD("d", DataTypes.ARRAY(DataTypes.INT()))]))], False)
          cls.pdf = cls.create_pandas_data_frame()
  
      @classmethod
      def create_pandas_data_frame(cls):
          data_dict = {}
          for j, name in enumerate(cls.data_type.names):
              data_dict[name] = [cls.data[i][j] for i in range(len(cls.data))]
          # need convert to numpy types
          import numpy as np
          data_dict["f1"] = np.int8(data_dict["f1"])
          data_dict["f2"] = np.int16(data_dict["f2"])
          data_dict["f3"] = np.int32(data_dict["f3"])
          data_dict["f4"] = np.int64(data_dict["f4"])
          data_dict["f6"] = np.float32(data_dict["f6"])
          data_dict["f7"] = np.float64(data_dict["f7"])
          data_dict["f15"] = [row.as_dict() for row in data_dict["f15"]]
          import pandas as pd
          return pd.DataFrame(data=data_dict,
                              index=[2., 3.],
                              columns=['f1', 'f2', 'f3', 'f4', 'f5', 'f6', 'f7', 'f8', 'f9',
                                       'f10', 'f11', 'f12', 'f13', 'f14', 'f15'])
  
  
  class PandasConversionTests(PandasConversionTestBase):
  
      def test_from_pandas_with_incorrect_schema(self):
          fields = self.data_type.fields.copy()
          fields[0], fields[7] = fields[7], fields[0]  # swap str with tinyint
          wrong_schema = DataTypes.ROW(fields)  # should be DataTypes.STRING()
          with self.assertRaisesRegex(Exception, "Expected a string.*got int8"):
              self.t_env.from_pandas(self.pdf, schema=wrong_schema)
  
      def test_from_pandas_with_names(self):
          # skip decimal as currently only decimal(38, 18) is supported
          pdf = self.pdf.drop(['f10', 'f11', 'f12', 'f13', 'f14', 'f15'], axis=1)
          new_names = list(map(str, range(len(pdf.columns))))
          table = self.t_env.from_pandas(pdf, schema=new_names)
          self.assertEqual(new_names, table.get_schema().get_field_names())
          table = self.t_env.from_pandas(pdf, schema=tuple(new_names))
          self.assertEqual(new_names, table.get_schema().get_field_names())
  
      def test_from_pandas_with_types(self):
          new_types = self.data_type.field_types()
          new_types[0] = DataTypes.BIGINT()
          table = self.t_env.from_pandas(self.pdf, schema=new_types)
          self.assertEqual(new_types, table.get_schema().get_field_data_types())
          table = self.t_env.from_pandas(self.pdf, schema=tuple(new_types))
          self.assertEqual(new_types, table.get_schema().get_field_data_types())
  
  
  class PandasConversionITTests(PandasConversionTestBase):
  
      def test_from_pandas(self):
          table = self.t_env.from_pandas(self.pdf, self.data_type, 5)
          self.assertEqual(self.data_type, table.get_schema().to_row_data_type())
  
          table = table.filter(table.f2 < 2)
          table_sink = source_sink_utils.TestAppendSink(
              self.data_type.field_names(),
              self.data_type.field_types())
          self.t_env.register_table_sink("Results", table_sink)
          table.execute_insert("Results").wait()
          actual = source_sink_utils.results()
          self.assert_equals(actual,
                             ["+I[1, 1, 1, 1, true, 1.1, 1.2, hello, [97, 97, 97], "
                              "1000000000000000000.010000000000000000, 2014-09-13, 01:00:01, "
                              "1970-01-01 00:00:00.123, [hello, 中文], +I[1, hello, "
                              "1970-01-01 00:00:00.123, [1, 2]]]"])
  
      def test_to_pandas(self):
          table = self.t_env.from_pandas(self.pdf, self.data_type)
          result_pdf = table.to_pandas()
          result_pdf.index = self.pdf.index
          self.assertEqual(2, len(result_pdf))
          assert_frame_equal(self.pdf, result_pdf)
  
      def test_empty_to_pandas(self):
          table = self.t_env.from_pandas(self.pdf, self.data_type)
          pdf = table.filter(table.f1 < 0).to_pandas()
          self.assertTrue(pdf.empty)
  
      def test_to_pandas_for_retract_table(self):
          table = self.t_env.from_pandas(self.pdf, self.data_type)
          result_pdf = table.group_by(table.f1).select(table.f2.max.alias('f2')).to_pandas()
          import pandas as pd
          import numpy as np
          assert_frame_equal(result_pdf, pd.DataFrame(data={'f2': np.int16([2])}))
  
          result_pdf = table.group_by("f2").select("max(f1) as f2").to_pandas()
          assert_frame_equal(result_pdf, pd.DataFrame(data={'f2': np.int8([1, 1])}))
  
  
  class StreamPandasConversionTests(PandasConversionITTests,
                                    PyFlinkOldStreamTableTestCase):
      pass
  
  
  class BlinkBatchPandasConversionTests(PandasConversionTests,
                                        PandasConversionITTests,
                                        PyFlinkBlinkBatchTableTestCase):
      pass
  
  
  class BlinkStreamPandasConversionTests(PandasConversionITTests,
                                         PyFlinkBlinkStreamTableTestCase):
      def test_to_pandas_with_event_time(self):
          self.t_env.get_config().get_configuration().set_string("parallelism.default", "1")
          # create source file path
          import tempfile
          import os
          tmp_dir = tempfile.gettempdir()
          data = [
              '2018-03-11 03:10:00',
              '2018-03-11 03:10:00',
              '2018-03-11 03:10:00',
              '2018-03-11 03:40:00',
              '2018-03-11 04:20:00',
              '2018-03-11 03:30:00'
          ]
          source_path = tmp_dir + '/test_to_pandas_with_event_time.csv'
          with open(source_path, 'w') as fd:
              for ele in data:
                  fd.write(ele + '\n')
  
          self.t_env.get_config().get_configuration().set_string(
              "pipeline.time-characteristic", "EventTime")
  
          source_table = """
              create table source_table(
                  rowtime TIMESTAMP(3),
                  WATERMARK FOR rowtime AS rowtime - INTERVAL '60' MINUTE
              ) with(
                  'connector.type' = 'filesystem',
                  'format.type' = 'csv',
                  'connector.path' = '%s',
                  'format.ignore-first-line' = 'false',
                  'format.field-delimiter' = ','
              )
          """ % source_path
          self.t_env.execute_sql(source_table)
          t = self.t_env.from_path("source_table")
          result_pdf = t.to_pandas()
          import pandas as pd
          os.remove(source_path)
          assert_frame_equal(result_pdf, pd.DataFrame(
              data={"rowtime": [
                  datetime.datetime(2018, 3, 11, 3, 10),
                  datetime.datetime(2018, 3, 11, 3, 10),
                  datetime.datetime(2018, 3, 11, 3, 10),
                  datetime.datetime(2018, 3, 11, 3, 40),
                  datetime.datetime(2018, 3, 11, 4, 20),
                  datetime.datetime(2018, 3, 11, 3, 30),
              ]}))
  
apis: pandas.Series
apis: numpy.int8 pandas.DataFrame numpy.float32 numpy.int64 numpy.int32 pandas.util.testing.assert_frame_equal numpy.int16 numpy.float64