[SPARK-14270][SQL] whole stage codegen support for typed filter by cloud-fan · Pull Request #12061 · apache/spark

cloud-fan · 2016-03-30T09:45:32Z

What changes were proposed in this pull request?

We implement typed filter by MapPartitions, which doesn't work well with whole stage codegen. This PR use Filter to implement typed filter and we can get the whole stage codegen support for free.

This PR also introduced DeserializeToObject and SerializeFromObject, to seperate serialization logic from object operator, so that it's eaiser to write optimization rules for adjacent object operators.

How was this patch tested?

existing tests.

cloud-fan · 2016-03-30T09:46:39Z

cc @marmbrus @rxin @davies @yhuai @liancheng

SparkQA · 2016-03-30T09:52:17Z

Test build #54510 has finished for PR 12061 at commit be182a9.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- s\"Unable to generate an encoder for inner class$`

viirya · 2016-03-30T10:11:40Z

Because it involves extra deserialization step, will it get performance benefit?

Previously we need to deserialize input row, call function and serialize to row, but now we don't need to do the final serialization, it should be faster even without whole stage codegen.

SparkQA · 2016-03-30T15:10:47Z

Test build #54520 has finished for PR 12061 at commit c5b1fc3.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- s\"Unable to generate an encoder for inner class$`

rxin · 2016-03-30T22:15:24Z

this is a risky thing because as soon as we introduce https://issues.apache.org/jira/browse/SPARK-14083 this test will be useless.
maybe we should introduce a config option now and then explicitly turn off that future optimization
cc @JoshRosen - any good names for the config option?

Alternatively, we could have a special function wrapper which makes the code un-expression-convertable.

SparkQA · 2016-03-31T06:19:52Z

Test build #54593 has finished for PR 12061 at commit aa95fd6.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- s\"Unable to generate an encoder for inner class$`

SparkQA · 2016-03-31T07:32:36Z

Test build #54600 has finished for PR 12061 at commit bf9f5b5.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- s\"Unable to generate an encoder for inner class$`

DeepSparkBot · 2016-04-01T00:10:46Z

