apache / apache/gluten

[VL] scala.MatchError for KeyGroupedPartitioning in ExecUtil

Open
#10,992 0 comments 0 reactions 0 assignees View on GitHub
bug triage
Dominant language
Scala
Stars
1.6k
Forks
657
Avg merge
2d 14h
Merged PRs (30d)
80

Description

### Backend

VL (Velox)

### Bug description

Spark version is 4.0. How to reproduce this issue:
```scala
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.spark

import java.util.Collections

import org.apache.gluten.config.GlutenConfig
import org.apache.spark.sql.connector.DistributionAndOrderingSuiteBase
import org.apache.spark.sql.connector.catalog.functions.{UnboundBucketFunction, UnboundDaysFunction, UnboundTruncateFunction, UnboundYearsFunction}
import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryTableCatalog}
import org.apache.spark.sql.connector.distributions.Distributions
import org.apache.spark.sql.{GlutenSQLTestsBaseTrait, GlutenTestsBaseTrait, Row}
import org.apache.spark.sql.connector.expressions.Expressions.identity
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.{AUTO_BROADCASTJOIN_THRESHOLD, V2_BUCKETING_ENABLED}
import org.apache.spark.sql.types.{FloatType, LongType, StringType, StructType, TimestampType}

class MyKeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with GlutenTestsBaseTrait {

private val functions = Seq(
UnboundYearsFunction,
UnboundDaysFunction,
UnboundBucketFunction,
UnboundTruncateFunction)

private var originalV2BucketingEnabled: Boolean = false
private var originalAutoBroadcastJoinThreshold: Long = -1

override def sparkConf: SparkConf = {
// Native SQL configs
super.sparkConf
.set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
.set("spark.sql.adaptive.enabled", "false")
.set("spark.sql.shuffle.partitions", "5")
GlutenSQLTestsBaseTrait.nativeSparkConf(super.sparkConf, warehouse)
}

override def beforeAll(): Unit = {
super.beforeAll()
originalV2BucketingEnabled = conf.getConf(V2_BUCKETING_ENABLED)
conf.setConf(V2_BUCKETING_ENABLED, true)
originalAutoBroadcastJoinThreshold = conf.getConf(AUTO_BROADCASTJOIN_THRESHOLD)
conf.setConf(AUTO_BROADCASTJOIN_THRESHOLD, -1L)
conf.setConfString("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
conf.setConfString(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
conf.setConfString("spark.sql.adaptive.enabled", "false")
conf.setConfString("spark.sql.shuffle.partitions", "5")
}

override def afterAll(): Unit = {
try {
super.afterAll()
} finally {
conf.setConf(V2_BUCKETING_ENABLED, originalV2BucketingEnabled)
conf.setConf(AUTO_BROADCASTJOIN_THRESHOLD, originalAutoBroadcastJoinThreshold)
}
}

before {
functions.foreach { f =>
catalog.createFunction(Identifier.of(Array.empty, f.name()), f)
}
}

after {
catalog.clearTables()
catalog.clearFunctions()
}

private val emptyProps: java.util.Map[String, String] = {
Collections.emptyMap[String, String]
}

private val items: String = "items"
private val items_schema: StructType = new StructType()
.add("id", LongType)
.add("name", StringType)
.add("price", FloatType)
.add("arrive_time", TimestampType)

private val purchases: String = "purchases"
private val purchases_schema: StructType = new StructType()
.add("item_id", LongType)
.add("price", FloatType)
.add("time", TimestampType)

private def createTable(
table: String,
schema: StructType,
partitions: Array[Transform],
catalog: InMemoryTableCatalog = catalog): Unit = {
catalog.createTable(Identifier.of(Array("ns"), table),
schema, partitions, emptyProps, Distributions.unspecified(), Array.empty, None, None,
numRowsPerSplit = 1)
}

test("SPARK-41471: shuffle one side: only one side reports partitioning") {
val items_partitions = Array(identity("id"))
createTable(items, items_schema, items_partitions)

sql(s"INSERT INTO testcat.ns.$items VALUES " +
"(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
"(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
"(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")

createTable(purchases, purchases_schema, Array.empty)
sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
"(1, 42.0, cast('2020-01-01' as timestamp)), " +
"(3, 19.5, cast('2020-02-01' as timestamp))")

Seq(true).foreach { shuffle =>
withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> shuffle.toString) {
val df = sql("SELECT id, name, i.price as purchase_price, p.price as sale_price " +
s"FROM testcat.ns.$items i JOIN testcat.ns.$purchases p " +
"ON i.id = p.item_id ORDER BY id, purchase_price, sale_price")

df.explain(true)

checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0), Row(3, "bb", 10.0, 19.5)))
}
}
}
}
```

Exception:
```

keygroupedpartitioning(item_id#27L, 3, [1], [3], [4]) (of class org.apache.spark.sql.catalyst.plans.physical.KeyGroupedPartitioning)
scala.MatchError: keygroupedpartitioning(item_id#27L, 3, [1], [3], [4]) (of class org.apache.spark.sql.catalyst.plans.physical.KeyGroupedPartitioning)
at org.apache.spark.sql.execution.utils.ExecUtil$.genShuffleDependency(ExecUtil.scala:165)
at org.apache.gluten.backendsapi.velox.VeloxSparkPlanExecApi.genShuffleDependency(VeloxSparkPlanExecApi.scala:550)
at org.apache.spark.sql.execution.ColumnarShuffleExchangeExec.columnarShuffleDependency$lzycompute(ColumnarShuffleExchangeExec.scala:92)
at org.apache.spark.sql.execution.ColumnarShuffleExchangeExec.columnarShuffleDependency(ColumnarShuffleExchangeExec.scala:83)
at org.apache.spark.sql.execution.ColumnarShuffleExchangeExec.doExecuteColumnar(ColumnarShuffleExchangeExec.scala:146)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeColumnar$1(SparkPlan.scala:228)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:252)
...
```

### Gluten version

main branch

### Spark version

None

### Spark configurations

_No response_

### System information

_No response_

### Relevant logs

```bash

```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.