Skip to content

[SPARK-22249][SQL] isin with empty list throws exception on cached DataFrame #19494

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
mgaido91 wants to merge 4 commits into from
Closed

[SPARK-22249][SQL] isin with empty list throws exception on cached DataFrame #19494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
549ff15
[SPARK-22249][SQL] isin with empty list throws exception on cached Da…
mgaido91 Oct 13, 2017
979ac9d
fix wrong implementation
mgaido91 Oct 13, 2017
9efb897
rewrite not to add a false to the normal case
mgaido91 Oct 15, 2017
f5bc105
added tests
mgaido91 Oct 15, 2017
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
1 change: 1 addition & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ case class InMemoryTableScanExec(
case IsNull(a: Attribute) => statsFor(a).nullCount > 0
case IsNotNull(a: Attribute) => statsFor(a).count - statsFor(a).nullCount > 0

case In(_: AttributeReference, list: Seq[Expression]) if list.isEmpty => Literal.FalseLiteral
Copy link
Member

@gatorsmile gatorsmile Oct 17, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

This should be moved to optimizer rules. It will cover both cached and non-cached cases.

Copy link
Member

@gatorsmile gatorsmile Oct 17, 2017

Choose a reason for hiding this comment

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

BTW, this is wrong. If the left value is null, it should return null instead of false.

Copy link
Member

@srowen srowen Oct 17, 2017

Choose a reason for hiding this comment

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

Ah, really. OK, yeah we need to change this. It can be reverted too.

Copy link
Contributor Author

@mgaido91 mgaido91 Oct 17, 2017

Choose a reason for hiding this comment

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

since this is only for filters, does it make any difference null or false?

Copy link
Member

@viirya viirya Oct 19, 2017

Choose a reason for hiding this comment

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

Under this special context of filtering partitions, this In with empty list will result in a false literal in the end, no matter the attribute is null or not. We don't possibly have some expressions like IsNull(In(a, Nil)) as the filter predicate for now.

case In(a: AttributeReference, list: Seq[Expression]) if list.forall(_.isInstanceOf[Literal]) =>
Copy link
Member

@gatorsmile gatorsmile Oct 17, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

Please change it to

if list.forall(_.isInstanceOf[Literal]) && list.nonEmpty()

or using exists.

list.map(l => statsFor(a).lowerBound <= l.asInstanceOf[Literal] &&
l.asInstanceOf[Literal] <= statsFor(a).upperBound).reduce(_ || _)
Expand Down
15 changes: 15 additions & 0 deletions ...e/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -429,4 +429,19 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSQLContext {
checkAnswer(agg_without_cache, agg_with_cache)
}
}

test("SPARK-22249: IN should work also with cached DataFrame") {
Copy link
Member

@gatorsmile gatorsmile Oct 17, 2017

Choose a reason for hiding this comment

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

This test case is just an end-to-end test. This test will still pass if the optimizer has a change. We also need a unit test case.

Copy link
Contributor Author

@mgaido91 mgaido91 Oct 17, 2017

Choose a reason for hiding this comment

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

do you mean this test case is not enough and I should add a test case to check the proper behavior of the optimizer after the change?

val df = spark.range(10).cache()
// with an empty list
assert(df.filter($"id".isin()).count() == 0)
// with a non-empty list
assert(df.filter($"id".isin(2)).count() == 1)
assert(df.filter($"id".isin(2, 3)).count() == 2)
df.unpersist()
val dfNulls = spark.range(10).selectExpr("null as id").cache()
// with null as value for the attribute
assert(dfNulls.filter($"id".isin()).count() == 0)
assert(dfNulls.filter($"id".isin(2, 3)).count() == 0)
dfNulls.unpersist()
}
}