Detected style violation.
Suggested improvement:
override lazy val resolved: Boolean = { // If the class to construct is an inner class, we need to get its outer pointer, or this // expression should be regarded as unresolved. // Note that static inner classes (e.g., inner classes within Scala objects) don't need // outer pointer registration. val innerStaticClass = outerPointer.isEmpty && cls.isMemberClass && !Modifier.isStatic(cls.getModifiers) childrenResolved && !innerStaticClass }

SparkQA · 2016-04-05T07:07:29Z

Test build #54957 has finished for PR 12061 at commit 892bdd3.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

…bjectOperator ## What changes were proposed in this pull request? This PR decouples deserializer expression resolution from `ObjectOperator`, so that we can use deserializer expression in normal operators. This is needed by #12061 and #12067 , I abstracted the logic out and put them in this PR to reduce code change in the future. ## How was this patch tested? existing tests. Author: Wenchen Fan <wenchen@databricks.com> Closes#12131 from cloud-fan/separate.

SparkQA · 2016-04-06T06:21:46Z

Test build #55087 has finished for PR 12061 at commit 98744f0.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-04-06T11:24:35Z

generated code for a single filter

case class Data(l: Long, s: String)
Seq(Data(1, "a")).toDS().filter(d => d.l % 2 == 0)

is:

/* 032 */ protected void processNext() throws java.io.IOException {
/* 033 */ /*** PRODUCE: Filter <function1>.apply */
/* 034 */
/* 035 */ /*** PRODUCE: INPUT */
/* 036 */
/* 037 */ while (inputadapter_input.hasNext()) {
/* 038 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 039 */ /*** CONSUME: Filter <function1>.apply */
/* 040 */
/* 041 */ /* input[0, bigint] */
/* 042 */ long inputadapter_value = inputadapter_row.getLong(0);
/* 043 */ /* input[1, string] */
/* 044 */ boolean inputadapter_isNull1 = inputadapter_row.isNullAt(1);
/* 045 */ UTF8String inputadapter_value1 = inputadapter_isNull1 ? null : (inputadapter_row.getUTF8String(1));
/* 046 */
/* 047 */ /* <function1>.apply */
/* 048 */ /* <function1> */
/* 049 */ /* expression: <function1> */
/* 050 */ Object filter_obj = ((Expression) references[1]).eval(null);
/* 051 */ scala.Function1 filter_value1 = (scala.Function1) filter_obj;
/* 052 */ /* newInstance(class org.apache.spark.sql.execution.Data) */
/* 053 */ /* assertnotnull(input[0, bigint], - field (class: "scala.Long", name: "l"), - root class: "org.apache.spark.sql.execution.Data") */
/* 054 */ if (false) {
/* 055 */ throw new RuntimeException((String) references[2]);
/* 056 */ }
/* 057 */ /* input[1, string].toString */
/* 058 */ java.lang.String filter_value5 = inputadapter_isNull1 ? null : (java.lang.String) inputadapter_value1.toString();
/* 059 */ boolean filter_isNull5 = filter_value5 == null;
/* 060 */
/* 061 */ final org.apache.spark.sql.execution.Data filter_value2 = new org.apache.spark.sql.execution.Data(inputadapter_value, filter_value5);
/* 062 */ final boolean filter_isNull2 = false;
/* 063 */ boolean filter_value = false ? false : (boolean) ((java.lang.Boolean)filter_value1.apply(filter_value2)).booleanValue();
/* 064 */ if (false || !filter_value) continue;
/* 065 */
/* 066 */ filter_metricValue.add(1);
/* 067 */
/* 068 */ /*** CONSUME: WholeStageCodegen */
/* 069 */
/* 070 */ filter_holder.reset();
/* 071 */
/* 072 */ filter_rowWriter.zeroOutNullBytes();
/* 073 */
/* 074 */ filter_rowWriter.write(0, inputadapter_value);
/* 075 */
/* 076 */ if (inputadapter_isNull1) {
/* 077 */ filter_rowWriter.setNullAt(1);
/* 078 */ } else {
/* 079 */ filter_rowWriter.write(1, inputadapter_value1);
/* 080 */ }
/* 081 */ filter_result.setTotalSize(filter_holder.totalSize());
/* 082 */ append(filter_result);
/* 083 */ if (shouldStop()) return;
/* 084 */ }
/* 085 */ }

for back-to-back filters

case class Data(l: Long, s: String)
Seq(Data(1, "a")).toDS().filter(d => d.l % 2 == 0).filter(d => d.l % 2 == 0)

is:

/* 049 */ protected void processNext() throws java.io.IOException {
/* 050 */ /*** PRODUCE: SerializeFromObject [input[0, object].l AS l#20L,staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromSt... */
/* 051 */
/* 052 */ /*** PRODUCE: Project [obj#15 AS obj#19] */
/* 053 */
/* 054 */ /*** PRODUCE: Filter (<function1>.apply && <function1>.apply) */
/* 055 */
/* 056 */ /*** PRODUCE: DeserializeToObject newInstance(class org.apache.spark.sql.execution.Data) AS obj#15 */
/* 057 */
/* 058 */ /*** PRODUCE: INPUT */
/* 059 */
/* 060 */ while (inputadapter_input.hasNext()) {
/* 061 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 062 */ /*** CONSUME: DeserializeToObject newInstance(class org.apache.spark.sql.execution.Data) AS obj#15 */
/* 063 */ /* input[0, bigint] */
/* 064 */ long inputadapter_value = inputadapter_row.getLong(0);
/* 065 */ /* input[1, string] */
/* 066 */ boolean inputadapter_isNull1 = inputadapter_row.isNullAt(1);
/* 067 */ UTF8String inputadapter_value1 = inputadapter_isNull1 ? null : (inputadapter_row.getUTF8String(1));
/* 068 */
/* 069 */ /*** CONSUME: Filter (<function1>.apply && <function1>.apply) */
/* 070 */
/* 071 */ /* newInstance(class org.apache.spark.sql.execution.Data) */
/* 072 */ /* assertnotnull(input[0, bigint], - field (class: "scala.Long", name: "l"), - root class: "org.apache.spark.sql.execution.Data") */
/* 073 */ if (false) {
/* 074 */ throw new RuntimeException((String) references[0]);
/* 075 */ }
/* 076 */ /* input[1, string].toString */
/* 077 */ java.lang.String deserializetoobject_value3 = inputadapter_isNull1 ? null : (java.lang.String) inputadapter_value1.toString();
/* 078 */ boolean deserializetoobject_isNull3 = deserializetoobject_value3 == null;
/* 079 */
/* 080 */ final org.apache.spark.sql.execution.Data deserializetoobject_value = new org.apache.spark.sql.execution.Data(inputadapter_value, deserializetoobject_value3);
/* 081 */ final boolean deserializetoobject_isNull = false;
/* 082 */
/* 083 */ /* <function1>.apply */
/* 084 */ /* <function1> */
/* 085 */ /* expression: <function1> */
/* 086 */ Object filter_obj = ((Expression) references[2]).eval(null);
/* 087 */ scala.Function1 filter_value1 = (scala.Function1) filter_obj;
/* 088 */
/* 089 */ boolean filter_value = false ? false : (boolean) ((java.lang.Boolean)filter_value1.apply(deserializetoobject_value)).booleanValue();
/* 090 */ if (false || !filter_value) continue;
/* 091 */ /* <function1>.apply */
/* 092 */ /* <function1> */
/* 093 */ /* expression: <function1> */
/* 094 */ Object filter_obj1 = ((Expression) references[3]).eval(null);
/* 095 */ scala.Function1 filter_value4 = (scala.Function1) filter_obj1;
/* 096 */
/* 097 */ boolean filter_value3 = false ? false : (boolean) ((java.lang.Boolean)filter_value4.apply(deserializetoobject_value)).booleanValue();
/* 098 */ if (false || !filter_value3) continue;
/* 099 */
/* 100 */ filter_metricValue.add(1);
/* 101 */
/* 102 */ /*** CONSUME: Project [obj#15 AS obj#19] */
/* 103 */
/* 104 */ /*** CONSUME: SerializeFromObject [input[0, object].l AS l#20L,staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromSt... */
/* 105 */
/* 106 */ /*** CONSUME: WholeStageCodegen */
/* 107 */
/* 108 */ /* input[0, object].l */
/* 109 */ long serializefromobject_value = deserializetoobject_isNull ? -1L : (long) deserializetoobject_value.l();
/* 110 */ /* staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, object].s, true) */
/* 111 */ /* input[0, object].s */
/* 112 */ java.lang.String serializefromobject_value3 = deserializetoobject_isNull ? null : (java.lang.String) deserializetoobject_value.s();
/* 113 */ boolean serializefromobject_isNull3 = serializefromobject_value3 == null;
/* 114 */
/* 115 */ boolean serializefromobject_isNull2 = !!(serializefromobject_isNull3);
/* 116 */ UTF8String serializefromobject_value2 = null;
/* 117 */
/* 118 */ if (!(serializefromobject_isNull3)) {
/* 119 */ serializefromobject_value2 = org.apache.spark.unsafe.types.UTF8String.fromString(serializefromobject_value3);
/* 120 */ serializefromobject_isNull2 = serializefromobject_value2 == null;
/* 121 */ }
/* 122 */ serializefromobject_holder.reset();
/* 123 */
/* 124 */ serializefromobject_rowWriter.zeroOutNullBytes();
/* 125 */
/* 126 */ if (deserializetoobject_isNull) {
/* 127 */ serializefromobject_rowWriter.setNullAt(0);
/* 128 */ } else {
/* 129 */ serializefromobject_rowWriter.write(0, serializefromobject_value);
/* 130 */ }
/* 131 */
/* 132 */ if (serializefromobject_isNull2) {
/* 133 */ serializefromobject_rowWriter.setNullAt(1);
/* 134 */ } else {
/* 135 */ serializefromobject_rowWriter.write(1, serializefromobject_value2);
/* 136 */ }
/* 137 */ serializefromobject_result.setTotalSize(serializefromobject_holder.totalSize());
/* 138 */ append(serializefromobject_result);
/* 139 */ if (shouldStop()) return;
/* 140 */ }
/* 141 */ }

cloud-fan · 2016-04-06T12:32:50Z

The benchmark result for master branch is:

back-to-back filter: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------
Dataset 724 / 824 13.8 72.4 1.0X

The whole stage version is about 30% faster.

SparkQA · 2016-04-06T13:52:25Z

Test build #55109 has finished for PR 12061 at commit 92af545.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-04-06T14:15:52Z

cc @davies

SparkQA · 2016-04-06T14:37:42Z

Test build #55110 has finished for PR 12061 at commit 0fcaa06.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

davies · 2016-04-06T17:43:19Z

LGTM.

SparkQA · 2016-04-07T02:52:59Z

Test build #55172 has finished for PR 12061 at commit cabe067.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-04-07T03:56:07Z

cc @marmbrus , do you have time to take a look? thx!

SparkQA · 2016-04-07T04:07:40Z

Test build #55177 has finished for PR 12061 at commit 7f1e770.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2016-04-07T05:37:48Z

retest this please

SparkQA · 2016-04-07T07:09:11Z

Test build #55187 has finished for PR 12061 at commit 7f1e770.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

yhuai · 2016-04-08T00:22:55Z

Thanks! Merging to master.

cloud-fan changed the title ~~[SPARK-14270][SQL] whole stage codegen for typed filter~~[SPARK-14270][SQL] whole stage codegen support for typed filterMar 30, 2016

viirya reviewed Mar 30, 2016
View reviewed changes

cloud-fan force-pushed the whole-stage-codegen branch from be182a9 to c5b1fc3Compare March 30, 2016 13:35

rxin reviewed Mar 30, 2016
View reviewed changes

cloud-fan force-pushed the whole-stage-codegen branch 2 times, most recently from aa95fd6 to bf9f5b5Compare March 31, 2016 06:07

DeepSparkBot reviewed Apr 1, 2016
View reviewed changes

cloud-fan mentioned this pull request Apr 2, 2016
[SPARK-14345][SQL] Decouple deserializer expression resolution from ObjectOperator #12131
Closed

cloud-fan force-pushed the whole-stage-codegen branch from bf9f5b5 to 892bdd3Compare April 5, 2016 05:39

whole stage codegen support for typed filter
98744f0

cloud-fan force-pushed the whole-stage-codegen branch from 892bdd3 to 98744f0Compare April 6, 2016 04:53

add optimize rules
0e554cd

add benchmark
0fcaa06

cloud-fan force-pushed the whole-stage-codegen branch from 92af545 to 0fcaa06Compare April 6, 2016 13:07

Merge remote-tracking branch 'origin/master' into whole-stage-codegen
cabe067

add comments
7f1e770

asfgit closed this in 49fb237Apr 8, 2016

Uh oh!

Conversation

cloud-fan commented Mar 30, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

cloud-fan commented Mar 30, 2016

Uh oh!

SparkQA commented Mar 30, 2016

Uh oh!

viiryaMar 30, 2016

Choose a reason for hiding this comment

Uh oh!

cloud-fanMar 30, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Mar 30, 2016

Uh oh!

rxinMar 30, 2016

Choose a reason for hiding this comment

Uh oh!

JoshRosenApr 1, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Mar 31, 2016

Uh oh!

SparkQA commented Mar 31, 2016

Uh oh!

DeepSparkBotApr 1, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 5, 2016

Uh oh!

SparkQA commented Apr 6, 2016

Uh oh!

cloud-fan commented Apr 6, 2016

Uh oh!

cloud-fan commented Apr 6, 2016

Uh oh!

SparkQA commented Apr 6, 2016

Uh oh!

cloud-fan commented Apr 6, 2016

Uh oh!

SparkQA commented Apr 6, 2016

Uh oh!

davies commented Apr 6, 2016

Uh oh!

SparkQA commented Apr 7, 2016

Uh oh!

cloud-fan commented Apr 7, 2016

Uh oh!

SparkQA commented Apr 7, 2016

Uh oh!

cloud-fan commented Apr 7, 2016

Uh oh!

SparkQA commented Apr 7, 2016

Uh oh!

yhuai commented Apr 8, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants