This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 9203f173 fix: Fallback to Spark when shuffling on struct with
duplicate field name (#776)
9203f173 is described below
commit 9203f1734671f83893520618df82ca1c712cb3aa
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Sun Aug 4 17:43:32 2024 -0700
fix: Fallback to Spark when shuffling on struct with duplicate field name
(#776)
---
.../src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala | 4 +++-
.../org/apache/comet/exec/CometColumnarShuffleSuite.scala | 10 ++++++++++
2 files changed, 13 insertions(+), 1 deletion(-)
diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
index ad53c84a..8f08eeba 100644
--- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
@@ -2936,7 +2936,9 @@ object QueryPlanSerde extends Logging with
ShimQueryPlanSerde with CometExprShim
_: DateType | _: BooleanType =>
true
case StructType(fields) =>
- fields.forall(f => supportedDataType(f.dataType))
+ fields.forall(f => supportedDataType(f.dataType)) &&
+ // Java Arrow stream reader cannot work on duplicate field name
+ fields.map(f => f.name).distinct.length == fields.length
case ArrayType(ArrayType(_, _), _) => false // TODO: nested array is not
supported
case ArrayType(MapType(_, _, _), _) => false // TODO: map array element
is not supported
case ArrayType(elementType, _) =>
diff --git
a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
index 9ae882d8..78b4bbb9 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
@@ -68,6 +68,16 @@ abstract class CometColumnarShuffleSuite extends
CometTestBase with AdaptiveSpar
setupTestData()
+ test("Fallback to Spark when shuffling on struct with duplicate field name")
{
+ val df = sql("""
+ | SELECT max(struct(a, record.*, b)) as r FROM
+ | (select a as a, b as b, struct(a,b) as record from testData2) tmp
+ | GROUP BY a
+ """.stripMargin).select($"r.*")
+
+ checkSparkAnswer(df)
+ }
+
test("Unsupported types for SinglePartition should fallback to Spark") {
checkSparkAnswer(spark.sql("""
|SELECT
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
