SPARK-1171: when executor is removed, we should minus totalCores instead of just freeCores on that executor by CodingCat · Pull Request #63 · apache/spark

CodingCat · 2014-03-03T04:30:27Z

https://spark-project.atlassian.net/browse/SPARK-1171

When the executor is removed, the current implementation will only minus the freeCores of that executor. Actually we should minus the totalCores...

AmplabJenkins · 2014-03-03T04:30:51Z

Can one of the admins verify this patch?

rxin · 2014-03-03T07:27:53Z

Jenkins, add to whitelist.

AmplabJenkins · 2014-03-03T07:28:24Z

Merged build triggered.

AmplabJenkins · 2014-03-03T07:28:24Z

Merged build started.

AmplabJenkins · 2014-03-03T07:30:01Z

Merged build finished.

AmplabJenkins · 2014-03-03T07:30:01Z

One or more automated tests failed
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/12964/

AmplabJenkins · 2014-03-03T15:22:24Z

Merged build triggered.

AmplabJenkins · 2014-03-03T15:22:24Z

Merged build started.

AmplabJenkins · 2014-03-03T16:21:23Z

Merged build finished.

AmplabJenkins · 2014-03-03T16:21:23Z

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/12971/

kayousterhout · 2014-03-03T21:36:06Z

Why does this need to be transient? also use camelcase for naming (totalCores)

Actually on second thought, can CoarseGrainedSchedulerBackend just store the total cores for each worker in a hash map? I'd prefer that solution since other classes use WorkerOffer and don't use it to keep track of the total cores on each worker.

+1 I'd also like to see WorkerOffer remain more like an immutable message type, with derived, mutable structures created only locally within the implementations that need it.

On that note, it seems to me that WorkerOffer should just be a case class, since all the constructor parameters are public vals anyway.

AmplabJenkins · 2014-03-04T04:22:22Z

Merged build triggered.

AmplabJenkins · 2014-03-04T04:22:22Z

Merged build started.

AmplabJenkins · 2014-03-04T05:22:18Z

Merged build finished.

AmplabJenkins · 2014-03-04T05:22:18Z

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/12982/

CodingCat · 2014-03-04T05:30:58Z

@kayousterhout @markhamstra @andrewor14 Thank you for your comments,

I updated the code, how about this?

markhamstra · 2014-03-04T06:41:53Z

Now that WorkerOffer is a case class, you can do this and the one in makeOffers with the copy idiom:
workerOffers.keys.foreach { executorId => workerOffers(executorId) = workerOffers(executorId).copy(cores = freeCores(executorId)) }

AmplabJenkins · 2014-03-04T13:24:27Z

Merged build triggered.

AmplabJenkins · 2014-03-04T13:24:27Z

Merged build started.

AmplabJenkins · 2014-03-04T14:23:01Z

Merged build finished.

AmplabJenkins · 2014-03-04T14:23:02Z

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/12994/

markhamstra · 2014-03-04T16:38:39Z

superfluous ';'

oops, sorry, fixed

AmplabJenkins · 2014-03-04T17:23:46Z

Merged build triggered.

AmplabJenkins · 2014-03-04T17:23:47Z

Merged build started.

kayousterhout · 2014-03-04T17:54:08Z

This new version of the change doesn't look any simpler to me than the current version of the code and I think is a slightly confusing way of using worker offers to store info about the executors. Can you just remove executorAddress, the unused variable, and fix the bug, but keep the original way of generating worker offers?

AmplabJenkins · 2014-03-04T18:23:12Z

Merged build finished.

AmplabJenkins · 2014-03-04T18:25:33Z

Merged build triggered.

AmplabJenkins · 2014-03-04T18:25:33Z

Merged build started.

AmplabJenkins · 2014-03-04T19:23:14Z

Merged build finished.

AmplabJenkins · 2014-03-04T19:23:14Z

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/12996/

CodingCat · 2014-03-05T01:59:31Z

How about this?

kayousterhout · 2014-03-05T22:00:56Z

This looks good -- I've merged this into master.

CodingCat · 2014-03-05T22:06:19Z

@kayousterhout Thank you very much!

Fixing spark streaming example and a bug in examples build. - Examples assembly included a log4j.properties which clobbered Spark's - Example had an error where some classes weren't serializable - Did some other clean-up in this example (cherry picked from commit 28e9c2a) Signed-off-by: Patrick Wendell <pwendell@gmail.com>

