This is an automated email from the ASF dual-hosted git repository.
hxb pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push:
new c9995a9 [FLINK-24317][python][tests] Optimize the implementation of
Top2 in test_flat_aggregate
c9995a9 is described below
commit c9995a9566ee16e7d77f898330ccccd92efb4439
Author: huangxingbo <[email protected]>
AuthorDate: Fri Sep 17 19:15:22 2021 +0800
[FLINK-24317][python][tests] Optimize the implementation of Top2 in
test_flat_aggregate
---
.../table/tests/test_row_based_operation.py | 25 +++++++++-------------
1 file changed, 10 insertions(+), 15 deletions(-)
diff --git a/flink-python/pyflink/table/tests/test_row_based_operation.py
b/flink-python/pyflink/table/tests/test_row_based_operation.py
index a2b78be..b8e0771 100644
--- a/flink-python/pyflink/table/tests/test_row_based_operation.py
+++ b/flink-python/pyflink/table/tests/test_row_based_operation.py
@@ -340,27 +340,22 @@ class CountAndSumAggregateFunction(AggregateFunction):
class Top2(TableAggregateFunction):
def emit_value(self, accumulator):
- yield accumulator[0]
- yield accumulator[1]
+ accumulator.sort()
+ accumulator.reverse()
+ size = len(accumulator)
+ if size > 1:
+ yield accumulator[0]
+ if size > 2:
+ yield accumulator[1]
def create_accumulator(self):
- return [None, None]
+ return []
def accumulate(self, accumulator, *args):
- if args[0][0] is not None:
- if accumulator[0] is None or args[0][0] > accumulator[0]:
- accumulator[1] = accumulator[0]
- accumulator[0] = args[0][0]
- elif accumulator[1] is None or args[0][0] > accumulator[1]:
- accumulator[1] = args[0][0]
+ accumulator.append(args[0][0])
def retract(self, accumulator, *args):
- accumulator[0] = accumulator[0] - 1
-
- def merge(self, accumulator, accumulators):
- for other_acc in accumulators:
- self.accumulate(accumulator, other_acc[0])
- self.accumulate(accumulator, other_acc[1])
+ accumulator.remove(args[0][0])
def get_accumulator_type(self):
return DataTypes.ARRAY(DataTypes.BIGINT())
