Skip to content

[SPARK-25558][SQL] Pushdown predicates for nested fields in DataSource Strategy #22573

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dbtsai wants to merge 1 commit into from
Closed

[SPARK-25558][SQL] Pushdown predicates for nested fields in DataSource Strategy #22573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 59 additions & 47 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -442,53 +442,65 @@ object DataSourceStrategy {
* @return a `Some[Filter]` if the input [[Expression]] is convertible, otherwise a `None`.
*/
protected[sql] def translateFilter(predicate: Expression): Option[Filter] = {
// Recursively try to find an attribute name from the top level that can be pushed down.
def attrName(e: Expression): Option[String] = e match {
// In Spark and many data sources such as parquet, dots are used as a column path delimiter;
// thus, we don't translate such expressions.
case a: Attribute if !a.name.contains(".") =>
Some(a.name)
Copy link
Member

@viirya viirya Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will this cause regression for data source supporting dot in column name?

Copy link
Member Author

@dbtsai dbtsai Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have any data source currently supporting dot in the column name with pushdown? The worst case will be no pushdown for those data sources.

I know ORC doesn't work for now. We can have another followup PR to address this.

Copy link
Member

@viirya viirya Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JDBC data source seems having no such restrict. So I worry that this change can cause some regressions.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Sep 30, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, @dbtsai . This PR has a regression on ORC at least. The following is ORC result in Spark 2.3.2 and it will slowdown at least 5 times like Parquet.

I know ORC doesn't work for now. We can have another followup PR to address this.

scala> val df = spark.range(Int.MaxValue).sample(0.2).toDF("col.with.dot")
scala> df.write.mode("overwrite").orc("/tmp/orc")
scala> df.write.mode("overwrite").parquet("/tmp/parquet")
scala> spark.sql("set spark.sql.orc.impl=native")
scala> spark.sql("set spark.sql.orc.filterPushdown=true")
scala> spark.time(spark.read.orc("/tmp/orc").where("`col.with.dot` = 50000").count)
Time taken: 803 ms

scala> spark.time(spark.read.parquet("/tmp/parquet").where("`col.with.dot` = 50000").count)
Time taken: 5573 ms

scala> spark.version
res6: String = 2.3.2

Copy link
Member

@dongjoon-hyun dongjoon-hyun Sep 30, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Apache Spark 2.4.0 RC2 has a regression on this case. So, for now, this PR doesn't have regssion on master branch.

scala> spark.time(spark.read.orc("/tmp/orc").where("`col.with.dot` = 50000").count)
Time taken: 2405 ms

Copy link
Contributor

@hvanhovell hvanhovell Apr 16, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably a dumb question. Is it possible to store a column with . in its name in parquet?

More holistically, I think it would be better to create an abstraction for a multipart identifier in a filter as opposed to encoding it using a '.'.

case s: GetStructField if !s.childSchema(s.ordinal).name.contains(".") =>
attrName(s.child).map(_ + s".${s.childSchema(s.ordinal).name}")
Copy link
Contributor

@hvanhovell hvanhovell Apr 16, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_ + "." + s.childSchema(s.ordinal).name?

case _ =>
None
}

predicate match {
case expressions.EqualTo(a: Attribute, Literal(v, t)) =>
Some(sources.EqualTo(a.name, convertToScala(v, t)))
case expressions.EqualTo(Literal(v, t), a: Attribute) =>
Some(sources.EqualTo(a.name, convertToScala(v, t)))

case expressions.EqualNullSafe(a: Attribute, Literal(v, t)) =>
Some(sources.EqualNullSafe(a.name, convertToScala(v, t)))
case expressions.EqualNullSafe(Literal(v, t), a: Attribute) =>
Some(sources.EqualNullSafe(a.name, convertToScala(v, t)))

case expressions.GreaterThan(a: Attribute, Literal(v, t)) =>
Some(sources.GreaterThan(a.name, convertToScala(v, t)))
case expressions.GreaterThan(Literal(v, t), a: Attribute) =>
Some(sources.LessThan(a.name, convertToScala(v, t)))

case expressions.LessThan(a: Attribute, Literal(v, t)) =>
Some(sources.LessThan(a.name, convertToScala(v, t)))
case expressions.LessThan(Literal(v, t), a: Attribute) =>
Some(sources.GreaterThan(a.name, convertToScala(v, t)))

case expressions.GreaterThanOrEqual(a: Attribute, Literal(v, t)) =>
Some(sources.GreaterThanOrEqual(a.name, convertToScala(v, t)))
case expressions.GreaterThanOrEqual(Literal(v, t), a: Attribute) =>
Some(sources.LessThanOrEqual(a.name, convertToScala(v, t)))

case expressions.LessThanOrEqual(a: Attribute, Literal(v, t)) =>
Some(sources.LessThanOrEqual(a.name, convertToScala(v, t)))
case expressions.LessThanOrEqual(Literal(v, t), a: Attribute) =>
Some(sources.GreaterThanOrEqual(a.name, convertToScala(v, t)))

case expressions.InSet(a: Attribute, set) =>
val toScala = CatalystTypeConverters.createToScalaConverter(a.dataType)
Some(sources.In(a.name, set.toArray.map(toScala)))
case expressions.EqualTo(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.EqualTo(name, convertToScala(v, t)))
case expressions.EqualTo(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.EqualTo(name, convertToScala(v, t)))

case expressions.EqualNullSafe(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.EqualNullSafe(name, convertToScala(v, t)))
case expressions.EqualNullSafe(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.EqualNullSafe(name, convertToScala(v, t)))

case expressions.GreaterThan(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.GreaterThan(name, convertToScala(v, t)))
case expressions.GreaterThan(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.LessThan(name, convertToScala(v, t)))

case expressions.LessThan(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.LessThan(name, convertToScala(v, t)))
case expressions.LessThan(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.GreaterThan(name, convertToScala(v, t)))

case expressions.GreaterThanOrEqual(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.GreaterThanOrEqual(name, convertToScala(v, t)))
case expressions.GreaterThanOrEqual(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.LessThanOrEqual(name, convertToScala(v, t)))

case expressions.LessThanOrEqual(e: Expression, Literal(v, t)) =>
attrName(e).map(name => sources.LessThanOrEqual(name, convertToScala(v, t)))
case expressions.LessThanOrEqual(Literal(v, t), e: Expression) =>
attrName(e).map(name => sources.GreaterThanOrEqual(name, convertToScala(v, t)))

case expressions.InSet(e: Expression, set) =>
val toScala = CatalystTypeConverters.createToScalaConverter(e.dataType)
attrName(e).map(name => sources.In(name, set.toArray.map(toScala)))

// Because we only convert In to InSet in Optimizer when there are more than certain
// items. So it is possible we still get an In expression here that needs to be pushed
// down.
case expressions.In(a: Attribute, list) if list.forall(_.isInstanceOf[Literal]) =>
val hSet = list.map(_.eval(EmptyRow))
val toScala = CatalystTypeConverters.createToScalaConverter(a.dataType)
Some(sources.In(a.name, hSet.toArray.map(toScala)))
case expressions.In(e: Expression, list) if list.forall(_.isInstanceOf[Literal]) =>
val hSet = list.map(e => e.eval(EmptyRow))
val toScala = CatalystTypeConverters.createToScalaConverter(e.dataType)
attrName(e).map(name => sources.In(name, hSet.toArray.map(toScala)))

case expressions.IsNull(a: Attribute) =>
Some(sources.IsNull(a.name))
case expressions.IsNotNull(a: Attribute) =>
Some(sources.IsNotNull(a.name))
case expressions.IsNull(e: Expression) =>
attrName(e).map(name => sources.IsNull(name))
case expressions.IsNotNull(e: Expression) =>
attrName(e).map(name => sources.IsNotNull(name))

case expressions.And(left, right) =>
// See SPARK-12218 for detailed discussion
Expand All @@ -514,14 +526,14 @@ object DataSourceStrategy {
case expressions.Not(child) =>
translateFilter(child).map(sources.Not)

case expressions.StartsWith(a: Attribute, Literal(v: UTF8String, StringType)) =>
Some(sources.StringStartsWith(a.name, v.toString))
case expressions.StartsWith(e: Expression, Literal(v: UTF8String, StringType)) =>
attrName(e).map(name => sources.StringStartsWith(name, v.toString))

case expressions.EndsWith(a: Attribute, Literal(v: UTF8String, StringType)) =>
Some(sources.StringEndsWith(a.name, v.toString))
case expressions.EndsWith(e: Expression, Literal(v: UTF8String, StringType)) =>
attrName(e).map(name => sources.StringEndsWith(name, v.toString))

case expressions.Contains(a: Attribute, Literal(v: UTF8String, StringType)) =>
Some(sources.StringContains(a.name, v.toString))
case expressions.Contains(e: Expression, Literal(v: UTF8String, StringType)) =>
attrName(e).map(name => sources.StringContains(name, v.toString))

case expressions.Literal(true, BooleanType) =>
Some(sources.AlwaysTrue)
Expand Down
Loading