Github user shaoxuan-wang commented on a diff in the pull request:
https://github.com/apache/flink/pull/3470#discussion_r104302135
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
---
@@ -36,14 +36,23 @@ abstract class AggFunctionTestBase[T] {
def aggregator: AggregateFunction[T]
+ def ifSupportRetraction: Boolean = true
+
@Test
- // test aggregate functions without partial merge
- def testAggregateWithoutMerge(): Unit = {
+ // test aggregate and retract functions without partial merge
+ def testAccumulateAndRetractWithoutMerge(): Unit = {
// iterate over input sets
for ((vals, expected) <- inputValueSets.zip(expectedResults)) {
- val accumulator = aggregateVals(vals)
- val result = aggregator.getValue(accumulator)
+ val accumulator = accumulateVals(vals)
+ var result = aggregator.getValue(accumulator)
validateResult(expected, result)
+
+ if (ifSupportRetraction) {
+ retractVals(accumulator, vals)
--- End diff --
Regarding to "We also need to check the retraction of a single value not
all values (the SumAggregator retraction of all values was correct because the
count was 0)"
Instead of adding a new test to just retraction one, I think we could just
add the check to compare two accumulators (comparing the getValues is not
sufficient enough anymore).
assertEquals(accumulator, resultAccum)
The accumulator should go back the initial status, after retracting all the
previous inputs. This should be good enough to valid accumulate and retraction.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
