voonhous commented on code in PR #20039:
URL: https://github.com/apache/hudi/pull/20039#discussion_r4093155589
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala:
##########
@@ -490,6 +490,18 @@ trait SparkAdapter extends Serializable {
*/
def isVariantProjectionStruct(structType: StructType): Boolean = false
+ /**
+ * True when `dataType` is a variant projection struct or holds one below a
struct path: the two
+ * places PushVariantIntoScan puts them (the root of the relation output and
STRUCT members), so
+ * an array element or a map value never matches. The reader context's
schema overlay and the
+ * adapter's row projector both key off this, and they have to agree on it.
+ */
+ def containsVariantProjection(dataType: DataType): Boolean = dataType match {
Review Comment:
Valid. By code reading they did take the vectorized path: both gates only
looked at top-level fields, and `ParquetUtils.isBatchReadSupported` accepts a
nested variant once `spark.sql.parquet.enableNestedColumnVectorizedReader` is
on, which is the default.
Fixed: ported the recursive `containsType` gates from #19777 into
`supportBatch`, so a variant or projection struct nested in a struct, array or
map now reads row-based on Spark 4.1, same as master.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.hudi.dml.schema
+
+import org.apache.hudi.{HoodieSparkUtils, SparkAdapterSupport}
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+
+import org.apache.spark.sql.execution.FileSourceScanExec
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import org.apache.spark.sql.types.{DataType, StructType}
+
+import scala.collection.mutable.ListBuffer
+import scala.util.{Failure, Success, Try}
+
+/**
+ * The pushVariantIntoScan legs of master's TestVariantShreddingMixedLayouts
(#19688, #19783),
+ * ported to release-1.2.1 over unshredded data only, since this branch has no
shredding writer.
+ * spark.sql.variant.pushVariantIntoScan is swept on and off: both arms expect
the very same rows,
+ * and only the plan tells them apart (whether the rule rewrote the variant
into a projection
+ * struct inside the scan). Record types are swept too, which on table version
9 gives avro
+ * (AVRO) and parquet (SPARK) log blocks on the MOR legs.
+ *
+ * Every leg runs to the end and its verdict is recorded, so one wrong result
does not hide the
+ * others; a leg that crashes the JVM ends the run, so the leg filter below
lets a suspect leg run
+ * in its own JVM. VARIANT_LEGS / VARIANT_SKIP_LEGS are comma-separated key
prefixes over
+ * "<scope>:<tableType>:<pushVariantIntoScan>:<recordType>", e.g.
"nested:mor:true".
+ */
+class TestVariantPushVariantIntoScan extends HoodieSparkSqlTestBase {
+
+ private val SPARK_4_1_GATE = "PushVariantIntoScan is on by default from
Spark 4.1"
+
+ // ids 0-2 keep their inserted value, id 3 is updated and id 4's variant is
nulled by the update.
+ private def merged(id: Int): String = if (id < 3) s"x$id" else if (id == 3)
"y3" else null
+ private def mergedJson(id: Int): String = Option(merged(id)).map(k =>
s"""{"k":"$k"}""").orNull
+
+ private def prefixes(name: String): Seq[String] =
+
sys.env.get(name).toSeq.flatMap(_.split(",")).map(_.trim).filter(_.nonEmpty)
+
+ private def legSelected(key: String): Boolean = {
+ val only = prefixes("VARIANT_LEGS")
+ val skip = prefixes("VARIANT_SKIP_LEGS")
+ (only.isEmpty || only.exists(key.startsWith)) &&
!skip.exists(key.startsWith)
+ }
+
+ private def variantProjectionPushedIntoScan(sql: String): Boolean = {
+ def containsProjection(dataType: DataType): Boolean = dataType match {
+ case st: StructType =>
+ SparkAdapterSupport.sparkAdapter.isVariantProjectionStruct(st) ||
+ st.fields.exists(f => containsProjection(f.dataType))
+ case _ => false
+ }
+ val scans = spark.sql(sql).queryExecution.sparkPlan.collect { case scan:
FileSourceScanExec => scan }
+ assert(scans.nonEmpty, s"expected a file scan in the plan of: $sql")
+ scans.exists(_.requiredSchema.fields.exists(f =>
containsProjection(f.dataType)))
+ }
+
+ private def assertPushed(sql: String, pushed: Boolean, leg: String): Unit = {
+ val verdict = if (pushed) "should have" else "must not have"
+ assert(variantProjectionPushedIntoScan(sql) == pushed,
+ s"[$leg] PushVariantIntoScan $verdict rewritten the variant into a
projection struct")
+ }
+
+ /**
+ * Sweeps table type x pushVariantIntoScan x record type, runs `body` once
per leg on a fresh
+ * table, and fails at the end with every leg that failed.
+ */
+ private def sweep(scope: String)(body: (String, String, String, Boolean) =>
Unit): Unit = {
+ val failures = ListBuffer.empty[String]
+ var ran = 0
+ // The conf-off arm runs first so that, when a conf-on leg crashes the
JVM, every other verdict
+ // is already in the log.
+ Seq("cow", "mor").foreach { tableType =>
+ Seq("false", "true").foreach { pushIntoScan =>
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK).foreach {
recordType =>
+ val key = s"$scope:$tableType:$pushIntoScan:$recordType"
+ if (legSelected(key)) {
+ ran += 1
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" ->
pushIntoScan) {
+ withRecordType(Seq(recordType))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val leg = s"$key, $tableName"
+ // scalastyle:off println
+ println(s"LEG START $key")
+ Try(body(tableName, tmp.getCanonicalPath, tableType,
pushIntoScan.toBoolean)) match {
+ case Success(_) => println(s"LEG PASS $key")
+ case Failure(e) =>
+ val msg =
Option(e.getMessage).getOrElse(e.toString).linesIterator.take(3).mkString(" | ")
+ println(s"LEG FAIL $key: ${e.getClass.getSimpleName}:
$msg")
+ failures += s"[$leg] ${e.getClass.getSimpleName}: $msg"
+ }
+ // scalastyle:on println
+ })
+ }
+ }
+ }
+ }
+ }
+ assume(ran > 0, s"no $scope leg selected by VARIANT_LEGS /
VARIANT_SKIP_LEGS")
Review Comment:
Fixed: dropped the `VARIANT_LEGS` / `VARIANT_SKIP_LEGS` filter and the
`assume`, so every leg always runs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
