Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,8 @@ object DataSourceStrategy extends Strategy with Logging {
// `Filter`s or cannot be handled by `relation`.
val filterCondition = unhandledPredicates.reduceLeftOption(expressions.And)

// These metadata values make scan plans uniquely identifiable for equality checking.
// TODO(SPARK-17701) using strings for equality checking is brittle
val metadata: Map[String, String] = {
val pairs = ArrayBuffer.empty[(String, String)]

Expand All @@ -350,6 +352,8 @@ object DataSourceStrategy extends Strategy with Logging {
}
pairs += ("PushedFilters" -> markedFilters.mkString("[", ", ", "]"))
}
pairs += ("ReadSchema" ->
StructType.fromAttributes(projects.map(_.toAttribute)).catalogString)
pairs.toMap
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -791,4 +791,12 @@ class JDBCSuite extends SparkFunSuite
val schema = JdbcUtils.schemaString(df, "jdbc:mysql://localhost:3306/temp")
assert(schema.contains("`order` TEXT"))
}

test("SPARK-17673: Exchange reuse respects differences in output schema") {
val df = sql("SELECT * FROM inttypes WHERE a IS NOT NULL")
val df1 = df.groupBy("a").agg("c" -> "min")
val df2 = df.groupBy("a").agg("d" -> "min")
val res = df1.union(df2)
assert(res.distinct().count() == 2) // would be 1 if the exchange was incorrectly reused
}
}