…ead of just freeCores on that executor https://spark-project.atlassian.net/browse/SPARK-1171 When the executor is removed, the current implementation will only minus the freeCores of that executor. Actually we should minus the totalCores... Author: CodingCat <zhunansjtu@gmail.com> Author: Nan Zhu <CodingCat@users.noreply.github.com> Closesapache#63 from CodingCat/simplify_CoarseGrainedSchedulerBackend and squashes the following commits: f6bf93f [Nan Zhu] code clean 19c2bb4 [CodingCat] use copy idiom to reconstruct the workerOffers 43c13e9 [CodingCat] keep WorkerOffer immutable af470d3 [CodingCat] style fix 0c0e409 [CodingCat] simplify the implementation of CoarseGrainedSchedulerBackend

Catch up with branch-1.4 bug fixes and bump jersey

"Fix" timestamp parsing and update spark.

…mance (apache#63)

Implement a Go Spark CLI

## What changes were proposed in this pull request? In DataSourceV2Strategy, it seems we eliminate the subqueries by mistake after normalizing filters. We have a sql with a scalar subquery: ``` scala val plan = spark.sql("select * from t2 where t2a > (select max(t1a) from t1)") plan.explain(true) ``` And we get the log info of DataSourceV2Strategy: ``` Pushing operators to csv:examples/src/main/resources/t2.txt Pushed Filters: Post-Scan Filters: isnotnull(t2a#30) Output: t2a#30, t2b#31 ``` The `Post-Scan Filters` should contain the scalar subquery, but we eliminate it by mistake. ``` == Parsed Logical Plan == 'Project [*] +- 'Filter ('t2a > scalar-subquery#56 []) : +- 'Project [unresolvedalias('max('t1a), None)] : +- 'UnresolvedRelation `t1` +- 'UnresolvedRelation `t2` == Analyzed Logical Plan == t2a: string, t2b: string Project [t2a#30, t2b#31] +- Filter (t2a#30 > scalar-subquery#56 []) : +- Aggregate [max(t1a#13) AS max(t1a)#63] : +- SubqueryAlias `t1` : +- RelationV2[t1a#13, t1b#14] csv:examples/src/main/resources/t1.txt +- SubqueryAlias `t2` +- RelationV2[t2a#30, t2b#31] csv:examples/src/main/resources/t2.txt == Optimized Logical Plan == Filter (isnotnull(t2a#30) && (t2a#30 > scalar-subquery#56 [])) : +- Aggregate [max(t1a#13) AS max(t1a)#63] : +- Project [t1a#13] : +- RelationV2[t1a#13, t1b#14] csv:examples/src/main/resources/t1.txt +- RelationV2[t2a#30, t2b#31] csv:examples/src/main/resources/t2.txt == Physical Plan == *(1) Project [t2a#30, t2b#31] +- *(1) Filter isnotnull(t2a#30) +- *(1) BatchScan[t2a#30, t2b#31] class org.apache.spark.sql.execution.datasources.v2.csv.CSVScan ``` ## How was this patch tested? ut Closes#24321 from francis0407/SPARK-27411. Authored-by: francis0407 <hanmingcong123@hotmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

* Modify existed Partioning & Distribution to reduce shuffles for full outer join * Refactor and test

* Fix the opentelekomcloud job - Add the OS_VPC_ID environment variable - Change the OS_FLAVOR_ID and OS_FLAVOR_ID_RESIZE value - Enable all the tests running For apache#63 * Update run.yaml * change to use an ubuntu image for testing * add az environment variable for otc job

### What changes were proposed in this pull request? Push down filter through expand. For case below: ``` create table t1(pid int, uid int, sid int, dt date, suid int) using parquet; create table t2(pid int, vs int, uid int, csid int) using parquet; SELECT years, appversion, SUM(uusers) AS users FROM (SELECT Date_trunc('year', dt) AS years, CASE WHEN h.pid = 3 THEN 'iOS' WHEN h.pid = 4 THEN 'Android' ELSE 'Other' END AS viewport, h.vs AS appversion, Count(DISTINCT u.uid) AS uusers ,Count(DISTINCT u.suid) AS srcusers FROM t1 u join t2 h ON h.uid = u.uid GROUP BY 1, 2, 3) AS a WHERE viewport = 'iOS' GROUP BY 1, 2 ``` Plan. before this pr: ``` == Physical Plan == *(5) HashAggregate(keys=[years#30, appversion#32], functions=[sum(uusers#33L)]) +- Exchange hashpartitioning(years#30, appversion#32, 200), true, [id=#251] +- *(4) HashAggregate(keys=[years#30, appversion#32], functions=[partial_sum(uusers#33L)]) +- *(4) HashAggregate(keys=[date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12], functions=[count(if ((gid#44 = 1)) u.`uid`#47 else null)]) +- Exchange hashpartitioning(date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12, 200), true, [id=#246] +- *(3) HashAggregate(keys=[date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12], functions=[partial_count(if ((gid#44 = 1)) u.`uid`#47 else null)]) +- *(3) HashAggregate(keys=[date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12, u.`uid`#47, u.`suid`#48, gid#44], functions=[]) +- Exchange hashpartitioning(date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12, u.`uid`#47, u.`suid`#48, gid#44, 200), true, [id=#241] +- *(2) HashAggregate(keys=[date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12, u.`uid`#47, u.`suid`#48, gid#44], functions=[]) +- *(2) Filter (CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46 = iOS) +- *(2) Expand [ArrayBuffer(date_trunc(year, cast(dt#9 as timestamp), Some(Etc/GMT+7)), CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END, vs#12, uid#7, null, 1), ArrayBuffer(date_trunc(year, cast(dt#9 as timestamp), Some(Etc/GMT+7)), CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END, vs#12, null, suid#10, 2)], [date_trunc('year', CAST(u.`dt` AS TIMESTAMP))#45, CASE WHEN (h.`pid` = 3) THEN 'iOS' WHEN (h.`pid` = 4) THEN 'Android' ELSE 'Other' END#46, vs#12, u.`uid`#47, u.`suid`#48, gid#44] +- *(2) Project [uid#7, dt#9, suid#10, pid#11, vs#12] +- *(2) BroadcastHashJoin [uid#7], [uid#13], Inner, BuildRight :- *(2) Project [uid#7, dt#9, suid#10] : +- *(2) Filter isnotnull(uid#7) : +- *(2) ColumnarToRow : +- FileScan parquet default.t1[uid#7,dt#9,suid#10] Batched: true, DataFilters: [isnotnull(uid#7)], Format: Parquet, Location: InMemoryFileIndex[file:/root/spark-3.0.0-bin-hadoop3.2/spark-warehouse/t1], PartitionFilters: [], PushedFilters: [IsNotNull(uid)], ReadSchema: struct<uid:int,dt:date,suid:int> +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[2, int, true] as bigint))), [id=#233] +- *(1) Project [pid#11, vs#12, uid#13] +- *(1) Filter isnotnull(uid#13) +- *(1) ColumnarToRow +- FileScan parquet default.t2[pid#11,vs#12,uid#13] Batched: true, DataFilters: [isnotnull(uid#13)], Format: Parquet, Location: InMemoryFileIndex[file:/root/spark-3.0.0-bin-hadoop3.2/spark-warehouse/t2], PartitionFilters: [], PushedFilters: [IsNotNull(uid)], ReadSchema: struct<pid:int,vs:int,uid:int> ``` Plan. after. this pr. : ``` == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[years#0, appversion#2], functions=[sum(uusers#3L)], output=[years#0, appversion#2, users#5L]) +- Exchange hashpartitioning(years#0, appversion#2, 5), true, [id=#71] +- HashAggregate(keys=[years#0, appversion#2], functions=[partial_sum(uusers#3L)], output=[years#0, appversion#2, sum#22L]) +- HashAggregate(keys=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12], functions=[count(distinct uid#7)], output=[years#0, appversion#2, uusers#3L]) +- Exchange hashpartitioning(date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, 5), true, [id=#67] +- HashAggregate(keys=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12], functions=[partial_count(distinct uid#7)], output=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, count#27L]) +- HashAggregate(keys=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, uid#7], functions=[], output=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, uid#7]) +- Exchange hashpartitioning(date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, uid#7, 5), true, [id=#63] +- HashAggregate(keys=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles)) AS date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END AS CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, uid#7], functions=[], output=[date_trunc(year, cast(dt#9 as timestamp), Some(America/Los_Angeles))#23, CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END#24, vs#12, uid#7]) +- Project [uid#7, dt#9, pid#11, vs#12] +- BroadcastHashJoin [uid#7], [uid#13], Inner, BuildRight, false :- Filter isnotnull(uid#7) : +- FileScan parquet default.t1[uid#7,dt#9] Batched: true, DataFilters: [isnotnull(uid#7)], Format: Parquet, Location: InMemoryFileIndex[file:/private/var/folders/4l/7_c5c97s1_gb0d9_d6shygx00000gn/T/warehouse-c069d87..., PartitionFilters: [], PushedFilters: [IsNotNull(uid)], ReadSchema: struct<uid:int,dt:date> +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[2, int, false] as bigint)),false), [id=#58] +- Filter ((CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END = iOS) AND isnotnull(uid#13)) +- FileScan parquet default.t2[pid#11,vs#12,uid#13] Batched: true, DataFilters: [(CASE WHEN (pid#11 = 3) THEN iOS WHEN (pid#11 = 4) THEN Android ELSE Other END = iOS), isnotnull..., Format: Parquet, Location: InMemoryFileIndex[file:/private/var/folders/4l/7_c5c97s1_gb0d9_d6shygx00000gn/T/warehouse-c069d87..., PartitionFilters: [], PushedFilters: [IsNotNull(uid)], ReadSchema: struct<pid:int,vs:int,uid:int> ``` ### Why are the changes needed? Improve performance, filter more data. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Added UT Closes#30278 from AngersZhuuuu/SPARK-33302. Authored-by: angerszhu <angers.zhu@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…join can be planned as broadcast join ### What changes were proposed in this pull request? Should not pushdown LeftSemi/LeftAnti over Aggregate for some cases. ```scala spark.range(50000000L).selectExpr("id % 10000 as a", "id % 10000 as b").write.saveAsTable("t1") spark.range(40000000L).selectExpr("id % 8000 as c", "id % 8000 as d").write.saveAsTable("t2") spark.sql("SELECT distinct a, b FROM t1 INTERSECT SELECT distinct c, d FROM t2").explain ``` Before this pr: ``` == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#72] +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- SortMergeJoin [coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L)], [coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L)], LeftSemi :- Sort [coalesce(a#16L, 0) ASC NULLS FIRST, isnull(a#16L) ASC NULLS FIRST, coalesce(b#17L, 0) ASC NULLS FIRST, isnull(b#17L) ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L), 5), ENSURE_REQUIREMENTS, [id=#65] : +- FileScan parquet default.t1[a#16L,b#17L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint> +- Sort [coalesce(c#18L, 0) ASC NULLS FIRST, isnull(c#18L) ASC NULLS FIRST, coalesce(d#19L, 0) ASC NULLS FIRST, isnull(d#19L) ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L), 5), ENSURE_REQUIREMENTS, [id=#66] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- Exchange hashpartitioning(c#18L, d#19L, 5), ENSURE_REQUIREMENTS, [id=#61] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- FileScan parquet default.t2[c#18L,d#19L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c:bigint,d:bigint> ``` After this pr: ``` == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#74] +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- SortMergeJoin [coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L)], [coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L)], LeftSemi :- Sort [coalesce(a#16L, 0) ASC NULLS FIRST, isnull(a#16L) ASC NULLS FIRST, coalesce(b#17L, 0) ASC NULLS FIRST, isnull(b#17L) ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L), 5), ENSURE_REQUIREMENTS, [id=#67] : +- HashAggregate(keys=[a#16L, b#17L], functions=[]) : +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#61] : +- HashAggregate(keys=[a#16L, b#17L], functions=[]) : +- FileScan parquet default.t1[a#16L,b#17L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint> +- Sort [coalesce(c#18L, 0) ASC NULLS FIRST, isnull(c#18L) ASC NULLS FIRST, coalesce(d#19L, 0) ASC NULLS FIRST, isnull(d#19L) ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L), 5), ENSURE_REQUIREMENTS, [id=#68] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- Exchange hashpartitioning(c#18L, d#19L, 5), ENSURE_REQUIREMENTS, [id=#63] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- FileScan parquet default.t2[c#18L,d#19L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c:bigint,d:bigint> ``` ### Why are the changes needed? 1. Pushdown LeftSemi/LeftAnti over Aggregate will affect performance. 2. It will remove user added DISTINCT operator, e.g.: [q38](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q38.sql), [q87](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q87.sql). ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Unit test and benchmark test. SQL | Before this PR(Seconds) | After this PR(Seconds) -- | -- | -- q14a | 660 | 594 q14b | 660 | 600 q38 | 55 | 29 q87 | 66 | 35 Before this pr: ![image](https://user-images.githubusercontent.com/5399861/104452849-8789fc80-55de-11eb-88da-44059899f9a9.png) After this pr: ![image](https://user-images.githubusercontent.com/5399861/104452899-9a043600-55de-11eb-9286-d8f3a23ca3b8.png) Closes#31145 from wangyum/SPARK-34081. Authored-by: Yuming Wang <yumwang@ebay.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

…Anti over Aggregate if join can be planned as broadcast join ### What changes were proposed in this pull request? Should not pushdown LeftSemi/LeftAnti over Aggregate for some cases. ```scala spark.range(50000000L).selectExpr("id % 10000 as a", "id % 10000 as b").write.saveAsTable("t1") spark.range(40000000L).selectExpr("id % 8000 as c", "id % 8000 as d").write.saveAsTable("t2") spark.sql("SELECT distinct a, b FROM t1 INTERSECT SELECT distinct c, d FROM t2").explain ``` Before this pr: ``` == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#72] +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- SortMergeJoin [coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L)], [coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L)], LeftSemi :- Sort [coalesce(a#16L, 0) ASC NULLS FIRST, isnull(a#16L) ASC NULLS FIRST, coalesce(b#17L, 0) ASC NULLS FIRST, isnull(b#17L) ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L), 5), ENSURE_REQUIREMENTS, [id=#65] : +- FileScan parquet default.t1[a#16L,b#17L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint> +- Sort [coalesce(c#18L, 0) ASC NULLS FIRST, isnull(c#18L) ASC NULLS FIRST, coalesce(d#19L, 0) ASC NULLS FIRST, isnull(d#19L) ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L), 5), ENSURE_REQUIREMENTS, [id=#66] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- Exchange hashpartitioning(c#18L, d#19L, 5), ENSURE_REQUIREMENTS, [id=#61] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- FileScan parquet default.t2[c#18L,d#19L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c:bigint,d:bigint> ``` After this pr: ``` == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#74] +- HashAggregate(keys=[a#16L, b#17L], functions=[]) +- SortMergeJoin [coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L)], [coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L)], LeftSemi :- Sort [coalesce(a#16L, 0) ASC NULLS FIRST, isnull(a#16L) ASC NULLS FIRST, coalesce(b#17L, 0) ASC NULLS FIRST, isnull(b#17L) ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(coalesce(a#16L, 0), isnull(a#16L), coalesce(b#17L, 0), isnull(b#17L), 5), ENSURE_REQUIREMENTS, [id=#67] : +- HashAggregate(keys=[a#16L, b#17L], functions=[]) : +- Exchange hashpartitioning(a#16L, b#17L, 5), ENSURE_REQUIREMENTS, [id=#61] : +- HashAggregate(keys=[a#16L, b#17L], functions=[]) : +- FileScan parquet default.t1[a#16L,b#17L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<a:bigint,b:bigint> +- Sort [coalesce(c#18L, 0) ASC NULLS FIRST, isnull(c#18L) ASC NULLS FIRST, coalesce(d#19L, 0) ASC NULLS FIRST, isnull(d#19L) ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(coalesce(c#18L, 0), isnull(c#18L), coalesce(d#19L, 0), isnull(d#19L), 5), ENSURE_REQUIREMENTS, [id=#68] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- Exchange hashpartitioning(c#18L, d#19L, 5), ENSURE_REQUIREMENTS, [id=#63] +- HashAggregate(keys=[c#18L, d#19L], functions=[]) +- FileScan parquet default.t2[c#18L,d#19L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/spark-warehouse/org.apache.spark.sql.Data..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<c:bigint,d:bigint> ``` ### Why are the changes needed? 1. Pushdown LeftSemi/LeftAnti over Aggregate will affect performance. 2. It will remove user added DISTINCT operator, e.g.: [q38](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q38.sql), [q87](https://github.com/apache/spark/blob/master/sql/core/src/test/resources/tpcds/q87.sql). ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Unit test and benchmark test. SQL | Before this PR(Seconds) | After this PR(Seconds) -- | -- | -- q14a | 660 | 594 q14b | 660 | 600 q38 | 55 | 29 q87 | 66 | 35 Before this pr: ![image](https://user-images.githubusercontent.com/5399861/104452849-8789fc80-55de-11eb-88da-44059899f9a9.png) After this pr: ![image](https://user-images.githubusercontent.com/5399861/104452899-9a043600-55de-11eb-9286-d8f3a23ca3b8.png) Closes#31145 from wangyum/SPARK-34081. Authored-by: Yuming Wang <yumwang@ebay.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit d3ea308)

…ne 5 for the other lanes; tasks 56-62 ### What changes were proposed in this pull request? Plan files only, in four commits, each standing on its own. **1. The task table audited against master.** Every row of `PLAN_MILESTONE_4.md` without a DONE marker was checked against the code, its plan file and the merged pull requests. One row was done and unmarked: 41, `unix_date`/`date_from_unix_date`, shipped as apache#63. Row 32 now says what master shows: steps A and B1 done (apache#72), step B2 planned (`PLAN_TASK_32.md` 10, apache#74) and not built, since `groupOutputs` is unchanged and the emitter suite pins one loop method per calendar output. Rows 25, 44, 46 and 47 say "not started" with what moved under them; 39, 42 and 49 say "planned" with their dependencies; 49's plan is located in section 2.19 rather than a task file. `PLAN_TASK_43.md` carried two "8. Outcome" headings; the template one is gone. Rows 34, 35, 38 and 52 are left to apache#115 and apache#116, which close them. **2. Milestone 4 re-scoped to the date family and the emitter; the other lanes become milestone 5.** The owner's decision: milestone 4 is `DateType` plus the emitter and evaluator infrastructure, and every task whose subject is another lane or output type moves out. Six tasks move, text and numbers unchanged, into a new `PLAN_MILESTONE_5.md`: 27 boolean outputs, 28 lane-width conversion, 29 int64 lanes, 30 ANSI integer arithmetic, 39 `date - date` and 49 civil-from-days in long lanes, with their design sections, table rows, verification, risks, the two open questions they own, and the catalogue items about other lanes and the follow-on ladder (1-5, 7-10), item numbers preserved. Milestone 4 keeps stubs under the old section numbers so citations resolve, a dated re-scope note at the top because a plan is a record, and its own catalogue items 6, 11, 12 and 13. The coverage milestone that was called 5 is now 6: `SCOPE_MILESTONE_5.md` is renamed to `SCOPE_MILESTONE_6.md` with a renumbering note, and every reference that meant the coverage milestone (README, VISION, AGENTS, SKILLS, milestone 3's plan, tasks 19, 21 and 22, the quote allowlist) says 6. README gains a milestone 5 entry. **3. Six date-lane tasks from a coverage survey, 56-61.** A survey of Spark's date surface (the function registry and `datetimeExpressions.scala`) against the compiler's arms, taken after the re-scope, left six gaps a date lane can close, each with a design section (2.23 to 2.28) and a row: 56 `date + INTERVAL n DAY` as a compiler rewrite of the int-cast form onto task 38's column-offset node, the stored int64 interval column staying out by decision; 57 `extract(DAYOFWEEK_ISO)` as one narrow arm to a new node, not general int arithmetic; 58 `extract(YEAROFWEEK)` as `year` over task 37's Thursday shift; 59 `next_day` with a weekday column, introducing the derived int32 leaf (an evaluator pre-pass through the row engine's own parser) that 61 reuses; 60 `add_months` with a month-count column, its compile-time bound moved to task 52's producer guard; 61 `trunc` with a format column as a node computing all four levels and selecting per lane. **4. Task 62, the closing measurement.** Every committed number so far compares two Varka lowerings on a laptop whose AVX-512 runs through a 256-bit datapath, against this fork with Varka off. The milestone closes with the number a reader wants: every covered date expression, in the projection and filter shapes, against stock Apache Spark on JDK 17 (its default) and on JDK 25, with Varka on JDK 25, on a runner with a real 512-bit datapath - one Java driver submitted to the three distributions in one dispatch of the benchmark workflow with `expected-cpu` pinned to a full-width Xeon, provenance including the 256-to-512 op-count ladder that proves the datapath, and README's benchmark section rewritten from the three files with a reproduction guide an outside reader can follow. Section 2.29 and row 62. ### Why are the changes needed? The table had drifted from master (one finished task unmarked, several rows with no state), the milestone's scope had grown past its title, and the coverage survey found date-lane gaps no row named. Milestone 4 now says what it is, what is left in it, and how it ends. ### Does this PR introduce _any_ user-facing change? No. Plan and documentation files only; no code. ### How was this patch tested? `dev/varka_quote_check.py` (every quoted figure traces to a committed results file; the four probe figures followed their paragraph to `PLAN_MILESTONE_5.md` in the allowlist) and `dev/varka_precommit.sh --working-tree` on every commit. Every cross-reference to the renamed scope document was found by `git grep` and repointed; the milestone 4 stubs keep every old section and item number resolvable. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Fable 5.1)

simplify the implementation of CoarseGrainedSchedulerBackend
0c0e409

style fix
af470d3

kayousterhout reviewed Mar 3, 2014
View reviewed changes

keep WorkerOffer immutable
43c13e9

markhamstra reviewed Mar 4, 2014
View reviewed changes

use copy idiom to reconstruct the workerOffers
19c2bb4

markhamstra reviewed Mar 4, 2014
View reviewed changes

code clean
f6bf93f

asfgit closed this in a3da508Mar 5, 2014

CodingCat deleted the simplify_CoarseGrainedSchedulerBackend branch March 17, 2014 17:22

wli600 pushed a commit to wli600/spark that referenced this pull request Jul 29, 2015
Merge pull requestapache#63from markhamstra/csd-1.4
25f3168
Catch up with branch-1.4 bug fixes and bump jersey

JasonMWhite pushed a commit to JasonMWhite/spark that referenced this pull request Dec 2, 2015
Merge pull requestapache#63from Shopify/kevincox-spark-timestamp
f7b4551
"Fix" timestamp parsing and update spark.

jlopezmalla pushed a commit to marcosdotps/spark that referenced this pull request Sep 18, 2017
Reduce method complexity in order to improve its readability & perfor…
9d7a92a
…mance (apache#63)

Igosuki pushed a commit to Adikteev/spark that referenced this pull request Jul 31, 2018
Merge pull requestapache#63from mesosphere/go-cli
7209ba3
Implement a Go Spark CLI

weixiuli pushed a commit to weixiuli/spark that referenced this pull request Jun 18, 2019
Reduce shuffles for successive full outer join (apache#63)
3bc83dd
* Modify existed Partioning & Distribution to reduce shuffles for full outer join * Refactor and test

hejian991 pushed a commit to growingio/spark that referenced this pull request Jun 24, 2019
Reduce shuffles for successive full outer join (apache#63)
61bd1c9
* Modify existed Partioning & Distribution to reduce shuffles for full outer join * Refactor and test

sarutak mentioned this pull request Sep 8, 2020
[SPARK-32820][SQL] Remove redundant shuffle exchanges inserted by EnsureRequirements #29677
Closed

XinDongSh pushed a commit to XinDongSh/spark that referenced this pull request Feb 3, 2021
Support append write for object writer (apache#63)
6ba4c5d

Uh oh!

Conversation

CodingCat commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

rxin commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

AmplabJenkins commented Mar 3, 2014

Uh oh!

kayousterhoutMar 3, 2014

Choose a reason for hiding this comment

Uh oh!

kayousterhoutMar 3, 2014

Choose a reason for hiding this comment

Uh oh!

markhamstraMar 3, 2014

Choose a reason for hiding this comment

Uh oh!

andrewor14Mar 3, 2014

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

CodingCat commented Mar 4, 2014

Uh oh!

markhamstraMar 4, 2014

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

markhamstraMar 4, 2014

Choose a reason for hiding this comment

Uh oh!

CodingCatMar 4, 2014

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

kayousterhout commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

AmplabJenkins commented Mar 4, 2014

Uh oh!

CodingCat commented Mar 5, 2014