apache / carbondata / 881

Build # 881

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/484b80c98403e60f50bae57ec3e13bc7d250915e">484b80c98<a href="https://github.com/apache/carbondata/commit/484b80c98403e60f50bae57ec3e13bc7d250915e">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/93a3ade60b6b2d082c0069d903a125fe8b89cb03">93a3ade60</a>">93a3ade60</a><a href="https://github.com/apache/carbondata/commit/484b80c98403e60f50bae57ec3e13bc7d250915e"> into e9a198ab6">e9a198ab6</a>

Run Details

28 of 28 new or added lines in 1 file covered. (100.0%)

60526 of 75096 relevant lines covered (80.6%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.29

/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonUDFTransformRule.scala

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.sql.optimizer

import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, PredicateHelper,
ScalaUDF}
import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.types.StringType

import org.apache.carbondata.core.constants.CarbonCommonConstants

class CarbonUDFTransformRule extends Rule[LogicalPlan] with PredicateHelper {
  override def apply(plan: LogicalPlan): LogicalPlan = {
      pushDownUDFToJoinLeftRelation(plan)
  }

  private def pushDownUDFToJoinLeftRelation(plan: LogicalPlan): LogicalPlan = {
    val output = plan.transform {
      case proj@Project(cols, Join(
      left, right, jointype: org.apache.spark.sql.catalyst.plans.JoinType, condition)) =>
        var projectionToBeAdded: Seq[org.apache.spark.sql.catalyst.expressions.Alias] = Seq.empty
        var udfExists = false
        val newCols = cols.map {
          case a@Alias(s: ScalaUDF, name)
            if name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID) ||
               name.equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID) =>
            udfExists = true
            projectionToBeAdded :+= a
            AttributeReference(name, StringType, nullable = true)().withExprId(a.exprId)
          case other => other
        }
        if (udfExists) {
          val newLeft = left match {
            case Project(columns, logicalPlan) =>
              Project(columns ++ projectionToBeAdded, logicalPlan)
            case filter: Filter =>
              Project(filter.output ++ projectionToBeAdded, filter)
            case relation: LogicalRelation =>
              Project(relation.output ++ projectionToBeAdded, relation)
            case other => other
          }
          Project(newCols, Join(newLeft, right, jointype, condition))
        } else {
          proj
        }
      case other => other
    }
    output
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.optimizer
19
20	import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, PredicateHelper,
21	ScalaUDF}
22	import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan, Project}
23	import org.apache.spark.sql.catalyst.rules.Rule
24	import org.apache.spark.sql.execution.datasources.LogicalRelation
25	import org.apache.spark.sql.types.StringType
26
27	import org.apache.carbondata.core.constants.CarbonCommonConstants
28
29	class CarbonUDFTransformRule extends Rule[LogicalPlan] with PredicateHelper {	2✔
30	override def apply(plan: LogicalPlan): LogicalPlan = {
31	pushDownUDFToJoinLeftRelation(plan)	2✔
32	}
33
34	private def pushDownUDFToJoinLeftRelation(plan: LogicalPlan): LogicalPlan = {
35	val output = plan.transform {	2✔
36	case proj@Project(cols, Join(	2✔
37	left, right, jointype: org.apache.spark.sql.catalyst.plans.JoinType, condition)) =>	2✔
38	var projectionToBeAdded: Seq[org.apache.spark.sql.catalyst.expressions.Alias] = Seq.empty	2✔
39	var udfExists = false	2✔
40	val newCols = cols.map {	2✔
41	case a@Alias(s: ScalaUDF, name)	2✔
42	if name.equalsIgnoreCase(CarbonCommonConstants.POSITION_ID) \|\|	2✔
43	name.equalsIgnoreCase(CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID) =>	2✔
44	udfExists = true	2✔
45	projectionToBeAdded :+= a	2✔
46	AttributeReference(name, StringType, nullable = true)().withExprId(a.exprId)	2✔
47	case other => other	2✔
48	}
49	if (udfExists) {	2✔
50	val newLeft = left match {	2✔
51	case Project(columns, logicalPlan) =>	2✔
52	Project(columns ++ projectionToBeAdded, logicalPlan)	×
53	case filter: Filter =>	2✔
54	Project(filter.output ++ projectionToBeAdded, filter)	×
55	case relation: LogicalRelation =>	2✔
56	Project(relation.output ++ projectionToBeAdded, relation)	2✔
57	case other => other	×
58	}
59	Project(newCols, Join(newLeft, right, jointype, condition))	2✔
60	} else {
61	proj	2✔
62	}
63	case other => other	2✔
64	}
65	output	2✔
66	}
67
68	}

apache / carbondata / 881

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous