[hotfix][python] Fix wrong implementation in of __le__ in MemorySize and TimeWindow

This closes #27240.

Author: bgeng777

flink-python/pyflink/datastream/slot_sharing_group.py

@@ -99,7 +99,7 @@ def __lt__(self, other: 'MemorySize'):
         return self._j_memory_size.compareTo(other._j_memory_size) == -1
 
     def __le__(self, other: 'MemorySize'):
-        return self.__eq__(other) and self.__lt__(other)
+        return self.__eq__(other) or self.__lt__(other)
 
     def __str__(self):
         return self._j_memory_size.toString()

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

@@ -64,3 +64,18 @@ def test_build_slot_sharing_group_without_all_required_config(self):
                 .set_cpu_cores(1.0) \
                 .set_task_off_heap_memory_mb(10) \
                 .build()
+
+
+class MemorySizeTests(PyFlinkTestCase):
+
+    def test_le_method(self):
+        """Test the __le__ method of MemorySize."""
+        m1 = MemorySize.of_mebi_bytes(100)
+        m2 = MemorySize.of_mebi_bytes(100)
+        m3 = MemorySize.of_mebi_bytes(200)
+        self.assertEqual(m1, m2)
+        self.assertTrue(m1 <= m2)
+        self.assertTrue(m2 <= m1)
+
+        self.assertTrue(m1 <= m3)
+        self.assertFalse(m3 <= m1)

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

@@ -33,7 +33,7 @@
 from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction, \
     SecondColumnTimestampAssigner
 from pyflink.java_gateway import get_gateway
-from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
+from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase, PyFlinkTestCase
 from pyflink.util.java_utils import get_j_env_configuration
 
 
@@ -668,3 +668,27 @@ def process(self,
                 context: 'ProcessAllWindowFunction.Context',
                 elements: Iterable[tuple]) -> Iterable[tuple]:
         return [(context.window().start, context.window().end, len([e for e in elements]))]
+
+
+class TestTimeWindow(PyFlinkTestCase):
+
+    def test_le_method(self):
+        """Test the __le__ method of TimeWindow."""
+        # Create test windows
+        w1 = TimeWindow(100, 200)
+        w2 = TimeWindow(100, 200)
+        w3 = TimeWindow(150, 250)
+        w4 = TimeWindow(50, 150)
+        w5 = TimeWindow(100, 180)
+
+        self.assertTrue(w1 <= w2)
+        self.assertTrue(w2 <= w1)
+
+        self.assertTrue(w1 <= w3)
+        self.assertFalse(w3 <= w1)
+
+        self.assertTrue(w4 <= w1)
+        self.assertFalse(w1 < w4)
+
+        self.assertTrue(w5 <= w1)
+        self.assertFalse(w1 <= w5)

flink-python/pyflink/datastream/window.py

@@ -171,7 +171,7 @@ def __lt__(self, other: 'TimeWindow'):
         return self.start == other.start and self.end < other.end or self.start < other.start
 
     def __le__(self, other: 'TimeWindow'):
-        return self.__eq__(other) and self.__lt__(other)
+        return self.__eq__(other) or self.__lt__(other)
 
     def __repr__(self):
         return "TimeWindow(start={}, end={})".format(self.start, self.end)