[
https://issues.apache.org/jira/browse/SPARK-59663?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Eric Smith updated SPARK-59663:
-------------------------------
Environment:
Image: docker.io/apache/spark:4.2.0
Spark Version: 4.2.0
Client: /opt/spark/bin/spark-pipelines
Server: start-connect-server.sh --conf spark.sql.warehouse.dir=/tmp/wh --conf
spark.sql.catalogImplementation=in-memory
Format: default parquet
reproduction
`spark-pipeline.yaml`:
```yaml
name: edge_drop_repro
storage: [file:///tmp/sdp-jira-abc-storage]
catalog: spark_catalog
database: repro
libraries:
- glob:
include: transformations/**
```
`transformations/pipeline.py`:
```python
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.materialized_view
def a():
return spark.sql("SELECT * FROM VALUES (1, 'north'), (2, 'south') AS t(id,
name)")
@dp.materialized_view
def b():
# withColumn -> proto WithColumns -> SparkConnectPlanner.transformWithColumns
return spark.read.table("a").withColumn("m", F.upper(F.col("name")))
@dp.materialized_view
def c():
# CONTROL: select -> proto Project -> transformProject, no eager analysis
return spark.read.table("a").select("id", "name",
F.upper(F.col("name")).alias("m"))
```
```bash
# the database must pre-exist: spark.sql("CREATE DATABASE IF NOT EXISTS
spark_catalog.repro")
/opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
spark-pipeline.yaml # run 1
/opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
spark-pipeline.yaml # run 2
```
*Expected:* run 2 identical to run 1 — `b` waits for `a`, then holds 2 rows.
*Run 1 (clean catalog) — correct.* `TriggeredGraphExecution` ordering:
```
26/09/19 19:28:12 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`a`
26/09/19 19:28:14 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`c`
26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`b`
26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
```
Rows: `a`=2, `b`=2, `c`=2.
```
ROWS a count=2
ROW a Row(id=1, name='north')
ROW a Row(id=2, name='south')
ROWS b count=2
ROW b Row(id=1, name='north', m='NORTH')
ROW b Row(id=2, name='south', m='SOUTH')
ROWS c count=2
ROW c Row(id=1, name='north', m='NORTH')
ROW c Row(id=2, name='south', m='SOUTH')
```
*Run 2 (identical command) — `b` loses its edge:*
```
26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`b`
26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`a`
26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:34 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`c`
26/09/19 19:28:35 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
```
Rows: `a`=2, *{*}`b`=0{*}*, `c`=2.
```
ROWS a count=2
ROW a Row(id=1, name='north')
ROW a Row(id=2, name='south')
ROWS b count=0
ROWS c count=2
ROW c Row(id=1, name='north', m='NORTH')
ROW c Row(id=2, name='south', m='SOUTH')
```
was:
Image: docker.io/apache/spark:4.2.0
Spark Version: 4.2.0
Client: /opt/spark/bin/spark-pipelines
Server: start-connect-server.sh --conf spark.sql.warehouse.dir=/tmp/wh --conf
spark.sql.catalogImplementation=in-memory
Format: default parquet
reproduction
`spark-pipeline.yaml`:
```yaml
name: edge_drop_repro
storage: file:///tmp/sdp-jira-abc-storage
catalog: spark_catalog
database: repro
libraries:
- glob:
include: transformations/**
```
`transformations/pipeline.py`:
```python
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.materialized_view
def a():
return spark.sql("SELECT * FROM VALUES (1, 'north'), (2, 'south') AS t(id,
name)")
@dp.materialized_view
def b():
# withColumn -> proto WithColumns -> SparkConnectPlanner.transformWithColumns
return spark.read.table("a").withColumn("m", F.upper(F.col("name")))
@dp.materialized_view
def c():
# CONTROL: select -> proto Project -> transformProject, no eager analysis
return spark.read.table("a").select("id", "name",
F.upper(F.col("name")).alias("m"))
```
```bash
# the database must pre-exist: spark.sql("CREATE DATABASE IF NOT EXISTS
spark_catalog.repro")
/opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
spark-pipeline.yaml # run 1
/opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
spark-pipeline.yaml # run 2
```
**Expected:** run 2 identical to run 1 — `b` waits for `a`, then holds 2 rows.
**Run 1 (clean catalog) — correct.** `TriggeredGraphExecution` ordering:
```
26/09/19 19:28:12 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`a`
26/09/19 19:28:14 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`c`
26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`b`
26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
```
Rows: `a`=2, `b`=2, `c`=2.
```
ROWS a count=2
ROW a Row(id=1, name='north')
ROW a Row(id=2, name='south')
ROWS b count=2
ROW b Row(id=1, name='north', m='NORTH')
ROW b Row(id=2, name='south', m='SOUTH')
ROWS c count=2
ROW c Row(id=1, name='north', m='NORTH')
ROW c Row(id=2, name='south', m='SOUTH')
```
**Run 2 (identical command) — `b` loses its edge:**
```
26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`b`
26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`a`
26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
26/09/19 19:28:34 INFO TriggeredGraphExecution: Starting flow
`spark_catalog`.`repro`.`c`
26/09/19 19:28:35 INFO TriggeredGraphExecution: Flow
`spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
```
Rows: `a`=2, **`b`=0**, `c`=2.
```
ROWS a count=2
ROW a Row(id=1, name='north')
ROW a Row(id=2, name='south')
ROWS b count=0
ROWS c count=2
ROW c Row(id=1, name='north', m='NORTH')
ROW c Row(id=2, name='south', m='SOUTH')
```
> SDP: `withColumn` on an in-graph table read silently drops the dependency
> edge once the upstream table exists
> -------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-59663
> URL: https://issues.apache.org/jira/browse/SPARK-59663
> Project: Spark
> Issue Type: Bug
> Components: Declarative Pipelines
> Affects Versions: 4.1.0, 4.2.0, 4.1.1, 4.1.2, 4.1.3
> Environment: Image: docker.io/apache/spark:4.2.0
> Spark Version: 4.2.0
> Client: /opt/spark/bin/spark-pipelines
> Server: start-connect-server.sh --conf spark.sql.warehouse.dir=/tmp/wh --conf
> spark.sql.catalogImplementation=in-memory
> Format: default parquet
>
> reproduction
> `spark-pipeline.yaml`:
> ```yaml
> name: edge_drop_repro
> storage: [file:///tmp/sdp-jira-abc-storage]
> catalog: spark_catalog
> database: repro
> libraries:
> - glob:
> include: transformations/**
> ```
> `transformations/pipeline.py`:
> ```python
> from pyspark import pipelines as dp
> from pyspark.sql import functions as F
> @dp.materialized_view
> def a():
> return spark.sql("SELECT * FROM VALUES (1, 'north'), (2, 'south') AS t(id,
> name)")
> @dp.materialized_view
> def b():
> # withColumn -> proto WithColumns -> SparkConnectPlanner.transformWithColumns
> return spark.read.table("a").withColumn("m", F.upper(F.col("name")))
> @dp.materialized_view
> def c():
> # CONTROL: select -> proto Project -> transformProject, no eager analysis
> return spark.read.table("a").select("id", "name",
> F.upper(F.col("name")).alias("m"))
> ```
> ```bash
> # the database must pre-exist: spark.sql("CREATE DATABASE IF NOT EXISTS
> spark_catalog.repro")
> /opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
> spark-pipeline.yaml # run 1
> /opt/spark/bin/spark-pipelines run --remote sc://localhost:15002 --spec
> spark-pipeline.yaml # run 2
> ```
> *Expected:* run 2 identical to run 1 — `b` waits for `a`, then holds 2 rows.
> *Run 1 (clean catalog) — correct.* `TriggeredGraphExecution` ordering:
> ```
> 26/09/19 19:28:12 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`a`
> 26/09/19 19:28:14 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
> 26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`c`
> 26/09/19 19:28:14 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`b`
> 26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
> 26/09/19 19:28:15 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
> ```
> Rows: `a`=2, `b`=2, `c`=2.
> ```
> ROWS a count=2
> ROW a Row(id=1, name='north')
> ROW a Row(id=2, name='south')
> ROWS b count=2
> ROW b Row(id=1, name='north', m='NORTH')
> ROW b Row(id=2, name='south', m='SOUTH')
> ROWS c count=2
> ROW c Row(id=1, name='north', m='NORTH')
> ROW c Row(id=2, name='south', m='SOUTH')
> ```
> *Run 2 (identical command) — `b` loses its edge:*
> ```
> 26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`b`
> 26/09/19 19:28:33 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`a`
> 26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`b` has COMPLETED in TriggeredFlowExecution.
> 26/09/19 19:28:34 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`a` has COMPLETED in TriggeredFlowExecution.
> 26/09/19 19:28:34 INFO TriggeredGraphExecution: Starting flow
> `spark_catalog`.`repro`.`c`
> 26/09/19 19:28:35 INFO TriggeredGraphExecution: Flow
> `spark_catalog`.`repro`.`c` has COMPLETED in TriggeredFlowExecution.
> ```
> Rows: `a`=2, *{*}`b`=0{*}*, `c`=2.
> ```
> ROWS a count=2
> ROW a Row(id=1, name='north')
> ROW a Row(id=2, name='south')
> ROWS b count=0
> ROWS c count=2
> ROW c Row(id=1, name='north', m='NORTH')
> ROW c Row(id=2, name='south', m='SOUTH')
> ```
> Reporter: Eric Smith
> Priority: Major
>
> `SparkConnectPlanner.transformWithColumns` analyzes its child eagerly, and
> `PipelinesHandler.defineFlow` runs a flow's relation through the planner at
> `DefineFlow` time.
> Once the upstream table exists in the catalog, the eager analysis succeeds,
> consumes the `UnresolvedRelation`, and `FlowAnalysis` (which discovers a
> flow's dependencies only by matching UnresolvedRelation) records no
> dependency. The downstream flow is then scheduled concurrently with its own
> upstream
> The first run of a pipeline is always correct. The eager analysis fails on a
> clean catalog and the `catch` preserves the read. So ths passes every smoke
> test and fails on the second scheduled run.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
