diff --git a/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala b/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala index 7c78eea59a1..a93e929f12d 100644 --- a/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala +++ b/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala @@ -16,7 +16,7 @@ */ package org.apache.spark.sql.delta -import org.apache.gluten.backendsapi.velox.VeloxBatchType +import org.apache.gluten.backendsapi.velox.{VeloxBatchType, VeloxValidatorApi} import org.apache.gluten.extension.columnar.transition.Transitions import org.apache.spark.sql.{AnalysisException, Dataset} @@ -48,6 +48,19 @@ class GlutenOptimisticTransaction(delegate: OptimisticTransaction) writeOptions: Option[DeltaOptions], isOptimize: Boolean, additionalConstraints: Seq[Constraint]): Seq[FileAction] = { + // Velox cannot write every Spark type: the native write path inserts a RowToVeloxColumnarExec + // transition whose SparkArrowUtil.toArrowSchema call throws (e.g. `Unsupported data type: + // variant`) for any type it has no Arrow mapping for. Reuse the backend's schema validator so + // such writes fall back to the vanilla Delta write path instead of failing at runtime. + val unsupportedReason = VeloxValidatorApi.validateSchema(inputData.schema) + unsupportedReason match { + case Some(reason) => + logInfo( + s"Schema is not supported by Velox ($reason); falling back to the " + + "vanilla Delta write path.") + return super.writeFiles(inputData, writeOptions, isOptimize, additionalConstraints) + case None => + } hasWritten = true val spark = inputData.sparkSession diff --git a/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaVariantWriteSuite.scala b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaVariantWriteSuite.scala new file mode 100644 index 00000000000..d7dcb3f7f32 --- /dev/null +++ b/backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaVariantWriteSuite.scala @@ -0,0 +1,94 @@ +/* + * 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.sql.delta + +import org.apache.spark.sql.QueryTest +import org.apache.spark.sql.Row +import org.apache.spark.sql.delta.test.DeltaSQLCommandTest +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Velox has no Arrow representation for VariantType, so a Delta write that Gluten offloads to the + * native columnar writer (DataFrameWriter.save, UPDATE, ...) used to throw + * `UnsupportedOperationException: Unsupported data type: variant` at runtime. + * GlutenOptimisticTransaction now detects a schema Velox cannot write (such as one containing a + * variant) and delegates such writes to the vanilla Delta write path. + * + * These tests use write commands that Gluten targets for native offload (DataFrameWriter.save and + * UPDATE): without the fix they reach the native path and fail; with the fix they fall back to + * vanilla Delta and succeed. Plain INSERT INTO is never offloaded, so it would pass regardless of + * the fix and is intentionally not used here. + */ +class DeltaVariantWriteSuite + extends QueryTest + with SharedSparkSession + with DeltaSQLCommandTest { + + test("write and read a top-level variant column") { + withTempDir { + dir => + val path = dir.getCanonicalPath + spark + .range(3) + .selectExpr("'foo' AS s", "parse_json(cast(id + 99 AS string)) AS v") + .write + .format("delta") + .mode("overwrite") + .save(path) + checkAnswer( + spark.read.format("delta").load(path).selectExpr("s", "to_json(v)"), + Seq(Row("foo", "99"), Row("foo", "100"), Row("foo", "101"))) + } + } + + test("write and read a variant nested in a struct") { + withTempDir { + dir => + val path = dir.getCanonicalPath + spark + .range(2) + .selectExpr( + "'foo' AS s", + "named_struct('inner', parse_json(cast(id + 99 AS string))) AS v") + .write + .format("delta") + .mode("overwrite") + .save(path) + checkAnswer( + spark.read.format("delta").load(path).selectExpr("s", "to_json(v.inner)"), + Seq(Row("foo", "99"), Row("foo", "100"))) + } + } + + test("update a table with a variant column") { + withTempDir { + dir => + val path = dir.getCanonicalPath + spark + .range(3) + .selectExpr("cast(id AS int) AS id", "parse_json(cast(id AS string)) AS v") + .write + .format("delta") + .mode("overwrite") + .save(path) + sql(s"UPDATE delta.`$path` SET v = parse_json('123') WHERE id = 1") + checkAnswer( + spark.read.format("delta").load(path).selectExpr("id", "to_json(v)"), + Seq(Row(0, "0"), Row(1, "123"), Row(2, "2"))) + } + } +}