Revert "fix python"

This reverts commit c540f35.

Author: XComp

flink-python/pyflink/common/__init__.py

@@ -64,7 +64,7 @@
 from pyflink.common.serializer import TypeSerializer
 from pyflink.common.typeinfo import Types, TypeInformation
 from pyflink.common.types import Row, RowKind
-from pyflink.common.time import Duration, Instant
+from pyflink.common.time import Duration, Instant, Time
 from pyflink.common.watermark_strategy import WatermarkStrategy, \
     AssignerWithPeriodicWatermarksWrapper
 
@@ -94,6 +94,7 @@
     "WatermarkStrategy",
     "Duration",
     "Instant",
+    "Time",
     "AssignerWithPeriodicWatermarksWrapper"
 ]
 

flink-python/pyflink/common/time.py

@@ -17,7 +17,7 @@
 ################################################################################
 from pyflink.java_gateway import get_gateway
 
-__all__ = ['Duration', 'Instant']
+__all__ = ['Duration', 'Instant', 'Time']
 
 
 class Duration(object):
@@ -51,9 +51,6 @@ def of_nanos(nanos: int):
     def of_seconds(seconds: int):
         return Duration(get_gateway().jvm.java.time.Duration.ofSeconds(seconds))
 
-    def to_millis(self):
-        return self._j_duration.to_millis()
-
     def __eq__(self, other):
         return isinstance(other, Duration) and self._j_duration.equals(other._j_duration)
 
@@ -86,3 +83,42 @@ def __eq__(self, other):
 
     def __repr__(self):
         return 'Instant<{}, {}>'.format(self.seconds, self.nanos)
+
+
+class Time(object):
+    """
+    The definition of a time interval.
+    """
+
+    def __init__(self, milliseconds: int):
+        self._milliseconds = milliseconds
+
+    def to_milliseconds(self) -> int:
+        return self._milliseconds
+
+    @staticmethod
+    def milliseconds(milliseconds: int):
+        return Time(milliseconds)
+
+    @staticmethod
+    def seconds(seconds: int):
+        return Time.milliseconds(seconds * 1000)
+
+    @staticmethod
+    def minutes(minutes: int):
+        return Time.seconds(minutes * 60)
+
+    @staticmethod
+    def hours(hours: int):
+        return Time.minutes(hours * 60)
+
+    @staticmethod
+    def days(days: int):
+        return Time.hours(days * 24)
+
+    def __eq__(self, other):
+        return (self.__class__ == other.__class__ and
+                self._milliseconds == other._milliseconds)
+
+    def __str__(self):
+        return "{} ms".format(self._milliseconds)

flink-python/pyflink/datastream/data_stream.py

@@ -1888,7 +1888,7 @@ def side_output_late_data(self, output_tag: OutputTag) -> 'WindowedStream':
 
             >>> tag = OutputTag("late-data", Types.TUPLE([Types.INT(), Types.STRING()]))
             >>> main_stream = ds.key_by(lambda x: x[1]) \\
