[SPARK-58979][SS] Preserve event time through streaming dedup pruning

[nahtonaj]

What changes were proposed in this pull request?

Include event-time watermark attributes in streaming Deduplicate.references, so ColumnPruning cannot remove the event-time column that StreamingDeduplicateExec needs for late-record filtering.

Batch deduplication continues to reference only its key attributes.

This follows the existing DeduplicateWithinWatermark.references pattern added by SPARK-50492.

Why are the changes needed?

Two otherwise equivalent streaming queries currently behave differently depending on their final projection.

When the event-time column is retained:

input.toDF()
  .selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
  .withWatermark("ts", "10 seconds")
  .dropDuplicates("id")
  .selectExpr("id", "CAST(ts AS LONG) AS tsl")

the watermark attribute reaches StreamingDeduplicateExec, and late input is filtered.

When the event-time column is projected away:

input.toDF()
  .selectExpr("_1 AS id", "CAST(_2 AS TIMESTAMP) AS ts")
  .withWatermark("ts", "10 seconds")
  .dropDuplicates("id")
  .select("id")

ColumnPruning removes ts below Deduplicate. The physical deduplication operator can no longer construct its late-event predicate, so the same late row is emitted if it has a new deduplication key.

A downstream projection should not change the internal watermark behavior of streaming deduplication.

JIRA: https://issues.apache.org/jira/browse/SPARK-58979

Related precedent: https://issues.apache.org/jira/browse/SPARK-50492

Does this PR introduce any user-facing change?

Yes.

Streaming dropDuplicates now preserves late-record filtering when the event-time column is not selected downstream. Previously, sufficiently late rows with new deduplication keys could be emitted only because column pruning removed the watermark attribute.

How was this patch tested?

Added paired tests to StreamingDeduplicationSuite using identical input:

• Event-time projected away after dropDuplicates("id").
• Event-time retained above dropDuplicates("id") as the control.

Both tests verify that the same late row is filtered.

Local verification:

• dev/lint-scala: Scalastyle passed.
• dev/lint-scala: Scalafmt passed.
• git diff --check: passed.

The focused SBT suite could not run locally because this host could not resolve Maven/SBT dependencies. GitHub Actions will run the complete test matrix.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: KiroCrew using Databricks GPT-5.6 Sol.

[uros-b]

Seems mechanically correct and minimal to me. Adding @viirya

[viirya]

Verified the mechanism end to end -- the pruning path, and that WatermarkSupport resolves the event-time column by EventTimeWatermark.delayKey on child.output, so pruning that attribute is exactly what silences the late-event predicate. The fix is correct and minimal, and gating on child.isStreaming is right since Deduplicate serves batch dropDuplicates too (the DeduplicateWithinWatermark precedent needs no guard, being streaming-only in practice).

Non-blocking: consider deferring to super.references instead of hardcoding AttributeSet(keys), so a future Expression-typed field on Deduplicate can't be silently dropped from the references:

  override def references: AttributeSet = if (child.isStreaming) {
    // Streaming deduplication filters late rows even when event time is not part of the key.
    super.references ++ AttributeSet(
      child.output.filter(_.metadata.contains(EventTimeWatermark.delayKey)))
  } else {
    super.references
  }

One thing worth calling out in the PR or a test comment: this suite already has test("SPARK-21546: dropDuplicates should ignore watermark when it's not a key"), whose name reads like the opposite of what this change does. It isn't -- that test's rows all arrive in one batch with nothing late, so what it actually pins is that state keying/eviction doesn't use a non-key watermark column (watermarkPredicateForKeys), while this PR only affects late-event filtering (watermarkPredicateForData, which needs the column in child.output). Both stay consistent, but the next person reading these two side by side will have to re-derive that. A sentence noting the keys-vs-data predicate distinction would save that.

I also checked whether other stateful operators have the same exposure, and I don't think they do, so this looks complete rather than one of several instances: FlatMapGroupsWithStateExec and TransformWithStateExec use the same watermarkPredicateForDataForLateEvents, but their logical nodes carry dataAttributes (fed to UnresolvedDeserializer(encoderFor[V].deserializer, ...)), so the whole value row -- event-time column included -- is already referenced and survives pruning. DeduplicateWithinWatermark was covered by SPARK-50492. Deduplicate is the one node that references only its keys, which is why it alone was exposed.

The tests are well chosen and discriminating -- with watermark at 999990s the second row (ts = 1000s) is late and carries a new dedup key, so CheckNewAnswer() would fail without the fix -- and the paired projected-away/retained pair states the invariant nicely. Two small things: please prefix the test names with SPARK-58979: to match the convention in this suite (SPARK-19841, SPARK-21546, SPARK-35896, ...), and note that CI hasn't actually run the matrix here -- the Build check is ACTION_REQUIRED, so combined with the local run being blocked by dependency resolution, these tests haven't executed anywhere yet. Enabling GitHub Actions on your fork (per the Spark contributing guide) so StreamingDeduplicationSuite runs green -- especially the existing SPARK-21546 / SPARK-19841 cases -- would be worth doing before this merges.