shrirangmhalgi commented on PR #56578:
URL: https://github.com/apache/spark/pull/56578#issuecomment-4856784201
@AnishMahto Thanks for the detailed analysis and apologies for the late
reply. - I agree with your approach. I traced through the code paths and your
diagnosis is correct.
### What I verified:
1. `spark.table("x")` **is NOT the issue** - it produces a `NamedTable` Read
proto → `transformReadRel` (line 1663 returns
`UnresolvedRelation(multipartIdentifier`) → FlowAnalysis already intercepts
this. Confirmed the `JIRA's spark.table()` repro claim is inaccurate.
2. `spark.read.format(...).load(identifier)` **IS the issue** - it produces
a `DATA_SOURCE` Read proto → `transformReadRel` line 1690 calls
`reader.load().queryExecution.analyzed` → plan arrives fully resolved →
`FlowAnalysis.analyze` never sees an `UnresolvedRelation` → dependency
invisible to the DAG scheduler.
3. **Existing pipeline infrastructure confirms your design direction:**
- `PipelineAnalysisContextUtils.hasPipelineAnalysisContext(userContext)` is
already used at line 3028 to block unsupported SQL commands in pipeline context
- `isInsidePipelineFlowFunction(userContext)` at line 3034 defers
`spark.sql()` analysis - same pattern we'd apply to `DATA_SOURCE`
- `SparkConnectPlanExecution` line 73 already throws
`ATTEMPT_ANALYSIS_IN_PIPELINE_QUERY_FUNCTION` for unsupported eager execution
inside flows
- `userContext` is accessible in `transformReadRel` via
`executeHolderOpt.get.request.getUserContext`
4. **Why the planner-level approach is better:**
- **Complete** - intercepts at the source rather than pattern-matching a
subset of resolved relation types
- **Prevents the eager analysis failure** - if the upstream table doesn't
exist yet, `reader.load()` throws `TABLE_OR_VIEW_NOT_FOUND` before we can even
track the dependency; deferring analysis avoids this entirely
- **Single interception point** - no maintenance burden from new relation
types
### Design question before I implement:
The `DATA_SOURCE` proto's `paths` field can be either a table identifier
(e.g., `catalog.db.table`) or a file path (e.g., `/data/parquet/`). When
`isInsidePipelineFlowFunction` is true, should I:
(a) Return an `UnresolvedRelation(paths)` unconditionally and let
`FlowAnalysis` determine whether it matches a known pipeline dataset
(non-matching identifiers would fail at analysis time as usual), or
(b) Only defer analysis for cases where the path looks like a multipart
identifier (no `/`, no scheme like `s3://`), and eagerly analyze file-path
loads as before?
I'm leaning toward (a) since `FlowAnalysis` already handles the "not a
pipeline dataset" case by letting it through for normal resolution. But wanted
your input before proceeding.
Happy to rework on the PR with the `SparkConnectPlanner`-level fix once we
confirm the direction.
--
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]
