[SPARK-19841][SS]watermarkPredicate should filter based on keys

[zsxwing]

What changes were proposed in this pull request?

StreamingDeduplicateExec.watermarkPredicate should filter based on keys. Otherwise, it may generate a wrong answer if the watermark column in keyExpression has a different position in the row.

StateStoreSaveExec has the same codes but its parent can makes sure the watermark column positions in keyExpression and row are the same.

How was this patch tested?

The added test.

[SparkQA]

Test build #74054 has finished for PR 17183 at commit c74a2d8.

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

[marmbrus]

This is very minor, but maybe we should just bind a different predicate to the child.output schema rather than have two projections here.

Perhaps the base class should have both watermarkExpression and keyPredicate to allow for this flexibility.

[marmbrus]

LGTM

[SparkQA]

Test build #74151 has finished for PR 17183 at commit 89679f4.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• trait WatermarkSupport extends UnaryExecNode

[zsxwing]

retest this please

[SparkQA]

Test build #74168 has finished for PR 17183 at commit 89679f4.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• trait WatermarkSupport extends UnaryExecNode

[zsxwing]

Thanks! Merging to master.