[GitHub] [flink] dianfu commented on a change in pull request #18957: [FLINK-26444][python]Window allocator supporting pyflink datastream API

[GitBox]

dianfu commented on a change in pull request #18957:
URL: https://github.com/apache/flink/pull/18957#discussion_r818274208



##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+

Review comment:
       need to override the can_merge method

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math

Review comment:
       missing license header

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():

Review comment:
       ```suggestion
           if windowMaxTimestamp > ctx.get_current_watermark():
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,

Review comment:
       need to override the can_merge method

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.

Review comment:
       check style issue, it exceeds 100 characters in one line.

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)

Review comment:
       ```suggestion
           self._window_size = window_size
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):

Review comment:
       ```suggestion
   class EventTimeTrigger(Trigger[T, TimeWindow]):
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(10), False))
+
+    A WindowAssigner that windows elements into windows based on the timestamp 
of the elements. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute:
+     ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, offset: Time, is_event_time: bool):

Review comment:
       ```suggestion
       def __init__(self, size: int, offset: int, is_event_time: bool):
   ```
   Use int to keep consistent with the Java API. Besides, it would be great to 
add a few static utility methods just as did in the Java API.

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(10), False))
+
+    A WindowAssigner that windows elements into windows based on the timestamp 
of the elements. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute:
+     ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, offset: Time, is_event_time: bool):
+        self._size = size.to_milliseconds()
+        self._offset = offset.to_milliseconds()
+        self._is_event_time = is_event_time
+
+    def assign_windows(self,
+                       element: object,
+                       timestamp: int,
+                       context: WindowAssigner.WindowAssignerContext) -> 
Collection[TimeWindow]:
+        if self._is_event_time is False:
+            timestamp = context.get_current_processing_time()
+
+        start = TimeWindow.get_window_start_with_offset(timestamp, 
self._offset, self._size)
+        return [TimeWindow(start, start + self._size)]
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        if self._is_event_time is True:
+            return EventTimeTrigger()
+        else:
+            return ProcessingTimeTrigger()
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return TimeWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return self._is_event_time
+
+    def __repr__(self):
+        return "TumblingWindowAssigner(%s,%s,%s)" % (self._size, self._offset, 
self.is_event_time)
+
+'''
+A WindowAssigner that windows elements into windows based on the number of the 
elements. Windows cannot overlap.

Review comment:
       could be removed

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(10), False))
+
+    A WindowAssigner that windows elements into windows based on the timestamp 
of the elements. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute:
+     ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, offset: Time, is_event_time: bool):
+        self._size = size.to_milliseconds()
+        self._offset = offset.to_milliseconds()
+        self._is_event_time = is_event_time
+
+    def assign_windows(self,
+                       element: object,
+                       timestamp: int,
+                       context: WindowAssigner.WindowAssignerContext) -> 
Collection[TimeWindow]:
+        if self._is_event_time is False:
+            timestamp = context.get_current_processing_time()
+
+        start = TimeWindow.get_window_start_with_offset(timestamp, 
self._offset, self._size)
+        return [TimeWindow(start, start + self._size)]
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        if self._is_event_time is True:
+            return EventTimeTrigger()
+        else:
+            return ProcessingTimeTrigger()
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return TimeWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return self._is_event_time
+
+    def __repr__(self):
+        return "TumblingWindowAssigner(%s,%s,%s)" % (self._size, self._offset, 
self.is_event_time)
+
+'''
+A WindowAssigner that windows elements into windows based on the number of the 
elements. Windows cannot overlap.
+'''
+class CountTumblingWindowAssigner(WindowAssigner[object, CountWindow]):
+    """
+    A WindowAssigner that windows elements into fixed-size windows based on 
the count number
+    of the elements. Windows cannot overlap.
+    """
+    def __init__(self, window_size: int):
+        """
+        Windows this KeyedStream into tumbling count windows.
+        :param window_size: The size of the windows in number of elements.
+        """
+        self._window_size = window_size
+        self._counter_state_descriptor = ReducingStateDescriptor(
+            "assigner_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def assign_windows(self,
+                       element: T,
+                       timestamp: int,
+                       context: 'WindowAssigner.WindowAssignerContext') -> 
Collection[W]:
+        counter = context.get_runtime_context().get_reducing_state(
+            self._counter_state_descriptor)
+        if counter.get() is None:
+            counter.add(0)
+        result = [CountWindow(counter.get() // self._window_size)]
+        counter.add(1)
+        return result
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        return CountTrigger(self._window_size)
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return CountWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return False
+
+    def __repr__(self) -> str:
+        return "CountTumblingWindowAssigner(%s)" % (self._window_size)
+
+
+class SlidingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into sliding windows based on the 
current system time of the machine the operation is running on. Windows can 
possibly overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(SlidingWindowAssigner(Time.minutes(1), 
Time.seconds(10), Time.seconds(0), False))
+
+    A WindowAssigner that windows elements into sliding windows based on the 
timestamp of the elements. Windows can possibly overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(SlidingWindowAssigner(Time.minutes(1), 
Time.seconds(10), Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, slide: Time, offset: Time, is_event_time: 
bool):
+        self._size = size.to_milliseconds()
+        self._slide = slide.to_milliseconds()
+        self._offset = offset.to_milliseconds()
+        self._is_event_time = is_event_time
+        self._pane_size = math.gcd(size, slide)
+        self._num_panes_per_window = size // self._pane_size
+
+    def assign_windows(
+        self,
+        element: T,
+        timestamp: int,
+        context: 'WindowAssigner.WindowAssignerContext') -> Collection[W]:
+        if self._is_event_time is False:
+            timestamp = context.get_current_processing_time()
+
+        last_start = TimeWindow.get_window_start_with_offset(timestamp, 
self._offset, self._slide)
+        windows = [TimeWindow(start, start + self._size)
+                   for start in range(last_start, timestamp - self._size, 
-self._slide)]
+        return windows
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        if self._is_event_time is True:
+            return EventTimeTrigger()
+        else:
+            return ProcessingTimeTrigger()
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return TimeWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return self._is_event_time
+
+    def __repr__(self) -> str:
+        return "SlidingWindowAssigner(%s, %s, %s, %s)" % (self._size, 
self._slide, self._offset, self._is_event_time)
+
+
+class CountSlidingWindowAssigner(WindowAssigner[object, CountWindow]):
+    """
+    A WindowAssigner that windows elements into sliding windows based on the 
count number of
+    the elements. Windows can possibly overlap.
+    """
+    def __init__(self, window_size: int, window_slide: int):
+        """
+        Windows this KeyedStream into sliding count windows.
+        :param window_size: The size of the windows in number of elements.
+        :param window_slide: The slide interval in number of elements.
+        """
+        self._window_size = window_size
+        self._window_slide = window_slide
+        self._count = None  # type: ValueState
+        self._counter_state_descriptor = ReducingStateDescriptor(
+            "slide-count-assigner", lambda a, b: a + b, Types.BIG_INT())
+
+    def assign_windows(self,
+                       element: T,
+                       timestamp: int,
+                       context: 'WindowAssigner.WindowAssignerContext') -> 
Collection[W]:
+        count_descriptor = ValueStateDescriptor('slide-count-assigner', 
Types.LONG())
+        self._count = context.get_runtime_context().get_state(count_descriptor)
+        count_value = self._count.value()
+        if count_value is None:
+            current_count = 0
+        else:
+            current_count = count_value
+        self._count.update(current_count + 1)
+        last_id = current_count // self._window_slide
+        last_start = last_id * self._window_slide
+        last_end = last_start + self._window_size - 1
+        windows = []
+        while last_id >= 0 and last_start <= current_count <= last_end:
+            if last_start <= current_count <= last_end:
+                windows.append(CountWindow(last_id))
+            last_id -= 1
+            last_start -= self._window_slide
+            last_end -= self._window_slide
+        return windows
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        return CountTrigger(self._window_size)
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return CountWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return False
+
+    def __repr__(self):
+        return "CountSlidingWindowAssigner(%s, %s)" % (self._window_size, 
self._window_slide)
+
+
+class SessionWindowAssigner(MergingWindowAssigner[object, TimeWindow]):

Review comment:
       It has already support dynamic session windows in the Java API and it 
would be great to align with it. See DynamicEventTimeSessionWindows for more 
details.

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):

Review comment:
       Move all these classes to module pyflink/datastream/window.py

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):

Review comment:
       ```suggestion
   class ProcessingTimeTrigger(Trigger[T, TimeWindow]):
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());

