Skip to content

[SPARK-6743][SQL] Fix empty projections of cached data #6165

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
marmbrus wants to merge 3 commits into from
Closed

[SPARK-6743][SQL] Fix empty projections of cached data #6165

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f1e8df1
[SPARK-6743][SQL] Fix empty projections of cached data
marmbrus May 15, 2015
aad7eab
rxins comments
marmbrus May 15, 2015
4fad158
Merge remote-tracking branch 'origin/master' into wrongColumn
marmbrus May 21, 2015
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 project/SparkBuild.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,7 @@ object Hive {
|import org.apache.spark.sql.functions._
|import org.apache.spark.sql.hive._
|import org.apache.spark.sql.hive.test.TestHive._
|import org.apache.spark.sql.hive.test.TestHive.implicits._
|import org.apache.spark.sql.types._""".stripMargin,
cleanupCommands in console := "sparkContext.stop()",
// Some of our log4j jars make it impossible to submit jobs from this JVM to Hive Map/Reduce
Expand Down
3 changes: 3 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/Row.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,9 @@ object Row {
// TODO: Improve the performance of this if used in performance critical part.
new GenericRow(rows.flatMap(_.toSeq).toArray)
}

/** Returns an empty row. */
val empty = apply()
}


Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/columnar/InMemoryColumnarTableScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,7 +314,7 @@ private[sql] case class InMemoryColumnarTableScan(
columnAccessors(i).extractTo(nextRow, i)
i += 1
}
nextRow
if (attributes.isEmpty) Row.empty else nextRow
Copy link
Contributor

@cloud-fan cloud-fan May 21, 2015

Choose a reason for hiding this comment

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

If attributes.isEmpty, we'll always return Row.empty, so we don't need to create ColumnAccessors, right? The code looks weird to me, why we handle attributes.isEmpty this way?

Copy link
Contributor Author

@marmbrus marmbrus May 21, 2015

Choose a reason for hiding this comment

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

We need at least one column so that we know how many rows to produce, but we should not include it in the output otherwise we will not match the schema we are claiming to have. It would be better to just remember the number of rows in a partition and use that instead. This is just a quick fix so we don't return the wrong answer.

}

override def hasNext: Boolean = columnAccessors(0).hasNext
Expand Down
17 changes: 15 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,19 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
import org.apache.spark.sql.test.TestSQLContext.implicits._
val sqlCtx = TestSQLContext

test("SPARK-6743: no columns from cache") {
Seq(
(83, 0, 38),
(26, 0, 79),
(43, 81, 24)
).toDF("a", "b", "c").registerTempTable("cachedData")

cacheTable("cachedData")
checkAnswer(
sql("SELECT t1.b FROM cachedData, cachedData t1 GROUP BY t1.b"),
Row(0) :: Row(81) :: Nil)
}

test("self join with aliases") {
Seq(1,2,3).map(i => (i, i.toString)).toDF("int", "str").registerTempTable("df")

Expand Down Expand Up @@ -142,7 +155,7 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
sql("SELECT ABS(2.5)"),
Row(2.5))
}

test("aggregation with codegen") {
val originalValue = conf.codegenEnabled
setConf(SQLConf.CODEGEN_ENABLED, "true")
Expand Down Expand Up @@ -194,7 +207,7 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
"SELECT value, sum(key) FROM testData3x GROUP BY value",
(1 to 100).map(i => Row(i.toString, 3 * i)))
testCodeGen(
"SELECT sum(key), SUM(CAST(key as Double)) FROM testData3x",
"SELECT sum(key), SUM(CAST(key as Double)) FROM testData3x",
Row(5050 * 3, 5050 * 3.0) :: Nil)
// AVERAGE
testCodeGen(
Expand Down