[SPARK-17425][SQL] Override sameResult in HiveTableScanExec to make ReuseExchange work in text format table

[watermen]

What changes were proposed in this pull request?

The PR will override the sameResult in HiveTableScanExec to make ReuseExchange work in text format table.

How was this patch tested?

SQL

SELECT * FROM src t1
JOIN src t2 ON t1.key = t2.key
JOIN src t3 ON t1.key = t3.key;

Before

== Physical Plan ==
*BroadcastHashJoin [key#30], [key#34], Inner, BuildRight
:- *BroadcastHashJoin [key#30], [key#32], Inner, BuildRight
:  :- *Filter isnotnull(key#30)
:  :  +- HiveTableScan [key#30, value#31], MetastoreRelation default, src
:  +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
:     +- *Filter isnotnull(key#32)
:        +- HiveTableScan [key#32, value#33], MetastoreRelation default, src
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
   +- *Filter isnotnull(key#34)
      +- HiveTableScan [key#34, value#35], MetastoreRelation default, src

After

== Physical Plan ==
*BroadcastHashJoin [key#2], [key#6], Inner, BuildRight
:- *BroadcastHashJoin [key#2], [key#4], Inner, BuildRight
:  :- *Filter isnotnull(key#2)
:  :  +- HiveTableScan [key#2, value#3], MetastoreRelation default, src
:  +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
:     +- *Filter isnotnull(key#4)
:        +- HiveTableScan [key#4, value#5], MetastoreRelation default, src
+- ReusedExchange [key#6, value#7], BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))

cc: @davies @cloud-fan

[cloud-fan]

why the default one doesn't work?

[watermen]

left.cleanArgs == right.cleanArgs in defalut sameResult return false, because equals in MetastoreRelation compare the output(AttributeReference) and exprIds are diff. We need to erase the exprId.

[cloud-fan]

ah, I think all leaf nodes suffer this problem, can you follow the way how they fix it? e.g.

  override def sameResult(plan: LogicalPlan): Boolean = {
    plan.canonicalized match {
      case LocalRelation(otherOutput, otherData) =>
        otherOutput.map(_.dataType) == output.map(_.dataType) && otherData == data
      case _ => false
    }
  }

[watermen]

ReuseExchange work in parquet/orc format, because FileSourceScanExec has override the sameResult.

[SparkQA]

Test build #65017 has finished for PR 14988 at commit 8e537a1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #65018 has finished for PR 14988 at commit d9ba28d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

this comment doesn't match the code. Can you explain more about why the default cleanExpression doesn't work?

[watermen]

Sorry for the delay, see the pic below, left.cleanArgs == right.cleanArgs in defalut sameResult return false, because equals in MetastoreRelation compare the output(AttributeReference) and exprIds are diff. cleanExpression can't clean the exprId in AttributeReference. So I think we need to override the sameResult in HiveTableScanExec like FileSourceScanExec. Let me know if I don't explain clearly.

[SparkQA]

Test build #65441 has finished for PR 14988 at commit d9ba28d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

can you follow the existing workaround?

override def sameResult(plan: LogicalPlan): Boolean = {
    plan.canonicalized match {
      case LocalRelation(otherOutput, otherData) =>
        otherOutput.map(_.dataType) == output.map(_.dataType) && otherData == data
      case _ => false
    }
  }

then we don't need to override cleanExpression

[watermen]

I only override the sameReult like FileSourceScanExec. I think HiveTableScanExec is used to text format and FileSourceScanExec is used to parquet/orc format.

[cloud-fan]

does the name matter? I'm not quite sure, but LogicalRelation only checks data type

[watermen]

For example, the full output of table src is (key: Int, value: Int), and output1 is (key: Int), output2 is (value: Int), their(output1, output2) dataType are same, but they are diff and can't be resued.

[watermen]

@cloud-fan

[SparkQA]

Test build #65646 has finished for PR 14988 at commit e410d14.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

LGTM, merging to master!