This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new e9da226db3a9 fix(spark): decide vectorized read per scan in
HoodieFileGroupReaderBasedFileFormat (#20183)
e9da226db3a9 is described below
commit e9da226db3a90747cc51d13fec56ebd1704c1699
Author: Sumit Singh <[email protected]>
AuthorDate: Tue Oct 13 17:36:08 2026 +0530
fix(spark): decide vectorized read per scan in
HoodieFileGroupReaderBasedFileFormat (#20183)
* fix(spark): decide vectorized read per scan in
HoodieFileGroupReaderBasedFileFormat
Spark caches the resolved relation per session, so one file format
instance serves every statement on a table, and supportBatch is only called for
scans narrower than spark.sql.codegen.maxFields. The vectorized decision was
kept in a mutable field and written into
spark.sql.parquet.enableVectorizedReader, so a wide scan could inherit a narrow
scan's answer and fail with ClassCastException: ColumnarBatchRow cannot be cast
to UnsafeRow. Evaluate canReadVectorized against each scan's ow [...]
* Fix new bug which got surface due to fix.
* Fix comments
* Fix comments
* test(spark): check fallback scans keep the session vectorized reader conf
Carry the two regression tests from #20193 into this change. The new
TestVectorizedReadDecisionPerScan checks the session conf after wide,
narrow, Vortex and timestamp-millis scans, but never after a scan that
falls back to row-based reads for a variant column or an incremental
query. Before this fix, buildReaderWithPartitionValues wrote that
fallback into spark.sql.parquet.enableVectorizedReader, so every later
read in the session lost vectorization.
- TestVariantShreddingMixedLayouts: a variant read on Spark 4.1+ falls
back to row-based reads and must leave the session conf at true.
- TestIncrementalReadWithFileGroupReader: a COW incremental read, which
is never vectorized, must leave the session conf at true. It saves
and restores one key instead of a one-entry list, and also checks
that the read returns the rows written.
---
.../HoodieVectorizedParquetRecordReader.java | 8 +-
.../HoodieFileGroupReaderBasedFileFormat.scala | 72 +++--
.../parquet/ParquetSchemaEvolutionUtils.scala | 13 +-
.../parquet/TestParquetSchemaEvolutionUtils.scala | 26 ++
.../TestIncrementalReadWithFileGroupReader.scala | 28 ++
.../TestNestedSchemaPruningOptimization.scala | 11 +-
.../common/TestVectorizedReadDecisionPerScan.scala | 323 +++++++++++++++++++++
.../schema/TestVariantShreddingMixedLayouts.scala | 34 ++-
8 files changed, 463 insertions(+), 52 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/spark/sql/execution/datasources/parquet/HoodieVectorizedParquetRecordReader.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/spark/sql/execution/datasources/parquet/HoodieVectorizedParquetRecordReader.java
index fa944ccde6de..e086aaec0f70 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/spark/sql/execution/datasources/parquet/HoodieVectorizedParquetRecordReader.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/spark/sql/execution/datasources/parquet/HoodieVectorizedParquetRecordReader.java
@@ -94,8 +94,12 @@ public class HoodieVectorizedParquetRecordReader extends
VectorizedParquetRecord
@Override
public void close() throws IOException {
super.close();
- for (Map.Entry<Integer, WritableColumnVector> e :
idToColumnVectors.entrySet()) {
- e.getValue().close();
+ // initBatch allocates these vectors. When initialize or initBatch throws,
Spark closes the
+ // reader before they exist, so skip them rather than mask the original
failure with an NPE.
+ if (idToColumnVectors != null) {
+ for (Map.Entry<Integer, WritableColumnVector> e :
idToColumnVectors.entrySet()) {
+ e.getValue().close();
+ }
}
idToColumnVectors = null;
columnarBatch = null;
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 6849798d259e..d0d503792e69 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -120,16 +120,6 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
private def evaluateRequiredFiltersPerRow: Boolean =
requiredFilters.nonEmpty && !requiredFiltersAppliedInPlan
- /**
- * Flag saying whether vectorized reading is supported.
- */
- private var supportVectorizedRead = false
-
- /**
- * Flag saying whether batch output is supported.
- */
- private var supportReturningBatch = false
-
/**
* Cached result of vector column detection keyed by schema identity.
* Avoids re-parsing metadata on repeated supportBatch calls with the same
schema.
@@ -166,30 +156,23 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
/**
- * Checks if the file format supports vectorized reading, please refer to
SPARK-40918.
- *
- * NOTE: for mor read, even for file-slice with only base file, we can read
parquet file with vectorized read,
- * but the return result of the whole data-source-scan phase cannot be batch,
- * because when there are any log file in a file slice, it needs to be read
by the file group reader.
- * Since we are currently performing merges based on rows, the result
returned by merging should be based on rows,
- * we cannot assume that all file slices have only base files.
- * So we need to set the batch result back to false.
+ * Whether the base-file reader may use a vectorized reader for a scan whose
output schema is `schema`.
*
+ * This is a pure function of the session conf, the schema and the table
properties. No state is kept
+ * between calls on purpose: Spark only calls [[supportBatch]] for scans
whose schema is below
+ * spark.sql.codegen.maxFields, so [[buildReaderWithPartitionValues]] must
evaluate this itself for
+ * the schema it is handed rather than reuse the answer given for some
earlier scan.
*/
- override def supportBatch(sparkSession: SparkSession, schema: StructType):
Boolean = {
+ private def canReadVectorized(sparkSession: SparkSession, schema:
StructType): Boolean = {
// Vector columns are stored as FIXED_LEN_BYTE_ARRAY in Parquet but read
as ArrayType in Spark.
// The binary→array conversion requires row-level access, so disable
vectorized batch reading.
if (detectVectorColumnsCached(schema).nonEmpty) {
- supportVectorizedRead = false
- supportReturningBatch = false
false
} else if (schema.fields.exists(f => containsType(f.dataType,
isVariantProjection))) {
// Spark 4.1's PushVariantIntoScan rewrites a variant column to a struct
of pushed-down
// extractions. The Spark vectorized parquet reader treats this as a
nested type change
// (data column is VariantType, required is a struct) and refuses to
read in vectorized
// mode (ParquetSchemaEvolutionUtils throws). Force row-based reading on
this path.
- supportVectorizedRead = false
- supportReturningBatch = false
false
} else if (HoodieSparkUtils.gteqSpark4_1
&& schema.fields.exists(f => containsType(f.dataType,
sparkAdapter.isVariantType))) {
@@ -206,8 +189,6 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
// carried inside a struct is exposed exactly the same way under any
range-partitioned query.
// The cost is that nested-variant tables lose vectorization on 4.1+ by
default, exactly as
// top-level ones do.
- supportVectorizedRead = false
- supportReturningBatch = false
false
} else {
val conf = sparkSession.sessionState.conf
@@ -217,8 +198,10 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
s.dataType,
sparkSession.sessionState.conf.orcVectorizedReaderNestedColumnEnabled))
// TODO: Implement columnar batch reading
https://github.com/apache/hudi/issues/17736
val lanceBatchSupported = false
+ // The Vortex reader returns rows only
+ val vortexBatchSupported = false
- val supportBatch = if (isMultipleBaseFileFormatsEnabled) {
+ val formatSupportsBatch = if (isMultipleBaseFileFormatsEnabled) {
parquetBatchSupported && orcBatchSupported
} else if (hoodieFileFormat == HoodieFileFormat.PARQUET) {
parquetBatchSupported
@@ -226,6 +209,8 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
orcBatchSupported
} else if (hoodieFileFormat == HoodieFileFormat.LANCE) {
lanceBatchSupported
+ } else if (hoodieFileFormat == HoodieFileFormat.VORTEX) {
+ vortexBatchSupported
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
@@ -234,17 +219,27 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
// the same reader, so vectorized reading must be disabled for both. All
other combinations
// (COW snapshot, COW incremental, MOR snapshot) either have no log
merging or handle it via
// a separate non-vectorized fileGroupBaseFileReader while the base file
reader stays vectorized.
- supportVectorizedRead = !(isIncremental && (isMOR || isCDC)) &&
!isBootstrap && supportBatch
- // Required filters (the commit-time range of an incremental query) must
hold for every row, but the
- // vectorized reader only uses them to prune row groups. Unless the plan
already filters on them, return
- // rows rather than batches so that the base file read can evaluate them
row by row; decoding stays vectorized.
- supportReturningBatch = !isMOR && supportVectorizedRead &&
!evaluateRequiredFiltersPerRow
- logDebug(s"supportReturningBatch: $supportReturningBatch,
supportVectorizedRead: $supportVectorizedRead, isIncremental: $isIncremental, "
+
- s"isCDC: $isCDC, isBootstrap: $isBootstrap, superSupportBatch:
$supportBatch")
- supportReturningBatch
+ !(isIncremental && (isMOR || isCDC)) && !isBootstrap &&
formatSupportsBatch
}
}
+ /**
+ * Checks if the file format supports returning columnar batches, please
refer to SPARK-40918.
+ *
+ * NOTE: for MOR, a file slice with only a base file can still be read with
the vectorized reader,
+ * but the scan as a whole cannot return batches because slices with log
files are merged row by row.
+ */
+ override def supportBatch(sparkSession: SparkSession, schema: StructType):
Boolean = {
+ val vectorized = canReadVectorized(sparkSession, schema)
+ // Required filters (the commit-time range of an incremental query) must
hold for every row, but the
+ // vectorized reader only uses them to prune row groups. Unless the plan
already filters on them, return
+ // rows rather than batches so that the base file read can evaluate them
row by row; decoding stays vectorized.
+ val returningBatch = !isMOR && vectorized && !evaluateRequiredFiltersPerRow
+ logDebug(s"supportReturningBatch: $returningBatch, supportVectorizedRead:
$vectorized, " +
+ s"isIncremental: $isIncremental, isCDC: $isCDC, isBootstrap:
$isBootstrap, isMOR: $isMOR")
+ returningBatch
+ }
+
//for partition columns that we read from the file, we don't want them to be
constant column vectors so we
//modify the vector types in this scenario
override def vectorTypes(requiredSchema: StructType,
@@ -362,10 +357,13 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val dataStructTypeWithMandatoryPartitionFields =
StructType(dataStructType.fields ++ partitionSchema.fields.filter(f =>
mandatoryFields.contains(f.name) && !isNestedPartitionField(f.name)))
val dataSchema = HoodieSchemaUtils.pruneDataSchema(schema,
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(dataStructTypeWithMandatoryPartitionFields,
sanitizedTableName), exclusionFields)
-
spark.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader",
supportVectorizedRead.toString)
+ // Decide per scan: Spark skips supportBatch for scans wider than
spark.sql.codegen.maxFields.
+ // Never write it to spark.sql.parquet.enableVectorizedReader: Spark uses
that conf to decide whether to
+ // convert the scan output to UnsafeRow, and skipping the conversion for
vectorized rows fails.
+ val vectorizedRead = canReadVectorized(spark, outputSchema)
- val baseFileReader =
spark.sparkContext.broadcast(buildBaseFileReader(spark, options,
augmentedStorageConf.unwrap(), dataStructType, supportVectorizedRead))
- val fileGroupBaseFileReader = if (isMOR && supportVectorizedRead) {
+ val baseFileReader =
spark.sparkContext.broadcast(buildBaseFileReader(spark, options,
augmentedStorageConf.unwrap(), dataStructType, vectorizedRead))
+ val fileGroupBaseFileReader = if (isMOR && vectorizedRead) {
// for file group reader to perform read, we always need to read the
record without vectorized reader because our merging is based on row level.
// TODO: please consider to support vectorized reader in file group
reader
spark.sparkContext.broadcast(buildBaseFileReader(spark, options,
augmentedStorageConf.unwrap(), dataStructType, enableVectorizedRead = false))
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index 47ee1c02cda9..dd462a3b529f 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -64,7 +64,10 @@ class ParquetSchemaEvolutionUtils(readConf: Configuration,
// Fetch internal schema
private lazy val querySchemaOption: util.Option[InternalSchema] =
pruneInternalSchema(internalSchemaOpt, requiredSchema)
- var shouldUseInternalSchema: Boolean = querySchemaOption.isPresent &&
tablePath != null
+ // Empty projections (count(*), select 1) read no column data, so skip
schema-on-read. Their query
+ // schema is unpruned (see pruneInternalSchema), and merging it would
request every column, including
+ // a variant as a plain struct that Spark 4.x's vectorized reader rejects.
+ var shouldUseInternalSchema: Boolean = requiredSchema.nonEmpty &&
querySchemaOption.isPresent && tablePath != null
private lazy val schemaUtils: HoodieSchemaUtils = sparkAdapter.getSchemaUtils
@@ -107,12 +110,8 @@ class ParquetSchemaEvolutionUtils(readConf: Configuration,
def setRequestedSchema(schema: StructType): Unit =
fileOwnConf().set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA,
schema.json)
typeChangeInfos = if (shouldUseInternalSchema) {
- // Empty projections (count(*), select 1) read no column data, so there
is nothing to
- // reconstruct - and querySchemaOption is the UNPRUNED table schema in
that case (see
- // pruneInternalSchema), so running the guard would fail queries that
work fine.
- if (requiredSchema.nonEmpty) {
- ParquetSchemaEvolutionUtils.validateNoShreddedVariants(requiredSchema,
querySchemaOption.get(), footerFileMetaData)
- }
+ // shouldUseInternalSchema implies a non-empty projection, so
querySchemaOption is pruned here.
+ ParquetSchemaEvolutionUtils.validateNoShreddedVariants(requiredSchema,
querySchemaOption.get(), footerFileMetaData)
val mergedInternalSchema = new InternalSchemaMerger(fileSchema,
querySchemaOption.get(), true, true).mergeSchema()
val mergedSchema =
SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index 87b0f5f3b59e..4e99d80fb6d0 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -19,12 +19,14 @@
package org.apache.spark.sql.execution.datasources.parquet
+import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
import org.apache.hudi.common.schema.internal.{InternalSchema, Type =>
InternalType, Types => InternalTypes}
import org.apache.hudi.common.schema.internal.action.TableChanges
import org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter
import org.apache.hudi.common.schema.internal.utils.SchemaChangeUtils
import org.apache.hudi.common.util
+import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.exception.HoodieException
import org.apache.hadoop.conf.Configuration
@@ -517,6 +519,30 @@ class TestParquetSchemaEvolutionUtils {
Assertions.assertNull(pushedFilterOf(readConf))
}
+ /**
+ * An empty projection (count(*), select 1, a partition-only select) reads
no column data, so it
+ * must bypass the schema-on-read merge. Its query schema is the UNPRUNED
table schema, and merging
+ * that would request every column - a variant one as the plain struct the
internal-schema
+ * converter emits, which Spark 4.x's vectorized reader rejects for a
VARIANT-annotated group.
+ */
+ @Test
+ def testEmptyProjectionBypassesInternalSchema(): Unit = {
+ val tablePath = "/tmp/hudi_table"
+ val conf = new Configuration()
+ conf.set(SparkInternalSchemaConverter.HOODIE_TABLE_PATH, tablePath)
+ val internalSchema = internalSchemaOf(
+ "id" -> HoodieSchema.create(HoodieSchemaType.INT),
+ "v" -> HoodieSchema.createVariant())
+ def utilsFor(requiredSchema: StructType): ParquetSchemaEvolutionUtils =
+ new ParquetSchemaEvolutionUtils(conf, new
Path(s"$tablePath/f1_0-1-1_20260101000000000.parquet"),
+ requiredSchema, new StructType(), HOption.of(internalSchema))
+
+ Assertions.assertFalse(utilsFor(new StructType()).shouldUseInternalSchema,
+ "An empty projection must be read without the internal schema")
+ Assertions.assertTrue(utilsFor(new StructType().add("id",
IntegerType)).shouldUseInternalSchema,
+ "A non-empty projection on a table with an internal schema must still
use it")
+ }
+
/** An internal schema over the given top-level columns, in order; field ids
are positional. */
private def internalSchemaOf(fields: (String, HoodieSchema)*):
InternalSchema =
InternalSchemaConverter.convert(HoodieSchema.createRecord("query",
"org.apache.hudi.test", null,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
index b8c9e9652d1f..5eff5bd932d6 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
@@ -31,6 +31,7 @@ import
org.apache.hudi.testutils.SparkClientFunctionalTestHarness
import org.apache.hadoop.fs.Path
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.{CsvSource, ValueSource}
@@ -309,6 +310,33 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
Set(("k1", 5), ("k2", 4), ("k3", 5), ("k4", 5)))
}
+ // MOR incremental scans never read vectorized. That decision belongs to the
scan alone: it used to be
+ // written into spark.sql.parquet.enableVectorizedReader, so every later
read in the session lost
+ // vectorization.
+ @Test
+ def testMorIncrementalReadDoesNotMutateVectorizedReaderSessionConf(): Unit =
{
+ val vectorizedReaderConf = "spark.sql.parquet.enableVectorizedReader"
+ val previous = spark.conf.getOption(vectorizedReaderConf)
+ try {
+ spark.conf.set(vectorizedReaderConf, "true")
+ write(batches.head._1, "MERGE_ON_READ", 8, batches.head._2,
SaveMode.Overwrite)
+
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance()).setBasePath(basePath()).build()
+ val instant =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ .lastInstant().get().getCompletionTime
+
+ assertEquals(batches.head._1.size, readIncremental(8, "000",
instant).collect().length)
+ assertTrue(spark.conf.get(vectorizedReaderConf).toBoolean,
+ "MOR incremental read must preserve the session vectorized-reader
setting")
+ } finally {
+ previous match {
+ case Some(value) => spark.conf.set(vectorizedReaderConf, value)
+ case None => spark.conf.unset(vectorizedReaderConf)
+ }
+ }
+ }
+
private def assertFullTableScanRange(metaClient: HoodieTableMetaClient,
useLegacyRdd: Boolean,
start: String,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
index f4b539de252d..6bcc84ee69d0 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
@@ -344,7 +344,10 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
test("Nested struct predicate pushdown with schema on read and a renamed
nested column") {
withTempDir { tmp =>
Seq("cow", "mor").foreach { tableType =>
- withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ // NOTE: Row-based reads, since the vectorized reader cannot read a
nested type change: the base file
+ // keeps the old layout of item, so
ParquetSchemaEvolutionUtils.getFileReadConf rejects the read
+ withSQLConf("hoodie.schema.on.read.enable" -> "true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
@@ -373,9 +376,11 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
test("Nested struct predicate pushdown with schema on read and quoted column
names") {
withTempDir { tmp =>
Seq("cow", "mor").foreach { tableType =>
- // NOTE: A top-level rename fails the incoming-schema check unless
column drops are allowed
+ // NOTE: A top-level rename fails the incoming-schema check unless
column drops are allowed. The nested
+ // rename needs row-based reads, as in the renamed nested column
test above.
withSQLConf("hoodie.schema.on.read.enable" -> "true",
- "hoodie.datasource.write.schema.allow.auto.evolution.column.drop" ->
"true") {
+ "hoodie.datasource.write.schema.allow.auto.evolution.column.drop" ->
"true",
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") {
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
// Names that are not plain identifiers reach the reader
backtick-quoted (`名字`,
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadDecisionPerScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadDecisionPerScan.scala
new file mode 100644
index 000000000000..57848b89f66a
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestVectorizedReadDecisionPerScan.scala
@@ -0,0 +1,323 @@
+/*
+ * 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.common
+
+import org.apache.hudi.{DefaultSparkRecordMerger, HoodieSparkUtils,
SparkAdapterSupport}
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.client.common.HoodieSparkEngineContext
+import org.apache.hudi.common.config.{HoodieMetadataConfig, RecordMergeMode}
+import org.apache.hudi.common.model.{HoodieAvroIndexedRecord, HoodieKey,
HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.apache.hudi.index.HoodieIndex
+
+import org.apache.avro.{LogicalTypes, Schema, SchemaBuilder}
+import org.apache.avro.generic.{GenericData, IndexedRecord}
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.{FileSourceScanExec,
WholeStageCodegenExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+
+import java.io.File
+import java.nio.file.{Files, Paths}
+import java.sql.Timestamp
+
+import scala.collection.JavaConverters._
+
+/**
+ * Spark caches the resolved relation of a table per session, so every
statement on the table shares one
+ * HoodieFileGroupReaderBasedFileFormat instance, and it asks that instance
about batch support only for scans
+ * narrower than spark.sql.codegen.maxFields. A wider scan therefore has to
decide about vectorized reading for
+ * its own schema when its reader is built, and no scan may change
spark.sql.parquet.enableVectorizedReader,
+ * which Spark reads to decide whether the scan output still needs an
UnsafeRow conversion.
+ */
+class TestVectorizedReadDecisionPerScan extends HoodieSparkSqlTestBase with
AdaptiveSparkPlanHelper {
+
+ private val vectorizedReaderConf = "spark.sql.parquet.enableVectorizedReader"
+ // Leaf fields of the payload struct; with them the table exceeds
spark.sql.codegen.maxFields, so a
+ // select * scan is planned row-based without a supportBatch call
+ private val payloadFieldCount = 120
+
+ Seq("cow", "mor").foreach { tableType =>
+ test(s"Test wide scan after narrow scan decides vectorized read per scan
for $tableType table") {
+ withSQLConf(vectorizedReaderConf -> "true",
"spark.sql.codegen.maxFields" -> "100") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ updateOneFileGroup(tableName, tableType, createWideTable(tableName,
tableType, tmp))
+
+ // The narrow scan is asked supportBatch and may read vectorized
+ checkNarrowScan(tableName, updatedRows)
+ checkAnswer(s"select count(*) from $tableName")(Seq(3))
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+
+ // The wide scan is never asked, and must not inherit the narrow
scan's answer
+ assert(!scanIsBatched(s"select * from $tableName"))
+ checkWideScan(tableName, updatedRows)
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+
+ // With the vectorized reader disabled Spark skips the UnsafeRow
conversion of the scan output, so a
+ // wide scan handed a leftover vectorized reader fails with
+ // ClassCastException: ColumnarBatchRow cannot be cast to UnsafeRow
+ withSQLConf(vectorizedReaderConf -> "false") {
+ checkWideScan(tableName, updatedRows)
+ checkNarrowScan(tableName, updatedRows)
+ assertResult("false")(spark.conf.get(vectorizedReaderConf))
+ }
+ }
+ }
+ }
+
+ test(s"Test wide scan keeps vectorized reads enabled for later scans for
$tableType table") {
+ withSQLConf(vectorizedReaderConf -> "true",
"spark.sql.codegen.maxFields" -> "100") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ updateOneFileGroup(tableName, tableType, createWideTable(tableName,
tableType, tmp))
+
+ checkWideScan(tableName, updatedRows)
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+
+ // Only COW scans return batches; MOR merges rows, so its scans are
never columnar
+ if (tableType == "cow") {
+ assert(scanIsBatched(s"select id, name, dt from $tableName"))
+ assert(scanIsBatched(s"select count(*) from $tableName"))
+ }
+ checkNarrowScan(tableName, updatedRows)
+ checkAnswer(s"select count(*) from $tableName")(Seq(3))
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+ }
+ }
+ }
+
+ // Every scan now asks canReadVectorized, so a format it does not know
fails every read, not only narrow ones
+ test(s"Test narrow and wide scans of a Vortex table read row-based for
$tableType table") {
+
assume(SparkAdapterSupport.sparkAdapter.createVortexFileReader(vectorized =
false, spark.sessionState.conf,
+ Map.empty, new Configuration()).isDefined, "Vortex reads need a Spark
4.x adapter")
+ // Vortex files are written from Spark records only
+ withSQLConf(vectorizedReaderConf -> "true",
"spark.sql.codegen.maxFields" -> "100",
+ HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key ->
classOf[DefaultSparkRecordMerger].getName) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ createWideTable(tableName, tableType, tmp,
Map(HoodieTableConfig.BASE_FILE_FORMAT.key -> "VORTEX"))
+
+ assert(!scanIsBatched(s"select id, name, dt from $tableName"))
+ checkNarrowScan(tableName, insertedRows)
+ assert(!scanIsBatched(s"select * from $tableName"))
+ checkWideScan(tableName, insertedRows)
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+ }
+ }
+ }
+ }
+
+ // Before Spark 3.5, Hudi never reads a table with a timestamp-millis field
vectorized (#14161), so these scans are
+ // row-based while spark.sql.parquet.enableVectorizedReader stays true.
Spark then converts the scan output to
+ // UnsafeRow, which the row reader already produces, so the conversion only
copies each row.
+ Seq(HoodieTableType.COPY_ON_WRITE -> "cow", HoodieTableType.MERGE_ON_READ ->
"mor").foreach { case (tableType, label) =>
+ test(s"Test narrow and wide scans of a timestamp-millis table with the
vectorized reader enabled for $label table") {
+ withSQLConf(vectorizedReaderConf -> "true",
"spark.sql.codegen.maxFields" -> "100") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ writeTimestampMillisTable(tableName, basePath, tableType)
+ // Read by path: without a Hive catalog, a catalog table hands Hudi
its Spark schema, which has no
+ // timestamp-millis type. One view keeps one file format instance
for every scan below.
+
spark.read.format("hudi").load(basePath).createOrReplaceTempView(tableName)
+ val expected = timestampMillisRows.map { case (id, name, eventTime,
payloadBase) =>
+ (id, name, eventTime, payloadBase + 1, payloadBase +
payloadFieldCount)
+ }
+
+ val narrowScan = s"select id, name, event_time from $tableName"
+ // Only COW scans return batches, and only from Spark 3.5 on for
this table
+ assertResult(HoodieSparkUtils.gteqSpark3_5 && tableType ==
HoodieTableType.COPY_ON_WRITE)(scanIsBatched(narrowScan))
+ val narrow = spark.sql(narrowScan).collect().map { row =>
+ (row.getAs[Int]("id"), row.getAs[String]("name"),
row.getAs[Timestamp]("event_time").getTime)
+ }.sorted.toSeq
+ assertResult(expected.map(r => (r._1, r._2, r._3)))(narrow)
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+
+ val wideScan = s"select * from $tableName"
+
assert(WholeStageCodegenExec.isTooManyFields(spark.sessionState.conf,
spark.table(tableName).schema),
+ "select * must be too wide for whole-stage codegen, otherwise
Spark asks supportBatch for it as well")
+ assert(!scanIsBatched(wideScan))
+ val wide = spark.sql(wideScan).collect().map { row =>
+ val payload = row.getAs[Row]("payload")
+ (row.getAs[Int]("id"), row.getAs[String]("name"),
row.getAs[Timestamp]("event_time").getTime,
+ payload.getAs[Int]("f1"),
payload.getAs[Int](s"f$payloadFieldCount"))
+ }.sorted.toSeq
+ assertResult(expected)(wide)
+ assertResult("true")(spark.conf.get(vectorizedReaderConf))
+ spark.catalog.dropTempView(tableName)
+ }
+ }
+ }
+ }
+
+ // Rows are (id, name, event_time in epoch millis, payload base); the second
commit updates id 2
+ private val insertedTimestampMillisRows = Seq(
+ (1, "a1", 1672567200124L, 1000), (2, "a2", 1672567200125L, 2000), (3,
"a3", 1672567200126L, 3000))
+ private val updatedTimestampMillisRow = (2, "a2_updated", 1672653600456L,
20000)
+ private val timestampMillisRows = insertedTimestampMillisRows.map { row =>
+ if (row._1 == updatedTimestampMillisRow._1) updatedTimestampMillisRow else
row
+ }
+
+ // Spark has no timestamp-millis type, so SQL and DataFrame writes cannot
produce this schema
+ private lazy val timestampMillisSchema: Schema = {
+ val payload = (1 to payloadFieldCount)
+ .foldLeft(SchemaBuilder.record("payload").fields())((fields, i) =>
fields.requiredInt(s"f$i"))
+ .endRecord()
+ SchemaBuilder.record("ts_millis_record").fields()
+ .requiredInt("id")
+ .requiredString("name")
+
.name("event_time").`type`(LogicalTypes.timestampMillis().addToSchema(Schema.create(Schema.Type.LONG))).noDefault()
+ .name("payload").`type`(payload).noDefault()
+ .endRecord()
+ }
+
+ private def timestampMillisRecord(row: (Int, String, Long, Int)):
HoodieRecord[IndexedRecord] = {
+ val (id, name, eventTime, payloadBase) = row
+ val payload = new
GenericData.Record(timestampMillisSchema.getField("payload").schema())
+ (1 to payloadFieldCount).foreach(i => payload.put(s"f$i", payloadBase + i))
+ val record = new GenericData.Record(timestampMillisSchema)
+ record.put("id", id)
+ record.put("name", name)
+ record.put("event_time", eventTime)
+ record.put("payload", payload)
+ new HoodieAvroIndexedRecord(new HoodieKey(id.toString, ""), record)
+ }
+
+ /** Writes three rows, then updates one, which lands in a log file on a MOR
table. */
+ private def writeTimestampMillisTable(tableName: String, basePath: String,
tableType: HoodieTableType): Unit = {
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableName(tableName)
+ .setTableType(tableType)
+ .setRecordKeyFields("id")
+ .setPartitionFields("")
+ .setRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING)
+
.initTable(HadoopFSUtils.getStorageConf(spark.sparkContext.hadoopConfiguration),
basePath)
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .forTable(tableName)
+ .withSchema(timestampMillisSchema.toString)
+ .withParallelism(1, 1)
+ .withRecordMergeMode(RecordMergeMode.COMMIT_TIME_ORDERING)
+
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.SIMPLE).build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .build()
+ val jsc = JavaSparkContext.fromSparkContext(spark.sparkContext)
+ val client = new SparkRDDWriteClient[IndexedRecord](new
HoodieSparkEngineContext(jsc), writeConfig)
+ try {
+ Seq(insertedTimestampMillisRows -> true, Seq(updatedTimestampMillisRow)
-> false).foreach { case (rows, isInsert) =>
+ val records = jsc.parallelize(rows.map(timestampMillisRecord).asJava,
1)
+ val instant = client.startCommit()
+ val statuses = if (isInsert) client.insert(records, instant) else
client.upsert(records, instant)
+ assert(client.commit(instant, jsc.parallelize(statuses.collect(), 1)))
+ }
+ } finally {
+ client.close()
+ }
+ }
+
+ // Rows of the wide table as (id, name, dt). Partition p2 holds only id 3,
so each partition is its own file group.
+ private val insertedRows = Seq((1, "a1", "p1"), (2, "a2", "p1"), (3, "a3",
"p2"))
+ private val updatedRows = insertedRows.map {
+ case (3, _, dt) => (3, "a3_updated", dt)
+ case row => row
+ }
+
+ /** Creates and fills the wide table, and returns its path. */
+ private def createWideTable(tableName: String, tableType: String, tmp: File,
+ extraProps: Map[String, String] = Map.empty):
String = {
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+ val payloadType = (1 to payloadFieldCount).map(i => s"f$i
int").mkString("struct<", ", ", ">")
+ val extraTblProps = extraProps.map { case (k, v) => s",\n '$k' = '$v'"
}.mkString
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | payload $payloadType,
+ | ts long,
+ | dt string
+ |) using hudi
+ | partitioned by (dt)
+ | location '$tablePath'
+ | tblproperties (
+ | type = '$tableType',
+ | primaryKey = 'id',
+ | orderingFields = 'ts'$extraTblProps
+ | )
+ """.stripMargin)
+ val rows = insertedRows.map { case (id, name, dt) =>
+ val payload = (1 to payloadFieldCount).map(i => s"'f$i', ${id * 1000 +
i}").mkString("named_struct(", ", ", ")")
+ s"select $id as id, '$name' as name, $payload as payload, ${id * 1000}L
as ts, '$dt' as dt"
+ }
+ spark.sql(s"insert into $tableName ${rows.mkString(" union all ")}")
+ assert(WholeStageCodegenExec.isTooManyFields(spark.sessionState.conf,
spark.table(tableName).schema),
+ "select * must be too wide for whole-stage codegen, otherwise Spark asks
supportBatch for it as well")
+ tablePath
+ }
+
+ /**
+ * Renames id 3, so on a MOR table p2 gets a base+log file slice, read by
the file group reader over the row
+ * base reader, while p1 keeps a base-only slice, read by the base reader,
which is vectorized when the scan
+ * allows it. Every MOR scan after this mixes the two readers.
+ */
+ private def updateOneFileGroup(tableName: String, tableType: String,
tablePath: String): Unit = {
+ spark.sql(s"update $tableName set name = 'a3_updated' where id = 3")
+ assertResult(if (tableType == "mor") Set("p2") else
Set.empty[String])(partitionsWithLogFiles(tablePath))
+ }
+
+ private def partitionsWithLogFiles(tablePath: String): Set[String] = {
+ val files = Files.walk(Paths.get(tablePath))
+ try {
+ files.iterator().asScala
+ .filter(f => Files.isRegularFile(f) &&
f.getFileName.toString.contains(".log.") && !f.toString.contains("/.hoodie/"))
+ .map(_.getParent.getFileName.toString.stripPrefix("dt="))
+ .toSet
+ } finally {
+ files.close()
+ }
+ }
+
+ /** Runs a scan narrower than spark.sql.codegen.maxFields, and checks the
rows that came back. */
+ private def checkNarrowScan(tableName: String, expected: Seq[(Int, String,
String)]): Unit = {
+ val actual = spark.sql(s"select id, name, dt from
$tableName").collect().map { row =>
+ (row.getAs[Int]("id"), row.getAs[String]("name"),
row.getAs[String]("dt"))
+ }.sorted.toSeq
+ assertResult(expected)(actual)
+ }
+
+ /** Runs select * so that the scan stays wide, and checks the rows that came
back. */
+ private def checkWideScan(tableName: String, expected: Seq[(Int, String,
String)]): Unit = {
+ val actual = spark.sql(s"select * from $tableName").collect().map { row =>
+ val payload = row.getAs[Row]("payload")
+ (row.getAs[Int]("id"), row.getAs[String]("name"),
row.getAs[String]("dt"),
+ payload.getAs[Int]("f1"), payload.getAs[Int](s"f$payloadFieldCount"))
+ }.sorted.toSeq
+ assertResult(expected.map { case (id, name, dt) => (id, name, dt, id *
1000 + 1, id * 1000 + payloadFieldCount) })(actual)
+ }
+
+ private def scanIsBatched(sql: String): Boolean = {
+ val plan = spark.sql(sql).queryExecution.executedPlan
+ collectFirst(plan) { case scan: FileSourceScanExec =>
scan.supportsColumnar }
+ .getOrElse(fail(s"No FileSourceScanExec in the plan of [$sql]:\n$plan"))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index c0b6bb5c7934..aa54553799ec 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -1073,9 +1073,11 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "false") {
checkAnswer(s"select id, ts from $tableName order by id")(Seq(1,
1000), Seq(2, 1000))
}
- // count(*) is deliberately not pinned: it fails on any
schema-on-read variant table once
- // an internal schema is committed, rename or not (#20139), and the
existing pin in
- // "Schema-on-read reads of shredded variant files fail fast" is
green only through #20140.
+ // count(*) reads no column data, so it skips the schema-on-read
merge
+ // (ParquetSchemaEvolutionUtils.shouldUseInternalSchema) and counts
on every leg. The merge
+ // used to request the variant as a plain struct, which Spark 4.x's
vectorized reader rejects
+ // for a VARIANT-annotated group, so COW counts failed whether the
file was shredded or not.
+ checkAnswer(s"select count(*) from $tableName")(Seq(2))
}
}
}
@@ -2356,6 +2358,32 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
}
}
+ test("Variant read does not mutate the session vectorized reader
configuration") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // A variant read falls back to row-based reads. That decision belongs to
the scan alone: it used
+ // to be written into spark.sql.parquet.enableVectorizedReader, so every
later read in the session
+ // lost vectorization.
+ withVariantTable(
+ "vectorized reader session conf",
+ "cow",
+ recordTypes = Seq(HoodieRecordType.SPARK)) { (tableName, _, _) =>
+
+ spark.sql(
+ s"""insert into $tableName values (1, parse_json('{"key":"v1"}'),
1000)""")
+
+ withSQLConf("spark.sql.parquet.enableVectorizedReader" -> "true") {
+ checkAnswer(
+ s"select id, cast(v as string) from $tableName")(
+ Seq(1, """{"key":"v1"}"""))
+
+ assert(
+ spark.conf.get("spark.sql.parquet.enableVectorizedReader") == "true",
+ "A query-local vectorization fallback must not mutate the session
configuration")
+ }
+ }
+ }
+
//
-----------------------------------------------------------------------------------------------
// G. Schema evolution
//
-----------------------------------------------------------------------------------------------
