Stephen Hodges created SPARK-60249:
--------------------------------------
Summary: Reusing a named streaming DataFrame in a union fails with
DUPLICATE_SOURCE_NAMES
Key: SPARK-60249
URL: https://issues.apache.org/jira/browse/SPARK-60249
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 4.2.0, 5.0.0
Reporter: Stephen Hodges
With spark.sql.streaming.queryEvolution.enableSourceEvolution=true, reusing a
named streaming data frame in multiple branches of a union causes query startup
to fail with "STREAMING_QUERY_EVOLUTION_ERROR.DUPLICATE_SOURCE_NAMES"
The query contains one source reused twice. Duplicate-name validation counts
both occurrences as *separate* sources, although micro-batch execution would
reuse the same source instance.
To reproduce:
# Write a parquet containing a single long column, id, with rows 1, 2, 2, 3
# Read it once as a named streaming data frame e.g.
val source = spark.readStream.schema("id
LONG").format("parquet").name("events").load(input)
# Create a second branch from that same frame, e.g.
val transformed = source.filter("id > 1").selectExpr("id + 10 AS id")
# Combine them like
val combined = source.unionByName(transformed)
# Start an append-mode memory sink with a query name and a fresh checkpoint
directory like
combined.writeStream.format("memory").queryName("named_source_reuse_result").outputMode("append").option("checkpointLocation",
checkpoint).start()
Actual result is that query startup throws AnalysisException with error class
"STREAMING_QUERY_EVOLUTION_ERROR.DUPLICATE_SOURCE_NAMES"
Expected result is that the query starts and uses one source, so after
processAllAvailable() the output contains 1, 2, 2, 3, 12, 12, 13.
I believe that the cause is that ResolveWriteToStream.checkDuplicateSourceNames
collects source names from every matching relation occurrence and rejects
repeated names without accounting for relation reuse.
MicroBatchExecution caches sources by full relation equality within each reader
route. The validation therefore rejects a reuse pattern that execution actually
supports.
Validation should IMO distinguish reuse of the same relation from independent
sources assigned the same name. Independent sources with actual duplicate names
should continue to fail.
Issue reproduced and possible fix tested on local source builds of Spark
4.2.0-rc6 and 5.0.0-SNAPSHOT
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
