[
https://issues.apache.org/jira/browse/SPARK-59315?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jungtaek Lim reassigned SPARK-59315:
------------------------------------
Assignee: Jungtaek Lim
> New resolution logic for dropDuplicates(WithinWatermark) introduced
> correctness issue with metadata column
> ----------------------------------------------------------------------------------------------------------
>
> Key: SPARK-59315
> URL: https://issues.apache.org/jira/browse/SPARK-59315
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 4.3.0
> Reporter: Jungtaek Lim
> Assignee: Jungtaek Lim
> Priority: Blocker
> Labels: pull-request-available
>
> https://issues.apache.org/jira/browse/SPARK-57489 changed the approach of
> resolving dropDuplicates(WithinWatermark) operator. While the new resolution
> fixed the issue of Scala/Spark version upgrade and Spark Classic vs Spark
> Connect interop, we figured that the change introduced a correctness issue.
> The issue happens with below query:
> {code:java}
> import java.nio.file.Files
> import org.apache.spark.sql.functions._
> val dir = Files.createTempDirectory("dedup-metadata")
> Files.writeString(dir.resolve("first.txt"), "same")
> Files.writeString(dir.resolve("second.txt"), "same")
> val result = spark.readStream
> .format("text")
> .load(dir.toString)
> .dropDuplicates()
> .select(
> $"value",
> $"_metadata.file_path".as("file_path"))
> val query = result.writeStream
> .format("memory")
> .queryName("dedup_metadata")
> .outputMode("append")
> .start()
> query.processAllAvailable()
> spark.table("dedup_metadata").show(false)
> query.stop() {code}
> Before SPARK-57489, the metadata column (_metadata) wasn't counted toward the
> dedup keys (subset), hence the query produced only one row for "same" and the
> file_path is referring to the first occurrence of the duplications.
> After SPARK-57489, the metadata column (_metadata) is counted, hence the
> query produces "two" rows for "same" with different file_path. The semantic
> is slightly confusing here, but mostly the intention would be just referring
> to "value" column for dedup keys. Users can add the _metadata column into
> dedup keys via applying projection before deduplication, so there is even a
> way to achieve the same outcome with intention.
> Since this is a correctness issue and we are yet to release the issue, we
> should consider this issue as a blocker.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
