Jungtaek Lim created SPARK-59315:
------------------------------------
Summary: 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
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]
