[SPARK-11812] [PySpark] invFunc=None works properly with python's reduceByKeyAndWindow

[dtolpin]

invFunc is optional and can be None. Instead of invFunc (the parameter) invReduceFunc (a local function) was checked for trueness (that is, not None, in this context). A local function is never None,
thus the case of invFunc=None (a common one when inverse reduction is not defined) was treated incorrectly, resulting in loss of data.

In addition, the docstring used wrong parameter names, also fixed.

[andrewor14]

@tdas @dtolpin can you file a JIRA attached to the title of this PR? See https://cwiki.apache.org/confluence/display/SPARK/Contributing+to+Spark

[zsxwing]

@dtolpin could you add a unit test for this fix?

[dtolpin]

I don't see how I can make a meaningful unit test for this bug, unfortunately. That would require that I use a time-based window with reduceByKeyAndWindow, and fixtures provided with unittests for python streaming do not support this.

However, this PR just changes checking a constant (which is meaningless) for trueness to checking a parameter. I understand that everything must be covered by unit tests, but the way the code is written it is an objective which is very difficult to achieve. Modifying the code so that it is unit-testable is a major rewrite, which won't have time to do.

[zsxwing]

@dtolpin could you add a summary to the PR title?

[dtolpin]

Sure, I have just put the summary back into the PR title.

[zsxwing]

@dtolpin could you add the following test to WindowFunctionTests in python/pyspark/streaming/tests.py? It should reproduce the issue.

    def test_reduce_by_key_and_window_with_none_invFunc(self):
        input = [range(1), range(2), range(3), range(4), range(5), range(6)]

        def func(dstream):
            return dstream.map(lambda x: (x, 1))\
                .reduceByKeyAndWindow(operator.add, None, 5, 1)\
                .filter(lambda kv: kv[1] > 0).count()

        expected = [[2], [4], [6], [6], [6], [6]]
        self._test_func(input, func, expected)

[dtolpin]

Got it, added it, thank you.

[zsxwing]

Jenkins, test this please

[zsxwing]

@andrewor14 could you help trigger Jenkins? Thanks. I think I don't have the permission :(

[andrewor14]

that's odd, you should have permission. retest this please

[zsxwing]

this is ok to test?

[zsxwing]

retest this please

[SparkQA]

Test build #46351 has finished for PR 9775 at commit e76be01.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zsxwing]

LGTM

[tdas]

Cool, I will merge it in master, 1.6 and 1.5