This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ee43f5a9fe [spark] Fix conditional query failing on variant casts
(#9385)
ee43f5a9fe is described below
commit ee43f5a9fe171a0bbb1e39c8521a94b23a650ca2
Author: Arnav Balyan <[email protected]>
AuthorDate: Wed Aug 26 08:13:26 2026 +0530
[spark] Fix conditional query failing on variant casts (#9385)
---
.../paimon/spark/read/VariantPushDownUtils.scala | 7 ++++---
.../apache/paimon/spark/sql/VariantTestBase.scala | 22 ++++++++++++++++++----
2 files changed, 22 insertions(+), 7 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/VariantPushDownUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/VariantPushDownUtils.scala
index 16abb0e2d9..1ac017ac66 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/VariantPushDownUtils.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/VariantPushDownUtils.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark.read
import org.apache.paimon.data.variant.VariantMetadataUtils
-import org.apache.paimon.types.{DataField, RowType}
+import org.apache.paimon.types.{DataField, RowType, VarCharType}
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -40,8 +40,9 @@ object VariantPushDownUtils {
var i = 0
while (i < extractions.length) {
- val (path, _, isVariantTarget) = extractions(i)
- if (path.isEmpty || isVariantTarget) {
+ val (path, info, isVariantTarget) = extractions(i)
+ val canThrow = info.failOnError &&
!info.paimonType.isInstanceOf[VarCharType]
+ if (path.isEmpty || isVariantTarget || canThrow) {
if (path.nonEmpty) {
rejected += path
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
index 4cb3837071..0799310b74 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
@@ -1034,19 +1034,33 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
sql("""INSERT INTO T VALUES (1,
parse_json('{"age":26,"city":"Beijing"}'))""")
val df =
- sql("SELECT id, variant_get(v, '$.age', 'int'), variant_get(v, '$.city',
'string') FROM T")
+ sql(
+ "SELECT id, try_variant_get(v, '$.age', 'int'), variant_get(v,
'$.city', 'string') FROM T")
val v = fieldByPath(scanReadSchemaOf(df), Seq("v"))
if (variantPushDownEnabled) assertVariantStruct(v, expectedFieldCount = 2)
else assert(isVariantType(v.dataType))
checkAnswer(df, Seq(Row(1, 26, "Beijing")))
}
+ test("Paimon Variant pushdown: strict cast preserves conditional
evaluation") {
+ assume(gteqSpark4_1)
+ sql("CREATE TABLE T (id INT, v VARIANT)")
+ sql("""INSERT INTO T VALUES
+ | (0, parse_json('{"a":"bad"}')),
+ | (1, parse_json('{"a":1}'))
+ |""".stripMargin)
+
+ val df = sql("SELECT id, IF(id = 1, variant_get(v, '$.a', 'int'), NULL)
FROM T ORDER BY id")
+ assert(isVariantType(fieldByPath(scanReadSchemaOf(df), Seq("v")).dataType))
+ checkAnswer(df, Seq(Row(0, null), Row(1, 1)))
+ }
+
test("Paimon Variant pushdown: nested variant column inside a struct") {
assume(gteqSpark4_1)
sql("CREATE TABLE T (id INT, nested STRUCT<v: VARIANT, x: INT>)")
sql("""INSERT INTO T VALUES (1, named_struct('v',
parse_json('{"age":26}'), 'x', 100))""")
- val df = sql("SELECT id, variant_get(nested.v, '$.age', 'int') FROM T")
+ val df = sql("SELECT id, try_variant_get(nested.v, '$.age', 'int') FROM T")
val v = fieldByPath(scanReadSchemaOf(df), Seq("nested", "v"))
if (variantPushDownEnabled) assertVariantStruct(v, expectedFieldCount = 1)
else assert(isVariantType(v.dataType))
@@ -1073,7 +1087,7 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
sql("""INSERT INTO T VALUES (1, parse_json('{"a":1}'),
parse_json('{"x":"hi","y":"bye"}'))""")
val df = sql(
- "SELECT variant_get(v1, '$.a', 'int'), variant_get(v2, '$.x', 'string'),
variant_get(v2, '$.y', 'string') FROM T")
+ "SELECT try_variant_get(v1, '$.a', 'int'), variant_get(v2, '$.x',
'string'), variant_get(v2, '$.y', 'string') FROM T")
val readSchema = scanReadSchemaOf(df)
val v1 = fieldByPath(readSchema, Seq("v1"))
val v2 = fieldByPath(readSchema, Seq("v2"))
@@ -1105,7 +1119,7 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
sql("CREATE TABLE T (id INT, v VARIANT)")
sql("""INSERT INTO T VALUES (1, parse_json('{"age":26}'))""")
- val df = sql("SELECT variant_get(v, '$.age', 'int') FROM T")
+ val df = sql("SELECT try_variant_get(v, '$.age', 'int') FROM T")
val desc = df.queryExecution.optimizedPlan
.collectFirst { case DataSourceV2ScanRelation(_, scan, _, _, _) =>
scan.description() }
.getOrElse(fail("expected a DataSourceV2ScanRelation in the plan"))
