wayne-x-wang commented on issue #7170:
URL: https://github.com/apache/kyuubi/issues/7170#issuecomment-6261767373
<p><span>I reproduced this on Kyuubi master at commit </span><code
dir="ltr">771be6a18</code><span>, using Spark 3.5.8, Scala 2.12.19, and JDK
17.</span></p><p><span>The results suggest that we need to distinguish
</span><code dir="ltr">QueryExecution.id</code><span>, the SQL </span><code
dir="ltr">executionId</code><span>, and </span><code
dir="ltr">rootExecutionId</code><span>.</span></p><p><strong><span>1. Reusing
one DataFrame for two actions</span></strong></p><pre dir="ltr"><code>val df =
spark.sql("SELECT 7170 AS lineage_execution_id")
df.collect()
df.collect()</code></pre><p><span>The captured events were:</span></p><pre
dir="ltr"><code>QueryExecution.id: 0
SQL Start/End execution IDs: [0, 1]
Lineage event execution IDs: [0, 0]</code></pre><p><span>One </span><code
dir="ltr">QueryExecution</code><span> can therefore correspond to multiple SQL
executions. A simple </span><code dir="ltr">qe.id ->
executionId</code><span> mapping would not represent this relationship
reliably.</span></p><p><strong><span>2. CTAS and
rootExecutionId</span></strong></p><p><span>I also tested:</span></p><pre
dir="ltr"><code>CREATE TABLE lineage_ctas_source (id INT) USING parquet;
INSERT INTO lineage_ctas_source VALUES (1), (2);
CREATE TABLE lineage_ctas_target USING parquet
AS SELECT id FROM lineage_ctas_source;</code></pre><p><span>The listener was
registered before the CTAS </span><code dir="ltr">spark.sql()</code><span>
call, since the command executes eagerly. The captured values
were:</span></p><div><div><div>
Execution | End.qe.id | SQL executionId | Start.rootExecutionId | Lineage
executionId
-- | -- | -- | -- | --
CreateDataSourceTableAsSelectCommand | 5 | 2 | 2 | 5
InsertIntoHadoopFsRelationCommand | 7 | 3 | 2 | 7
</div></div></div><p><span>This confirms the relationship mentioned in
SPARK-41752: the outer and nested executions have different execution IDs but
share the same root execution ID. However, </span><code
dir="ltr">rootExecutionId</code><span> does not resolve the mismatch between
</span><code dir="ltr">qe.id</code><span> and the SQL execution
ID.</span></p><p><span>The target table contained the expected rows
</span><code dir="ltr">[1, 2]</code><span>. Both regression cases fail on the
execution-ID assertions, while the three existing tests in </span><code
dir="ltr">OperationLineageEventSuite</code><span>
pass.</span></p><p><strong><span>Proposed
scope</span></strong></p><p><span>Process </span><code
dir="ltr">SparkListenerSQLExecutionEnd</code><span> directly and pass its
actual </span><code dir="ltr">executionId</code><span>, together with its
</span><code dir="ltr">qe</code><span>, to the lineage dispatchers. This would
preserve the existing lineage parser and event fields whi
le correcting the IDs emitted through </span><code
dir="ltr">SPARK_EVENT</code><span> and </span><code
dir="ltr">KYUUBI_EVENT</code><span>.</span></p><p><span>Root-level grouping
could remain a separate concern: consumers can correlate the corrected
execution ID with </span><code
dir="ltr">SparkListenerSQLExecutionStart.rootExecutionId</code><span>.</span></p><p><strong><span>Compatibility
questions</span></strong></p><p><span>The current entry point is registered
through </span><code dir="ltr">spark.sql.queryExecutionListeners</code><span>,
whose callbacks do not expose the SQL execution ID. An additional </span><code
dir="ltr">SparkListener</code><span> would typically use </span><code
dir="ltr">spark.extraListeners</code><span>.</span></p><p><span>Would
introducing this new entry point, retaining/deprecating the existing
QueryExecutionListener entry point, be acceptable? Or would you prefer
preserving the current configuration contract?</span></p><p><span>The
implementation would
also need to address duplicate emission when both entry points are
configured, session scope, failure events, and replayed events without runtime
</span><code dir="ltr">qe</code><span> metadata. Access to the runtime
End-event fields would require a version-tested adapter because those fields
are </span><code
dir="ltr">private[sql]</code><span>.</span></p><p><span>Separately, the nested
CTAS write produced </span><code dir="ltr">lineage=None</code><span>: its
</span><code dir="ltr">catalogTable</code><span> was absent and the parser
logged a </span><code dir="ltr">MatchError</code><span>. The outer CTAS event
contained the expected table and column lineage. I would keep that parser issue
separate from the execution-ID correction.</span></p><p><span>No production fix
has been implemented yet. I would appreciate feedback on the proposed direction
and configuration compatibility before proceeding.</span></p>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
