This is an automated email from the ASF dual-hosted git repository.
rui-mo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 1503e13b5f [VL] Offload transform_keys under the LAST_WIN map key
dedup policy (#13490)
1503e13b5f is described below
commit 1503e13b5fdd953fa46e20a91ee90d71c0494b22
Author: Pedrum Jalali <[email protected]>
AuthorDate: Fri Oct 30 13:34:05 2026 -0700
[VL] Offload transform_keys under the LAST_WIN map key dedup policy (#13490)
Co-authored-by: Pedrum Jalali <[email protected]>
---
.../backendsapi/velox/VeloxSparkPlanExecApi.scala | 8 +++++
.../functions/ScalarFunctionsValidateSuite.scala | 34 ++++++++++++++++++++++
.../functions/RegistrationAllFunctions.cc | 1 -
.../gluten/backendsapi/SparkPlanExecApi.scala | 19 ++++++++++++
.../gluten/expression/ExpressionConverter.scala | 18 +-----------
5 files changed, 62 insertions(+), 18 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
index 641f5bbfeb..3a7497482f 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
@@ -198,6 +198,14 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with
Logging {
GenericExpressionTransformer(substraitExprName, Seq(child), expr)
}
+ /** Transform transform_keys to Substrait. Velox honors either map key dedup
policy. */
+ override def genTransformKeysTransformer(
+ substraitExprName: String,
+ children: Seq[ExpressionTransformer],
+ expr: TransformKeys): ExpressionTransformer = {
+ GenericExpressionTransformer(substraitExprName, children, expr)
+ }
+
/** Transform array filter to Substrait. */
override def genArrayFilterTransformer(
substraitExprName: String,
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
index 73a630492d..e93b2c1316 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/functions/ScalarFunctionsValidateSuite.scala
@@ -604,6 +604,40 @@ class ScalarFunctionsValidateSuite extends
FunctionsValidateSuite {
}
}
+ test("transform_keys offloads to Velox under both mapKeyDedupPolicy values")
{
+ // k % 2 maps l_orderkey and l_orderkey + 2 to the same key.
+ val duplicateKeyQuery =
+ "select transform_keys(map(l_orderkey, l_partkey, l_orderkey + 1,
l_suppkey, " +
+ "l_orderkey + 2, l_linenumber), (k, v) -> k % 2) as m, " +
+ "map_keys(transform_keys(map(l_orderkey, l_partkey, l_orderkey + 1,
l_suppkey, " +
+ "l_orderkey + 2, l_linenumber), (k, v) -> k % 2)) as k from lineitem
limit 10"
+
+ // k % 2 maps l_orderkey and l_orderkey + 1 to different keys.
+ val distinctKeyQuery =
+ "select transform_keys(map(l_orderkey, l_partkey, l_orderkey + 1,
l_suppkey), " +
+ "(k, v) -> k % 2) from lineitem limit 10"
+
+ withSQLConf(SQLConf.MAP_KEY_DEDUP_POLICY.key ->
SQLConf.MapKeyDedupPolicy.EXCEPTION.toString) {
+ runQueryAndCompare(distinctKeyQuery) {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+
+ val df = sql(duplicateKeyQuery)
+ checkGlutenPlan[ProjectExecTransformer](df)
+ val e = intercept[SparkException] {
+ df.collect()
+ }
+ assert(e.getMessage.contains("Duplicate map key"))
+ }
+
+ withSQLConf(SQLConf.MAP_KEY_DEDUP_POLICY.key ->
SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
+ // LAST_WIN policy keeps the duplicate's first position and its last
value.
+ runQueryAndCompare(duplicateKeyQuery) {
+ checkGlutenPlan[ProjectExecTransformer]
+ }
+ }
+ }
+
test("raise_error, assert_true") {
runQueryAndCompare("""SELECT assert_true(l_orderkey >= 1), l_orderkey
| from lineitem limit 100""".stripMargin) {
diff --git a/cpp/velox/operators/functions/RegistrationAllFunctions.cc
b/cpp/velox/operators/functions/RegistrationAllFunctions.cc
index 7ed6308193..25ae12c19b 100644
--- a/cpp/velox/operators/functions/RegistrationAllFunctions.cc
+++ b/cpp/velox/operators/functions/RegistrationAllFunctions.cc
@@ -40,7 +40,6 @@ namespace facebook::velox::functions {
void registerPrestoVectorFunctions() {
// Presto function. To be removed.
VELOX_REGISTER_VECTOR_FUNCTION(udf_arrays_overlap, "arrays_overlap");
- VELOX_REGISTER_VECTOR_FUNCTION(udf_transform_keys, "transform_keys");
VELOX_REGISTER_VECTOR_FUNCTION(udf_transform_values, "transform_values");
}
} // namespace facebook::velox::functions
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
index c1dbaa0977..110ba9f78a 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
@@ -44,6 +44,7 @@ import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.execution.python.ArrowEvalPythonExec
import org.apache.spark.sql.execution.window._
import org.apache.spark.sql.hive.HiveUDFTransformer
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DecimalType, StructType}
import org.apache.spark.sql.vectorized.ColumnarBatch
@@ -193,6 +194,24 @@ trait SparkPlanExecApi {
GenericExpressionTransformer(substraitExprName, children, expr)
}
+ /** Falls back under LAST_WIN, which a backend's native transform_keys may
not honor. */
+ def genTransformKeysTransformer(
+ substraitExprName: String,
+ children: Seq[ExpressionTransformer],
+ expr: TransformKeys): ExpressionTransformer = {
+ // Calling `.toString` on both sides ensures compatibility across all
Spark versions.
+ // Starting from Spark 4.1,
`SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY)` returns
+ // an enum instead of a String.
+ if (
+ SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY).toString
+ == SQLConf.MapKeyDedupPolicy.LAST_WIN.toString
+ ) {
+ throw new GlutenNotSupportException(
+ "LAST_WIN policy is not supported yet in native to deduplicate map
keys")
+ }
+ GenericExpressionTransformer(substraitExprName, children, expr)
+ }
+
def genFormatNumberTransformer(
substraitExprName: String,
children: Seq[ExpressionTransformer],
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
index b43f54b555..d915dc805f 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
@@ -922,23 +922,7 @@ object ExpressionConverter extends SQLConfHelper with
Logging {
c
)
case t: TransformKeys =>
- // default is `EXCEPTION`
- val mapKeyDedupPolicy =
SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY)
-
- // Calling `.toString` on both sides ensures compatibility across all
Spark versions.
- // Starting from Spark 4.1,
`SQLConf.get.getConf(SQLConf.MAP_KEY_DEDUP_POLICY)` returns
- // an enum instead of a String. Without `.toString`, the comparison
- // `mapKeyDedupPolicy == SQLConf.MapKeyDedupPolicy.LAST_WIN.toString`
would silently fail
- // in tests, producing only a "Comparing unrelated types" warning in
IntelliJ IDEA,
- // but no compile-time error.
- if (mapKeyDedupPolicy.toString ==
SQLConf.MapKeyDedupPolicy.LAST_WIN.toString) {
- // TODO: Remove after fix ready for
- // https://github.com/facebookincubator/velox/issues/10219
- throw new GlutenNotSupportException(
- "LAST_WIN policy is not supported yet in native to deduplicate map
keys"
- )
- }
- GenericExpressionTransformer(
+
BackendsApiManager.getSparkPlanExecApiInstance.genTransformKeysTransformer(
substraitExprName,
t.children.map(replaceWithExpressionTransformer0(_, attributeSeq,
expressionsMap)),
t
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