-            ...                 .window(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            ...                 .window(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...                 .side_output_late_data(tag) \\
             ...                 .reduce(lambda a, b: a[0] + b[0], b[1])
             >>> late_stream = main_stream.get_side_output(tag)
@@ -1918,7 +1918,7 @@ def reduce(self,
         ::
 
             >>> ds.key_by(lambda x: x[1]) \\
-            ...     .window(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            ...     .window(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...     .reduce(lambda a, b: a[0] + b[0], b[1])
 
         :param reduce_function: The reduce function.
@@ -1982,7 +1982,7 @@ def aggregate(self,
             ...     def merge(self, a: Tuple[int, int], b: Tuple[int, int]) -> Tuple[int, int]:
             ...         return a[0] + b[0], a[1] + b[1]
             >>> ds.key_by(lambda x: x[1]) \\
-            ...     .window(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            ...     .window(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...     .aggregate(AverageAggregate(),
             ...                accumulator_type=Types.TUPLE([Types.LONG(), Types.LONG()]),
             ...                output_type=Types.DOUBLE())
@@ -2167,7 +2167,7 @@ def side_output_late_data(self, output_tag: OutputTag) -> 'AllWindowedStream':
         ::
 
             >>> tag = OutputTag("late-data", Types.TUPLE([Types.INT(), Types.STRING()]))
-            >>> main_stream = ds.window_all(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            >>> main_stream = ds.window_all(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...                 .side_output_late_data(tag) \\
             ...                 .process(MyProcessAllWindowFunction(),
             ...                          Types.TUPLE([Types.LONG(), Types.LONG(), Types.INT()]))
@@ -2190,7 +2190,7 @@ def reduce(self,
         Example:
         ::
 
-            >>> ds.window_all(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            >>> ds.window_all(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...   .reduce(lambda a, b: a[0] + b[0], b[1])
 
         :param reduce_function: The reduce function.
@@ -2255,7 +2255,7 @@ def aggregate(self,
             ...     def merge(self, a: Tuple[int, int], b: Tuple[int, int]) -> Tuple[int, int]:
             ...         return a[0] + b[0], a[1] + b[1]
             ...
-            >>> ds.window_all(TumblingEventTimeWindows.of(Duration.of_seconds(5))) \\
+            >>> ds.window_all(TumblingEventTimeWindows.of(Time.seconds(5))) \\
             ...   .aggregate(AverageAggregate(),
             ...              accumulator_type=Types.TUPLE([Types.LONG(), Types.LONG()]),
             ...              output_type=Types.DOUBLE())

flink-python/pyflink/datastream/state.py

@@ -19,7 +19,7 @@
 from enum import Enum
 from typing import TypeVar, Generic, Iterable, List, Iterator, Dict, Tuple, Optional
 
-from pyflink.common.time import Duration
+from pyflink.common.time import Duration, Time
 from pyflink.common.typeinfo import TypeInformation, Types
 
 __all__ = [
@@ -623,7 +623,7 @@ def __init__(self,
                  update_type: UpdateType,
                  state_visibility: StateVisibility,
                  ttl_time_characteristic: TtlTimeCharacteristic,
-                 ttl: Duration,
+                 ttl: Time,
                  cleanup_strategies: 'StateTtlConfig.CleanupStrategies'):
         self._update_type = update_type
         self._state_visibility = state_visibility
@@ -632,7 +632,7 @@ def __init__(self,
         self._cleanup_strategies = cleanup_strategies
 
     @staticmethod
-    def new_builder(ttl: Duration):
+    def new_builder(ttl: Time):
         return StateTtlConfig.Builder(ttl)
 
     def get_update_type(self) -> 'StateTtlConfig.UpdateType':
@@ -641,7 +641,7 @@ def get_update_type(self) -> 'StateTtlConfig.UpdateType':
     def get_state_visibility(self) -> 'StateTtlConfig.StateVisibility':
         return self._state_visibility
 
-    def get_time_to_live(self) -> Duration:
+    def get_ttl(self) -> Time:
         return self._ttl
 
     def get_ttl_time_characteristic(self) -> 'StateTtlConfig.TtlTimeCharacteristic':
@@ -669,7 +669,7 @@ def _from_proto(proto):
         state_visibility = StateTtlConfig.StateVisibility._from_proto(proto.state_visibility)
         ttl_time_characteristic = \
             StateTtlConfig.TtlTimeCharacteristic._from_proto(proto.ttl_time_characteristic)
-        ttl = Duration.of_millis(proto.ttl)
+        ttl = Time.milliseconds(proto.ttl)
         cleanup_strategies = StateTtlConfig.CleanupStrategies._from_proto(proto.cleanup_strategies)
         builder = StateTtlConfig.new_builder(ttl) \
             .set_update_type(update_type) \
@@ -694,7 +694,7 @@ class Builder(object):
         Builder for the StateTtlConfig.
         """
 
-        def __init__(self, ttl: Duration):
+        def __init__(self, ttl: Time):
             self._ttl = ttl
             self._update_type = StateTtlConfig.UpdateType.OnCreateAndWrite
             self._state_visibility = StateTtlConfig.StateVisibility.NeverReturnExpired
@@ -847,7 +847,7 @@ def disable_cleanup_in_background(self) -> 'StateTtlConfig.Builder':
             self._is_cleanup_in_background = False
             return self
 
-        def set_ttl(self, ttl: Duration) -> 'StateTtlConfig.Builder':
+        def set_ttl(self, ttl: Time) -> 'StateTtlConfig.Builder':
             """
             Sets the ttl time.
 

flink-python/pyflink/datastream/tests/test_data_stream.py

@@ -23,7 +23,7 @@
 from typing import Tuple
 
 from pyflink.common import Row, Configuration
-from pyflink.common.time import Duration
+from pyflink.common.time import Time
 from pyflink.common.typeinfo import Types
 from pyflink.common.watermark_strategy import WatermarkStrategy, TimestampAssigner
 from pyflink.datastream import (TimeCharacteristic, RuntimeContext, SlotSharingGroup,
@@ -151,7 +151,7 @@ def open(self, runtime_context: RuntimeContext):
                 self.list_state = runtime_context.get_list_state(list_state_descriptor)
                 map_state_descriptor = MapStateDescriptor('map_state', Types.INT(), Types.STRING())
                 state_ttl_config = StateTtlConfig \
-                    .new_builder(Duration.of_seconds(1)) \
+                    .new_builder(Time.seconds(1)) \
                     .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
                     .set_state_visibility(
                         StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp) \
@@ -268,7 +268,7 @@ def open(self, runtime_context: RuntimeContext):
                 descriptor = AggregatingStateDescriptor(
                     'aggregating_state', MyAggregateFunction(), Types.INT())
                 state_ttl_config = StateTtlConfig \
-                    .new_builder(Duration.of_seconds(1)) \
+                    .new_builder(Time.seconds(1)) \
                     .set_update_type(StateTtlConfig.UpdateType.OnReadAndWrite) \
                     .disable_cleanup_in_background() \
                     .build()

flink-python/pyflink/datastream/tests/test_window.py

@@ -18,7 +18,7 @@
 from typing import Iterable, Tuple, Dict
 
 from pyflink.common import Configuration
-from pyflink.common.time import Duration
+from pyflink.common.time import Time
 from pyflink.common.typeinfo import Types
 from pyflink.common.watermark_strategy import WatermarkStrategy, TimestampAssigner
 from pyflink.datastream.data_stream import DataStream
@@ -59,7 +59,7 @@ def test_event_time_tumbling_window(self):
             .with_timestamp_assigner(SecondColumnTimestampAssigner())
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(TumblingEventTimeWindows.of(Duration.of_millis(5))) \
+            .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) \
             .process(CountWindowProcessFunction(),
                      Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -94,7 +94,7 @@ def test_event_time_sliding_window(self):
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(SlidingEventTimeWindows.of(Duration.of_millis(5), Duration.of_millis(2))) \
+            .window(SlidingEventTimeWindows.of(Time.milliseconds(5), Time.milliseconds(2))) \
             .process(CountWindowProcessFunction(),
                      Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -128,7 +128,7 @@ def test_event_time_session_window(self):
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(5))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(5))) \
             .process(CountWindowProcessFunction(),
                      Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -166,7 +166,7 @@ def test_window_reduce_passthrough(self):
             .with_timestamp_assigner(SecondColumnTimestampAssigner())
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .reduce(lambda a, b: (b[0], a[1] + b[1]),
                     output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -194,7 +194,7 @@ def process(self, key, context: ProcessWindowFunction.Context,
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .reduce(lambda a, b: (b[0], a[1] + b[1]),
                     window_function=MyProcessFunction(),
                     output_type=Types.STRING()) \
@@ -243,7 +243,7 @@ def merge(self, acc_a: Tuple[str, Dict[int, int]], acc_b: Tuple[str, Dict[int, i
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .aggregate(MyAggregateFunction(),
                        output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -276,7 +276,7 @@ def merge(self, acc_a: Tuple[int, str], acc_b: Tuple[int, str]):
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .aggregate(MyAggregateFunction(),
                        accumulator_type=Types.TUPLE([Types.INT(), Types.STRING()]),
                        output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
@@ -317,7 +317,7 @@ def process(self, key: str, context: ProcessWindowFunction.Context,
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .aggregate(MyAggregateFunction(),
                        window_function=MyProcessWindowFunction(),
                        accumulator_type=Types.TUPLE([Types.INT(), Types.STRING()]),
@@ -341,7 +341,7 @@ def test_session_window_late_merge(self):
             .with_timestamp_assigner(SecondColumnTimestampAssigner())
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(5))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(5))) \
             .process(CountWindowProcessFunction(),
                      Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -360,7 +360,7 @@ def test_event_time_session_window_with_purging_trigger(self):
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda x: x[0], key_type=Types.STRING()) \
-            .window(EventTimeSessionWindows.with_gap(Duration.of_millis(3))) \
+            .window(EventTimeSessionWindows.with_gap(Time.milliseconds(3))) \
             .trigger(PurgingTrigger.of(EventTimeTrigger.create())) \
             .process(CountWindowProcessFunction(),
                      Types.TUPLE([Types.STRING(), Types.LONG(), Types.LONG(), Types.INT()])) \
@@ -408,7 +408,7 @@ def test_event_time_tumbling_window_all(self):
             .with_timestamp_assigner(SecondColumnTimestampAssigner())
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
-            .window_all(TumblingEventTimeWindows.of(Duration.of_millis(5))) \
+            .window_all(TumblingEventTimeWindows.of(Time.milliseconds(5))) \
             .process(CountAllWindowProcessFunction(),
                      Types.TUPLE([Types.LONG(), Types.LONG(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -426,7 +426,7 @@ def test_window_all_reduce(self):
         watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
             .with_timestamp_assigner(SecondColumnTimestampAssigner())
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
-            .window_all(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window_all(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .reduce(lambda a, b: (a[0], a[1] + b[1]),
                     output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -454,7 +454,7 @@ def process(self, context: 'ProcessAllWindowFunction.Context',
                 )
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
-            .window_all(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window_all(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .reduce(lambda a, b: (a[0], a[1] + b[1]),
                     window_function=MyProcessFunction(),
                     output_type=Types.STRING()) \
@@ -501,7 +501,7 @@ def merge(self, acc_a: Tuple[str, Dict[int, int]], acc_b: Tuple[str, Dict[int, i
                 return acc_a[0], new_number_map
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
-            .window_all(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window_all(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .aggregate(MyAggregateFunction(),
                        output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \
             .add_sink(self.test_sink)
@@ -540,7 +540,7 @@ def process(self, context: ProcessAllWindowFunction.Context,
                 yield "key {} timestamp sum {}".format(agg_result[0], agg_result[1])
 
         data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
-            .window_all(EventTimeSessionWindows.with_gap(Duration.of_millis(2))) \
+            .window_all(EventTimeSessionWindows.with_gap(Time.milliseconds(2))) \
             .aggregate(MyAggregateFunction(),
                        window_function=MyProcessWindowFunction(),
                        accumulator_type=Types.TUPLE([Types.INT(), Types.STRING()]),
@@ -573,7 +573,7 @@ def test_side_output_late_data(self):
                                        type_info=Types.ROW([Types.STRING(), Types.INT()]))
         ds2 = ds1.assign_timestamps_and_watermarks(watermark_strategy) \
             .key_by(lambda e: e[0]) \
-            .window(TumblingEventTimeWindows.of(Duration.of_millis(5))) \
+            .window(TumblingEventTimeWindows.of(Time.milliseconds(5))) \
             .allowed_lateness(0) \
             .side_output_late_data(tag) \
             .process(CountWindowProcessFunction(),
@@ -619,7 +619,7 @@ def extract_timestamp(self, value: tuple, record_timestamp: int) -> int:
         ds.key_by(
             lambda x: (x[0], x[1], x[2])
         ).window(
-            TumblingEventTimeWindows.of(Duration.of_minutes(1))
+            TumblingEventTimeWindows.of(Time.minutes(1))
         ).reduce(
             lambda x, y: (x[0], x[1], x[2], x[3] + y[3]),
             output_type=Types.TUPLE([Types.LONG(), Types.STRING(), Types.STRING(), Types.INT()])