This is an automated email from the ASF dual-hosted git repository.
zhli pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-gluten.git
The following commit(s) were added to refs/heads/main by this push:
new e975bf368 [VL] Enable map_zip_with, zip_with functions (#5610)
e975bf368 is described below
commit e975bf36881d031b8ef4e1b0c6eb5f1b675b895e
Author: Zhen Li <[email protected]>
AuthorDate: Wed May 8 10:34:21 2024 +0800
[VL] Enable map_zip_with, zip_with functions (#5610)
[VL] Enable map_zip_with, zip_with functions.
---
.../org/apache/gluten/utils/CHExpressionUtil.scala | 2 ++
.../execution/ScalarFunctionsValidateSuite.scala | 37 ++++++++++++++++++++++
docs/velox-backend-support-progress.md | 3 +-
.../gluten/expression/ExpressionMappings.scala | 2 ++
.../gluten/utils/velox/VeloxTestSettings.scala | 2 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 35 +++++++++++++++++++-
.../gluten/utils/velox/VeloxTestSettings.scala | 2 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 35 +++++++++++++++++++-
.../gluten/utils/velox/VeloxTestSettings.scala | 2 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 35 +++++++++++++++++++-
.../gluten/utils/velox/VeloxTestSettings.scala | 2 ++
.../spark/sql/GlutenDataFrameFunctionsSuite.scala | 35 +++++++++++++++++++-
.../apache/gluten/expression/ExpressionNames.scala | 2 ++
13 files changed, 189 insertions(+), 5 deletions(-)
diff --git
a/backends-clickhouse/src/main/scala/org/apache/gluten/utils/CHExpressionUtil.scala
b/backends-clickhouse/src/main/scala/org/apache/gluten/utils/CHExpressionUtil.scala
index 0e645d039..b4190f1b8 100644
---
a/backends-clickhouse/src/main/scala/org/apache/gluten/utils/CHExpressionUtil.scala
+++
b/backends-clickhouse/src/main/scala/org/apache/gluten/utils/CHExpressionUtil.scala
@@ -186,6 +186,8 @@ object CHExpressionUtil {
SKEWNESS -> DefaultValidator(),
BIT_LENGTH -> DefaultValidator(),
MAKE_YM_INTERVAL -> DefaultValidator(),
+ MAP_ZIP_WITH -> DefaultValidator(),
+ ZIP_WITH -> DefaultValidator(),
KURTOSIS -> DefaultValidator(),
REGR_R2 -> DefaultValidator(),
REGR_SLOPE -> DefaultValidator(),
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala
index 0a7a7d6cb..200c5f551 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/execution/ScalarFunctionsValidateSuite.scala
@@ -517,6 +517,43 @@ class ScalarFunctionsValidateSuite extends
FunctionsValidateTest {
}
}
+ test("map_zip_with") {
+ withTempPath {
+ path =>
+ Seq((Map("a" -> 1, "b" -> 2), Map("a" -> 2, "b" -> 3)))
+ .toDF("m1", "m2")
+ .write
+ .parquet(path.getCanonicalPath)
+
+
spark.read.parquet(path.getCanonicalPath).createOrReplaceTempView("map_tbl")
+
+ runQueryAndCompare(
+ "select map_zip_with(m1, m2, (k, v1, v2) -> k == v1 + v2) from
map_tbl") {
+ checkGlutenOperatorMatch[ProjectExecTransformer]
+ }
+ }
+ }
+
+ test("zip_with") {
+ withTempPath {
+ path =>
+ Seq[(Seq[Integer], Seq[Integer])](
+ (Seq(9001, 9002, 9003), Seq(4, 5, 6)),
+ (Seq(1, 2), Seq(3, 4)),
+ (Seq.empty, Seq.empty),
+ (null, null)
+ ).toDF("val1", "val2")
+ .write
+ .parquet(path.getCanonicalPath)
+
+
spark.read.parquet(path.getCanonicalPath).createOrReplaceTempView("array_tbl")
+
+ runQueryAndCompare("select zip_with(val1, val2, (x, y) -> x + y) from
array_tbl") {
+ checkGlutenOperatorMatch[ProjectExecTransformer]
+ }
+ }
+ }
+
test("Test isnan function") {
runQueryAndCompare(
"SELECT isnan(l_orderkey), isnan(cast('NaN' as double)),
isnan(0.0F/0.0F)" +
diff --git a/docs/velox-backend-support-progress.md
b/docs/velox-backend-support-progress.md
index e98587efb..1171b7d91 100644
--- a/docs/velox-backend-support-progress.md
+++ b/docs/velox-backend-support-progress.md
@@ -291,6 +291,7 @@ Gluten supports 199 functions. (Drag to right to see all
data types)
| map_from_entries | map_from_entries |
| | | | | | | |
| | | | | | | | |
| | | |
| map_keys | map_keys | map_keys
| S | | | | | | |
| | | | | | | | |
| | | |
| map_values | map_values | map_values
| S | | | | | | |
| | | | | | | | |
| S | | |
+| map_zip_with | map_zip_with | | S
| | | | | | | |
| | | | | | | |
| S | | |
| named_struct,struct | row_construct | named_struct
| S | | | | | | |
| | | | | | | | |
| | S | |
| posexplode_outer,posexplode | |
| | | | | | | |
| | | | | | | | |
| | | |
| sequence | |
| | | | | | | |
| | | | | | | | |
| | | |
@@ -302,7 +303,7 @@ Gluten supports 199 functions. (Drag to right to see all
data types)
| transform | transform | transofrm
| | | | | | | |
| | | | | | | | |
| | | |
| transform_keys | transform_keys |
| | | | | | | |
| | | | | | | | |
| | | |
| transform_values | transform_values |
| | | | | | | |
| | | | | | | | |
| | | |
-| zip_with | zip_with |
| | | | | | | |
| | | | | | | | |
| | | |
+| zip_with | zip_with | S
| | | | | | | |
| | | | | | | |
| | | | |
| add_months | |
| S | | | | | | |
| | | | | | | | |
| | | |
| current_date | |
| S* | | | | | | |
| | | | | | | | |
| | | |
| current_timestamp | |
| S* | | | | | | |
| | | | | | | | |
| | | |
diff --git
a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala
b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala
index 920b6fab8..ef43c2724 100644
---
a/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala
+++
b/gluten-core/src/main/scala/org/apache/gluten/expression/ExpressionMappings.scala
@@ -240,6 +240,7 @@ object ExpressionMappings {
Sig[ArrayForAll](FORALL),
Sig[ArrayExists](EXISTS),
Sig[Shuffle](SHUFFLE),
+ Sig[ZipWith](ZIP_WITH),
// Map functions
Sig[CreateMap](CREATE_MAP),
Sig[GetMapValue](GET_MAP_VALUE),
@@ -247,6 +248,7 @@ object ExpressionMappings {
Sig[MapValues](MAP_VALUES),
Sig[MapFromArrays](MAP_FROM_ARRAYS),
Sig[MapEntries](MAP_ENTRIES),
+ Sig[MapZipWith](MAP_ZIP_WITH),
Sig[StringToMap](STR_TO_MAP),
// Struct functions
Sig[GetStructField](GET_STRUCT_FIELD),
diff --git
a/gluten-ut/spark32/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark32/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 4844b7351..b9f8c0665 100644
---
a/gluten-ut/spark32/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark32/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -278,6 +278,8 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("shuffle function - array for primitive type not containing null")
.exclude("shuffle function - array for primitive type containing null")
.exclude("shuffle function - array for non-primitive type")
+ // Rewrite this test because Velox sorts rows by key for primitive data
types, which disrupts the original row sequence.
+ .exclude("map_zip_with function - map of primitive types")
enableSuite[GlutenDataFrameTungstenSuite]
enableSuite[GlutenDataFrameSetOperationsSuite]
// Result depends on the implementation for nondeterministic expression
rand.
diff --git
a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 44981e1ce..2b0b40790 100644
---
a/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++
b/gluten-ut/spark32/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,4 +16,37 @@
*/
package org.apache.spark.sql
-class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {}
+import org.apache.spark.sql.functions._
+
+class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {
+ import testImplicits._
+
+ testGluten("map_zip_with function - map of primitive types") {
+ val df = Seq(
+ (Map(8 -> 6L, 3 -> 5L, 6 -> 2L), Map[Integer, Integer]((6, 4), (8, 2),
(3, 2))),
+ (Map(10 -> 6L, 8 -> 3L), Map[Integer, Integer]((8, 4), (4, null))),
+ (Map.empty[Int, Long], Map[Integer, Integer]((5, 1))),
+ (Map(5 -> 1L), null)
+ ).toDF("m1", "m2")
+
+ GlutenQueryTestUtil.sameRows(
+ df.selectExpr("map_zip_with(m1, m2, (k, v1, v2) -> k == v1 +
v2)").collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+
+ GlutenQueryTestUtil.sameRows(
+ df.select(map_zip_with(df("m1"), df("m2"), (k, v1, v2) => k === v1 +
v2)).collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+ }
+}
diff --git
a/gluten-ut/spark33/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark33/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index c6f6b0d33..642873028 100644
---
a/gluten-ut/spark33/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark33/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -943,6 +943,8 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("shuffle function - array for primitive type not containing null")
.exclude("shuffle function - array for primitive type containing null")
.exclude("shuffle function - array for non-primitive type")
+ // Rewrite this test because Velox sorts rows by key for primitive data
types, which disrupts the original row sequence.
+ .exclude("map_zip_with function - map of primitive types")
enableSuite[GlutenDataFrameHintSuite]
enableSuite[GlutenDataFrameImplicitsSuite]
enableSuite[GlutenDataFrameJoinSuite]
diff --git
a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 44981e1ce..2b0b40790 100644
---
a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++
b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,4 +16,37 @@
*/
package org.apache.spark.sql
-class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {}
+import org.apache.spark.sql.functions._
+
+class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {
+ import testImplicits._
+
+ testGluten("map_zip_with function - map of primitive types") {
+ val df = Seq(
+ (Map(8 -> 6L, 3 -> 5L, 6 -> 2L), Map[Integer, Integer]((6, 4), (8, 2),
(3, 2))),
+ (Map(10 -> 6L, 8 -> 3L), Map[Integer, Integer]((8, 4), (4, null))),
+ (Map.empty[Int, Long], Map[Integer, Integer]((5, 1))),
+ (Map(5 -> 1L), null)
+ ).toDF("m1", "m2")
+
+ GlutenQueryTestUtil.sameRows(
+ df.selectExpr("map_zip_with(m1, m2, (k, v1, v2) -> k == v1 +
v2)").collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+
+ GlutenQueryTestUtil.sameRows(
+ df.select(map_zip_with(df("m1"), df("m2"), (k, v1, v2) => k === v1 +
v2)).collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+ }
+}
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 033e084b8..a046f0a02 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark34/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -948,6 +948,8 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("shuffle function - array for primitive type not containing null")
.exclude("shuffle function - array for primitive type containing null")
.exclude("shuffle function - array for non-primitive type")
+ // Rewrite this test because Velox sorts rows by key for primitive data
types, which disrupts the original row sequence.
+ .exclude("map_zip_with function - map of primitive types")
enableSuite[GlutenDataFrameHintSuite]
enableSuite[GlutenDataFrameImplicitsSuite]
enableSuite[GlutenDataFrameJoinSuite]
diff --git
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 44981e1ce..2b0b40790 100644
---
a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++
b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,4 +16,37 @@
*/
package org.apache.spark.sql
-class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {}
+import org.apache.spark.sql.functions._
+
+class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {
+ import testImplicits._
+
+ testGluten("map_zip_with function - map of primitive types") {
+ val df = Seq(
+ (Map(8 -> 6L, 3 -> 5L, 6 -> 2L), Map[Integer, Integer]((6, 4), (8, 2),
(3, 2))),
+ (Map(10 -> 6L, 8 -> 3L), Map[Integer, Integer]((8, 4), (4, null))),
+ (Map.empty[Int, Long], Map[Integer, Integer]((5, 1))),
+ (Map(5 -> 1L), null)
+ ).toDF("m1", "m2")
+
+ GlutenQueryTestUtil.sameRows(
+ df.selectExpr("map_zip_with(m1, m2, (k, v1, v2) -> k == v1 +
v2)").collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+
+ GlutenQueryTestUtil.sameRows(
+ df.select(map_zip_with(df("m1"), df("m2"), (k, v1, v2) => k === v1 +
v2)).collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+ }
+}
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 7c809ec6d..4583396b4 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -964,6 +964,8 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("shuffle function - array for primitive type not containing null")
.exclude("shuffle function - array for primitive type containing null")
.exclude("shuffle function - array for non-primitive type")
+ // Rewrite this test because Velox sorts rows by key for primitive data
types, which disrupts the original row sequence.
+ .exclude("map_zip_with function - map of primitive types")
enableSuite[GlutenDataFrameHintSuite]
enableSuite[GlutenDataFrameImplicitsSuite]
enableSuite[GlutenDataFrameJoinSuite]
diff --git
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
index 44981e1ce..2b0b40790 100644
---
a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
+++
b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameFunctionsSuite.scala
@@ -16,4 +16,37 @@
*/
package org.apache.spark.sql
-class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {}
+import org.apache.spark.sql.functions._
+
+class GlutenDataFrameFunctionsSuite extends DataFrameFunctionsSuite with
GlutenSQLTestsTrait {
+ import testImplicits._
+
+ testGluten("map_zip_with function - map of primitive types") {
+ val df = Seq(
+ (Map(8 -> 6L, 3 -> 5L, 6 -> 2L), Map[Integer, Integer]((6, 4), (8, 2),
(3, 2))),
+ (Map(10 -> 6L, 8 -> 3L), Map[Integer, Integer]((8, 4), (4, null))),
+ (Map.empty[Int, Long], Map[Integer, Integer]((5, 1))),
+ (Map(5 -> 1L), null)
+ ).toDF("m1", "m2")
+
+ GlutenQueryTestUtil.sameRows(
+ df.selectExpr("map_zip_with(m1, m2, (k, v1, v2) -> k == v1 +
v2)").collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+
+ GlutenQueryTestUtil.sameRows(
+ df.select(map_zip_with(df("m1"), df("m2"), (k, v1, v2) => k === v1 +
v2)).collect.toSeq,
+ Seq(
+ Row(Map(8 -> true, 3 -> false, 6 -> true)),
+ Row(Map(10 -> null, 8 -> false, 4 -> null)),
+ Row(Map(5 -> null)),
+ Row(null)),
+ false
+ )
+ }
+}
diff --git
a/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala
b/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala
index 2ad1c6f73..54a55b355 100644
---
a/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala
+++
b/shims/common/src/main/scala/org/apache/gluten/expression/ExpressionNames.scala
@@ -257,6 +257,7 @@ object ExpressionNames {
final val EXISTS = "exists"
final val TRANSFORM = "transform"
final val SHUFFLE = "shuffle"
+ final val ZIP_WITH = "zip_with"
// Map functions
final val CREATE_MAP = "map"
@@ -265,6 +266,7 @@ object ExpressionNames {
final val MAP_VALUES = "map_values"
final val MAP_FROM_ARRAYS = "map_from_arrays"
final val MAP_ENTRIES = "map_entries"
+ final val MAP_ZIP_WITH = "map_zip_with"
final val STR_TO_MAP = "str_to_map"
// struct functions
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
