[
https://issues.apache.org/jira/browse/SPARK-60249?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127224#comment-18127224
]
Stephen Hodges commented on SPARK-60249:
----------------------------------------
I’m working on a fix and regression tests and plan to open a linked PR
> 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
> Priority: Major
>
> 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]