Review comment:
       ```suggestion
           ctx.register_processing_time_timer(window.max_timestamp())
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()

Review comment:
       ```suggestion
           window_max_timestamp = window.max_timestamp()
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());

Review comment:
       ```suggestion
           ctx.delete_processing_time_timer(window.max_timestamp())
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.

Review comment:
       ```suggestion
               #               though, all elements are retained.
               state.clear()
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):

Review comment:
       ```suggestion
   class CountTrigger(Trigger[T, CountWindow]):
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):

Review comment:
       ```suggestion
   class TumblingWindowAssigner(WindowAssigner[T, TimeWindow]):
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:

Review comment:
       need to override the can_merge method

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();

Review comment:
       ```suggestion
           window_max_timestamp = window.max_timestamp();
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())

Review comment:
       ```suggestion
               "count", lambda a, b: a + b, Types.LONG())
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.

Review comment:
       need fix the check style issue. Please also fix other places.

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(10), False))
+
+    A WindowAssigner that windows elements into windows based on the timestamp 
of the elements. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute:
+     ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, offset: Time, is_event_time: bool):
+        self._size = size.to_milliseconds()
+        self._offset = offset.to_milliseconds()
+        self._is_event_time = is_event_time
+
+    def assign_windows(self,
+                       element: object,
+                       timestamp: int,
+                       context: WindowAssigner.WindowAssignerContext) -> 
Collection[TimeWindow]:
+        if self._is_event_time is False:
+            timestamp = context.get_current_processing_time()
+
+        start = TimeWindow.get_window_start_with_offset(timestamp, 
self._offset, self._size)
+        return [TimeWindow(start, start + self._size)]
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        if self._is_event_time is True:
+            return EventTimeTrigger()
+        else:
+            return ProcessingTimeTrigger()
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return TimeWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return self._is_event_time
+
+    def __repr__(self):
+        return "TumblingWindowAssigner(%s,%s,%s)" % (self._size, self._offset, 
self.is_event_time)
+
+'''
+A WindowAssigner that windows elements into windows based on the number of the 
elements. Windows cannot overlap.
+'''
+class CountTumblingWindowAssigner(WindowAssigner[object, CountWindow]):
+    """
+    A WindowAssigner that windows elements into fixed-size windows based on 
the count number
+    of the elements. Windows cannot overlap.
+    """
+    def __init__(self, window_size: int):
+        """
+        Windows this KeyedStream into tumbling count windows.
+        :param window_size: The size of the windows in number of elements.
+        """
+        self._window_size = window_size
+        self._counter_state_descriptor = ReducingStateDescriptor(
+            "assigner_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def assign_windows(self,
+                       element: T,
+                       timestamp: int,
+                       context: 'WindowAssigner.WindowAssignerContext') -> 
Collection[W]:
+        counter = context.get_runtime_context().get_reducing_state(
+            self._counter_state_descriptor)
+        if counter.get() is None:
+            counter.add(0)
+        result = [CountWindow(counter.get() // self._window_size)]
+        counter.add(1)
+        return result
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        return CountTrigger(self._window_size)
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return CountWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return False
+
+    def __repr__(self) -> str:
+        return "CountTumblingWindowAssigner(%s)" % (self._window_size)
+
+
+class SlidingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into sliding windows based on the 
current system time of the machine the operation is running on. Windows can 
possibly overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(SlidingWindowAssigner(Time.minutes(1), 
Time.seconds(10), Time.seconds(0), False))
+
+    A WindowAssigner that windows elements into sliding windows based on the 
timestamp of the elements. Windows can possibly overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(SlidingWindowAssigner(Time.minutes(1), 
Time.seconds(10), Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, slide: Time, offset: Time, is_event_time: 
bool):

Review comment:
       ```suggestion
       def __init__(self, size: int, slide: int, offset: int, is_event_time: 
bool):
   ```

##########
File path: 
flink-python/pyflink/fn_execution/datastream/window/window_assigner.py
##########
@@ -0,0 +1,376 @@
+import math
+from typing import Iterable, Collection
+
+from pyflink.common import TypeSerializer, Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import Trigger
+from pyflink.datastream.state import ValueStateDescriptor, ValueState, 
ReducingStateDescriptor
+from pyflink.datastream.window import TimeWindow, CountWindow, WindowAssigner, 
T, TimeWindowSerializer, TriggerResult, \
+    CountWindowSerializer, MergingWindowAssigner
+from pyflink.fn_execution.table.window_context import W
+
+
+class EventTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the watermark passes the end of the window to 
which a pane belongs.
+    """
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: TimeWindow,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if window.max_timestamp() <= ctx.get_current_watermark():
+            return TriggerResult.FIRE
+        else:
+            ctx.register_event_time_timer(window.max_timestamp())
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: TimeWindow,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: TimeWindow,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        if time == window.max_timestamp():
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: TimeWindow,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp()
+        if windowMaxTimestamp >= ctx.get_current_watermark():
+            ctx.register_event_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: TimeWindow,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_event_time_timer(window.max_timestamp())
+
+
+class ProcessingTimeTrigger(Trigger[object, TimeWindow]):
+    """
+    A Trigger that fires once the current system time passes the end of the 
window to which a pane belongs.
+    """
+    def on_element(self,
+                   element: T,
+                   timestamp: int,
+                   window: W,
+                   ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        ctx.register_processing_time_timer(window.max_timestamp());
+        return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: W,
+                           ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.FIRE
+
+    def on_event_time(self,
+                      time: int,
+                      window: W,
+                      ctx: 'Trigger.TriggerContext') -> TriggerResult:
+        return TriggerResult.CONTINUE
+
+    def on_merge(self,
+                 window: W,
+                 ctx: 'Trigger.OnMergeContext') -> None:
+        windowMaxTimestamp = window.max_timestamp();
+        if windowMaxTimestamp > ctx.get_current_processing_time():
+            ctx.register_processing_time_timer(windowMaxTimestamp)
+
+    def clear(self,
+              window: W,
+              ctx: 'Trigger.TriggerContext') -> None:
+        ctx.delete_processing_time_timer(window.max_timestamp());
+
+
+class CountTrigger(Trigger[object, CountWindow]):
+    """
+    A Trigger that fires once the count of elements in a pane reaches the 
given count.
+    """
+    def __init__(self, window_size: int):
+        self._window_size = int(window_size)
+        self._count_state_descriptor = ReducingStateDescriptor(
+            "trigger_counter", lambda a, b: a + b, Types.BIG_INT())
+
+    def on_element(self,
+                   element: object,
+                   timestamp: int,
+                   window: CountWindow,
+                   ctx: Trigger.TriggerContext) -> TriggerResult:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)  # 
type: ReducingState
+        state.add(1)
+        if state.get() >= self._window_size:
+            # On FIRE, the window is evaluated and results are emitted. The 
window is not purged
+            #               though, all elements are retained.
+            return TriggerResult.FIRE
+        else:
+            # No action is taken on the window.
+            return TriggerResult.CONTINUE
+
+    def on_processing_time(self,
+                           time: int,
+                           window: CountWindow,
+                           ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_event_time(self,
+                      time: int,
+                      window: CountWindow,
+                      ctx: Trigger.TriggerContext) -> TriggerResult:
+        # No action is taken on the window.
+        return TriggerResult.CONTINUE
+
+    def on_merge(self, window: CountWindow, ctx: Trigger.OnMergeContext) -> 
None:
+        ctx.merge_partitioned_state(self._count_state_descriptor)
+
+    def clear(self, window: CountWindow, ctx: Trigger.TriggerContext) -> None:
+        state = ctx.get_partitioned_state(self._count_state_descriptor)
+        state.clear()
+
+
+class TumblingWindowAssigner(WindowAssigner[object, TimeWindow]):
+    """
+    A WindowAssigner that windows elements into windows based on the current 
system time of the machine the operation is running on. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute, every 10 seconds:
+    ::
+            >>> data_stream.key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(10), False))
+
+    A WindowAssigner that windows elements into windows based on the timestamp 
of the elements. Windows cannot overlap.
+    For example, in order to window into windows of 1 minute:
+     ::
+            >>> 
data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+            >>> .key_by(lambda x: x[0], key_type=Types.STRING()) \
+            >>> .window(TumblingWindowAssigner(Time.minutes(1), 
Time.seconds(0), True))
+    """
+    def __init__(self, size: Time, offset: Time, is_event_time: bool):
+        self._size = size.to_milliseconds()
+        self._offset = offset.to_milliseconds()
+        self._is_event_time = is_event_time
+
+    def assign_windows(self,
+                       element: object,
+                       timestamp: int,
+                       context: WindowAssigner.WindowAssignerContext) -> 
Collection[TimeWindow]:
+        if self._is_event_time is False:
+            timestamp = context.get_current_processing_time()
+
+        start = TimeWindow.get_window_start_with_offset(timestamp, 
self._offset, self._size)
+        return [TimeWindow(start, start + self._size)]
+
+    def get_default_trigger(self, env) -> Trigger[T, W]:
+        if self._is_event_time is True:
+            return EventTimeTrigger()
+        else:
+            return ProcessingTimeTrigger()
+
+    def get_window_serializer(self) -> TypeSerializer[W]:
+        return TimeWindowSerializer()
+
+    def is_event_time(self) -> bool:
+        return self._is_event_time
+
+    def __repr__(self):
+        return "TumblingWindowAssigner(%s,%s,%s)" % (self._size, self._offset, 
self.is_event_time)
+
+'''
+A WindowAssigner that windows elements into windows based on the number of the 
elements. Windows cannot overlap.
+'''
+class CountTumblingWindowAssigner(WindowAssigner[object, CountWindow]):

Review comment:
       It uses GlobalWindows for count window in the Java API. Is it possible 
to do so in Python?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org