This is an automated email from the ASF dual-hosted git repository.
voonhous 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 842e84e4d4fe test(variant): pin Spark 3.x schema-on-read rejection of
variant tables (#19691)
842e84e4d4fe is described below
commit 842e84e4d4fe2fe858408142adc914d681fcb274
Author: voonhous <[email protected]>
AuthorDate: Mon Aug 24 20:15:23 2026 +0800
test(variant): pin Spark 3.x schema-on-read rejection of variant tables
(#19691)
* test(variant): pin Spark 3.x schema-on-read rejection of variant tables
Verifies #18021: with hoodie.schema.on.read.enable, Spark 3.x resolves
the schema through the InternalSchema round trip, whose sentinel
detection restores the VARIANT logical type - the input
HoodieSparkSchemaConverters rejects on Spark 3.x. Every leg fails
loudly with the same actionable error as plain auto-resolution
(HoodieSchemaException caused by 'VARIANT type is only supported in
Spark 4.0+'); no silent wrong data, no obscure secondary failure.
The notable interaction: the documented struct-DDL compat mode works
without schema-on-read but breaks once the table carries an internal
schema and the conf is on, because internal-schema resolution overrides
the user's DDL. Real support is #18285; until then Spark 3.x
compat-mode readers must keep the conf off. Fixture: a Spark
4.1-written COW variant table with a committed internal schema.
* review(variant): CI-enforce the pin, prove both sides of the compat
interaction
- Add org.apache.spark.sql.hudi.dml.schema to the Azure DDL & Others
wildcard set so the Spark 3-gated pins actually run on the spark3.5
lane (the package was in no Azure set, and the GHA dml lane is
spark4.2-only).
- Assert the fixture carries a loadable internal schema so the
schema-on-read legs cannot pass via the commit-metadata fallback.
- Add a conf-off struct-DDL leg on the same fixture, proving the conf,
not the committed internal schema itself, is what breaks compat mode.
- Record the exact seeding statements in the fixture README so the zip
is regenerable.
- Drop the answered #18021 TODO in HoodieSparkSchemaConverters.
* review(variant): narrow the pin's claims and guard the Azure leaf split
- README: the rejection test reads `latest_schema` out of the newest
schema-capable commit's extra metadata, not `.hoodie/.schema/`, so point
the regeneration check there (the first insert in this fixture has the
schemacommit files and no `latest_schema`).
- Scope the "keep hoodie.schema.on.read.enable off" advice to the non-Hive
catalog: DefaultSource discards the user schema when isUsingHiveCatalog,
so under HMS the struct DDL read throws with the conf off too.
- Narrow the fixture assert's claim: HoodieBaseHadoopFsRelationFactory
(reached via HoodieCopyOnWriteSnapshotHadoopFsRelationFactory, not
HoodieBaseRelation) swallows an internal-schema load failure into None
and falls back to getTableSchema, so the assert pins the fixture, not the
path; the conf-off/conf-on catalog-DDL pair is what discriminates.
- Pin the exact variant value/metadata bytes on the conf-off leg instead of
just isNullAt, so a swapped or malformed pair cannot pass.
- Guard the Azure leaf split in TestSparkSqlHudiPackageStructure: assert
every Scala test package under org.apache.spark.sql.hudi is named by some
wildcardSuites entry, so the next dml.* package cannot start out dark.
- Rename the assertVariantRejected parameter leg -> label.
* review(ci): assert an Azure step actually runs each Scala test package
The previous guard only proved a package was named by some wildcardSuites
entry. It did not tie that entry back to a step that expands it, nor check
the step's -pl builds the module holding the class - so a parameter list no
step referenced, or a step that stopped building hudi-spark, still read as
covered.
Resolve the pipeline properly instead: parse the parameters: lists, map the
variables: entries that join them, then for each step carrying both
-DwildcardSuites and -pl expand the $(VAR) tokens and record the pair. A
class counts as run only when some step both builds its module and names its
package. Renamed to testScalaTestPackagesAreRunByAnAzureStep to match.
Verified it still passes on the tree as-is and fails on all three of: a new
dml.darkpkg package, deleting the step that expands
JOB6_SPARK_DDL_OTHERS_WILDCARD_SUITES while keeping the list, and pointing
that step's -pl at a module other than hudi-spark.
---
azure-pipelines-20230430.yml | 1 +
.../sql/avro/HoodieSparkSchemaConverters.scala | 1 -
.../resources/variant_backward_compat/README.md | 39 ++++
.../variant_schema_on_read_cow.zip | Bin 0 -> 105946 bytes
.../hudi/TestSparkSqlHudiPackageStructure.java | 221 ++++++++++++++++++++-
.../sql/hudi/dml/schema/TestVariantDataType.scala | 102 ++++++++++
6 files changed, 356 insertions(+), 8 deletions(-)
diff --git a/azure-pipelines-20230430.yml b/azure-pipelines-20230430.yml
index a8bd40223295..617bc4d9e418 100644
--- a/azure-pipelines-20230430.yml
+++ b/azure-pipelines-20230430.yml
@@ -80,6 +80,7 @@ parameters:
- 'org.apache.spark.sql.hudi.command'
- 'org.apache.spark.sql.hudi.common'
- 'org.apache.spark.sql.hudi.ddl'
+ - 'org.apache.spark.sql.hudi.dml.schema'
- 'org.apache.spark.sql.hudi.procedure'
- name: jacocoModules
type: object
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
index 2ba3fcfc158b..dc925e60a624 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
@@ -395,7 +395,6 @@ object HoodieSparkSchemaConverters extends
SparkAdapterSupport {
}
// VARIANT type (Spark >4.x only), which will be handled via SparkAdapter
- // TODO: Check if internalSchema will throw any errors here: #18021
case HoodieSchemaType.VARIANT =>
sparkAdapter.getVariantDataType match {
case Some(variantType) => SchemaType(variantType, nullable = false)
diff --git a/hudi-common/src/test/resources/variant_backward_compat/README.md
b/hudi-common/src/test/resources/variant_backward_compat/README.md
index 94d45541b0bb..d297feeb9393 100644
--- a/hudi-common/src/test/resources/variant_backward_compat/README.md
+++ b/hudi-common/src/test/resources/variant_backward_compat/README.md
@@ -39,3 +39,42 @@ The test runs on these four arguments:
COW tables generated are the same for both AVRO/SPARK. But for MOR, the log
files metadata are
different. Hence, we only need to generate test files for either 1/2, 3 and 4,
hence, 3 test
resource files.
+
+# variant_schema_on_read_cow.zip
+
+A Spark 4.1-written COW variant table carrying a COMMITTED INTERNAL SCHEMA:
one insert, then a
+schema-on-read DDL, then a second insert. Used by the Spark 3.x schema-on-read
rejection test
+(#18021). Generated by running these exact statements in a Spark 4.1 SQL
session:
+
+```sql
+create table variant_schema_on_read_cow (
+ id int,
+ v variant,
+ ts long
+) using hudi
+location '/path/to/table/parent/variant_schema_on_read_cow'
+tblproperties (
+ primaryKey = 'id',
+ preCombineField = 'ts'
+);
+
+insert into variant_schema_on_read_cow select 1, parse_json('{"key": "v1"}'),
1000;
+
+set hoodie.schema.on.read.enable=true;
+alter table variant_schema_on_read_cow add columns (note string);
+insert into variant_schema_on_read_cow select 2, parse_json('{"key": "v2"}'),
1000, 'n2';
+```
+
+and zipping the table directory:
+
+```shell
+cd /path/to/table/parent/
+zip -r variant_schema_on_read_cow.zip variant_schema_on_read_cow
+```
+
+The rejection test asserts
`TableSchemaResolver.getTableInternalSchemaFromCommitMetadata`, which
+reads the `latest_schema` key out of the newest schema-capable commit's extra
metadata - NOT out of
+`.hoodie/.schema/`, which is a separate store. So after regenerating, confirm
the last commit in
+`.hoodie/timeline/` carries `latest_schema`; the presence of
`.hoodie/.schema/` schemacommit files
+does not imply it (the first insert here has those and no `latest_schema`).
The commit is Avro, so
+`strings <instant>_<completion>.commit | grep latest_schema` is enough to
check.
diff --git
a/hudi-common/src/test/resources/variant_backward_compat/variant_schema_on_read_cow.zip
b/hudi-common/src/test/resources/variant_backward_compat/variant_schema_on_read_cow.zip
new file mode 100644
index 000000000000..448ba2c7f758
Binary files /dev/null and
b/hudi-common/src/test/resources/variant_backward_compat/variant_schema_on_read_cow.zip
differ
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestSparkSqlHudiPackageStructure.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestSparkSqlHudiPackageStructure.java
index ddce176eb738..c37525c8e356 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestSparkSqlHudiPackageStructure.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestSparkSqlHudiPackageStructure.java
@@ -21,13 +21,20 @@ package org.apache.hudi;
import org.junit.jupiter.api.Test;
import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -42,6 +49,18 @@ public class TestSparkSqlHudiPackageStructure {
private static final String BASE_PACKAGE = "org.apache.spark.sql.hudi";
private static final String PACKAGE_PATH = "org/apache/spark/sql/hudi";
+ private static final String AZURE_PIPELINE_FILE =
"azure-pipelines-20230430.yml";
+ private static final String SPARK_DATASOURCE_DIR = "hudi-spark-datasource";
+
+ private static final Pattern PARAM_NAME = Pattern.compile("^\\s*-
name:\\s*(\\S+)\\s*$");
+ private static final Pattern LIST_ITEM = Pattern.compile("^\\s*-
'([^']*)'\\s*$");
+ /** A {@code variables:} entry that is just a join over a parameter list,
e.g. {@code
+ * JOB3456_MODULES: ${{ join(',',parameters.job3456UTModules) }}}. */
+ private static final Pattern VARIABLE_JOIN =
+
Pattern.compile("^\\s*([A-Za-z0-9_]+):\\s*\\$\\{\\{\\s*join\\('[^']*',\\s*parameters\\.([A-Za-z0-9_]+)\\)\\s*\\}\\}\\s*$");
+ private static final Pattern WILDCARD_CLI_ARG =
Pattern.compile("-DwildcardSuites=\"?([^\" ]+)");
+ private static final Pattern PL_CLI_ARG = Pattern.compile("-pl\\s+\"?([^\"
]+)");
+ private static final Pattern VAR_REF =
Pattern.compile("\\$\\(([A-Za-z0-9_]+)\\)");
/**
* Allowed sub-packages under org.apache.spark.sql.hudi for Scala test
classes.
@@ -89,6 +108,180 @@ public class TestSparkSqlHudiPackageStructure {
"Expected to find at least one Scala test class in " + BASE_PACKAGE);
}
+ /**
+ * Every Scala test class under {@link #BASE_PACKAGE} must be run by at
least one Azure step:
+ * some step whose {@code -pl} builds the module holding the class must also
pass a
+ * {@code -DwildcardSuites} prefix matching it. Otherwise the class silently
never runs on Azure.
+ *
+ * <p>The Azure jobs deliberately name leaf packages ({@code dml.others},
{@code dml.insert},
+ * {@code dml.schema}) rather than the recursive {@code dml} parent, because
ScalaTest's
+ * {@code -w} is a plain prefix match with no exclusion primitive: pointing
one job at
+ * {@code ...hudi.dml} would re-run the whole {@code dml.insert} set that
already has its own
+ * job. That split is what makes a newly added {@code dml.*} package start
out dark, so this
+ * test is the guard for it - the other, non-recursive {@code
testSparkSqlHudi...} check above
+ * lets {@code dml.*} through because it treats {@code dml} as one allowed
package.
+ */
+ @Test
+ public void testScalaTestPackagesAreRunByAnAzureStep() {
+ List<AzureScalaTestStep> steps = readAzureScalaTestSteps();
+ assertFalse(steps.isEmpty(),
+ "Expected to parse at least one -DwildcardSuites step from " +
AZURE_PIPELINE_FILE
+ + "; the parsing below has probably drifted from the pipeline's
shape");
+
+ Map<String, List<String>> classesByModule = findScalaTestClassesByModule();
+ List<String> uncovered = new ArrayList<>();
+ for (Map.Entry<String, List<String>> entry : classesByModule.entrySet()) {
+ String module = entry.getKey();
+ for (String className : entry.getValue()) {
+ boolean run = steps.stream().anyMatch(step -> step.builds(module) &&
step.runs(className));
+ if (!run) {
+ uncovered.add(className + " (module " + module + ")");
+ }
+ }
+ }
+
+ if (!uncovered.isEmpty()) {
+ StringBuilder message = new StringBuilder();
+ message.append("Found Scala test classes that no Azure step runs - no
step both builds ")
+ .append("their module (-pl) and names their package
(-DwildcardSuites), so they never ")
+ .append("run on Azure CI.\n\nUncovered classes:\n");
+ uncovered.forEach(cls -> message.append(" -
").append(cls).append("\n"));
+ message.append("\nAdd their package to one of the wildcardSuites sets in
")
+ .append(AZURE_PIPELINE_FILE)
+ .append(" (e.g. 'job6HudiSparkDdlOthersWildcardSuites'), balancing
against job runtime.");
+ fail(message.toString());
+ }
+ }
+
+ /**
+ * One Azure step that runs ScalaTest: the suite prefixes it passes to
{@code -DwildcardSuites}
+ * and the module list it passes to {@code -pl}, both with {@code $(VAR)}
references resolved
+ * back to the parameter list the {@code variables:} block joins them from.
+ */
+ private static final class AzureScalaTestStep {
+ private final List<String> suitePrefixes;
+ private final List<String> modules;
+
+ private AzureScalaTestStep(List<String> suitePrefixes, List<String>
modules) {
+ this.suitePrefixes = suitePrefixes;
+ this.modules = modules;
+ }
+
+ /** Whether this step's {@code -pl} selects the given module. */
+ private boolean builds(String module) {
+ if (modules.isEmpty()) {
+ // no -pl at all means the whole reactor
+ return true;
+ }
+ if (modules.contains("!" + module)) {
+ return false;
+ }
+ // an all-exclusion list selects everything it does not name; otherwise
it is an include list
+ boolean allExclusions = modules.stream().allMatch(m ->
m.startsWith("!"));
+ return allExclusions || modules.contains(module);
+ }
+
+ /** Whether this step's {@code -DwildcardSuites} names the given class.
ScalaTest's -w is a
+ * prefix match; the dot boundary keeps this check strictly narrower than
what -w accepts. */
+ private boolean runs(String className) {
+ return suitePrefixes.stream()
+ .anyMatch(prefix -> className.equals(prefix) ||
className.startsWith(prefix + "."));
+ }
+ }
+
+ /**
+ * Parses the pipeline into the ScalaTest steps it defines. Resolves {@code
$(VAR)} tokens
+ * through the {@code variables:} block back to the {@code parameters:} list
they join, so a
+ * parameter list that no step actually expands contributes nothing.
+ */
+ private List<AzureScalaTestStep> readAzureScalaTestSteps() {
+ List<String> lines = readAzurePipelineLines();
+
+ // parameters: <name> -> its list of values
+ Map<String, List<String>> parameterLists = new HashMap<>();
+ String currentParam = null;
+ for (String line : lines) {
+ Matcher paramName = PARAM_NAME.matcher(line);
+ if (paramName.matches()) {
+ currentParam = paramName.group(1);
+ parameterLists.put(currentParam, new ArrayList<>());
+ continue;
+ }
+ Matcher listItem = LIST_ITEM.matcher(line);
+ if (currentParam != null && listItem.matches()) {
+ parameterLists.get(currentParam).add(listItem.group(1).trim());
+ }
+ }
+
+ // variables: <VAR> -> the parameter list it joins
+ Map<String, List<String>> variableValues = new HashMap<>();
+ for (String line : lines) {
+ Matcher variable = VARIABLE_JOIN.matcher(line);
+ if (variable.matches()) {
+ List<String> values = parameterLists.get(variable.group(2));
+ if (values != null) {
+ variableValues.put(variable.group(1), values);
+ }
+ }
+ }
+
+ // steps: every line carrying both a wildcardSuites filter and a -pl
module list
+ List<AzureScalaTestStep> steps = new ArrayList<>();
+ for (String line : lines) {
+ Matcher suites = WILDCARD_CLI_ARG.matcher(line);
+ Matcher modules = PL_CLI_ARG.matcher(line);
+ if (!suites.find() || !modules.find()) {
+ continue;
+ }
+ List<String> suitePrefixes = expand(suites.group(1),
variableValues).stream()
+ .filter(suite -> suite.startsWith("org."))
+ .collect(Collectors.toList());
+ if (suitePrefixes.isEmpty()) {
+ // e.g. the java-only steps, which pass -DwildcardSuites=skipScalaTests
+ continue;
+ }
+ steps.add(new AzureScalaTestStep(suitePrefixes, expand(modules.group(1),
variableValues)));
+ }
+ return steps;
+ }
+
+ /**
+ * Splits a comma-separated Azure argument into its entries, replacing any
{@code $(VAR)} token
+ * with the entries of the parameter list that variable joins.
+ */
+ private List<String> expand(String argument, Map<String, List<String>>
variableValues) {
+ List<String> expanded = new ArrayList<>();
+ for (String token : argument.split(",")) {
+ String trimmed = token.trim();
+ Matcher varRef = VAR_REF.matcher(trimmed);
+ if (varRef.matches()) {
+ expanded.addAll(variableValues.getOrDefault(varRef.group(1), new
ArrayList<>()));
+ } else if (!trimmed.isEmpty()) {
+ expanded.add(trimmed);
+ }
+ }
+ return expanded;
+ }
+
+ private List<String> readAzurePipelineLines() {
+ File projectRoot = findProjectRoot();
+ if (projectRoot == null) {
+ fail("Could not locate project root directory");
+ return new ArrayList<>();
+ }
+ File pipeline = new File(projectRoot, AZURE_PIPELINE_FILE);
+ if (!pipeline.exists()) {
+ fail("Could not locate " + AZURE_PIPELINE_FILE + " at " +
pipeline.getAbsolutePath());
+ return new ArrayList<>();
+ }
+ try {
+ return Files.readAllLines(pipeline.toPath(), StandardCharsets.UTF_8);
+ } catch (Exception e) {
+ fail("Could not read " + AZURE_PIPELINE_FILE + ": " + e.getMessage());
+ return new ArrayList<>();
+ }
+ }
+
/**
* Checks if a class is in one of the allowed packages (including
sub-packages).
*/
@@ -105,20 +298,31 @@ public class TestSparkSqlHudiPackageStructure {
* Finds all Scala test classes under org.apache.spark.sql.hudi by scanning
source directories.
*/
private List<String> findScalaTestClasses() {
- List<String> classes = new ArrayList<>();
+ return findScalaTestClassesByModule().values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Finds all Scala test classes under org.apache.spark.sql.hudi, keyed by
the Maven module path
+ * that holds them (as Azure would name it in {@code -pl}), e.g.
+ * {@code hudi-spark-datasource/hudi-spark}.
+ */
+ private Map<String, List<String>> findScalaTestClassesByModule() {
+ Map<String, List<String>> classesByModule = new TreeMap<>();
// Find project root by traversing up from current class location
File projectRoot = findProjectRoot();
if (projectRoot == null) {
fail("Could not locate project root directory");
- return classes;
+ return classesByModule;
}
// Scan all hudi-spark-datasource modules for Scala test sources
- File sparkDatasourceDir = new File(projectRoot, "hudi-spark-datasource");
+ File sparkDatasourceDir = new File(projectRoot, SPARK_DATASOURCE_DIR);
if (!sparkDatasourceDir.exists()) {
- fail("Could not locate hudi-spark-datasource directory");
- return classes;
+ fail("Could not locate " + SPARK_DATASOURCE_DIR + " directory");
+ return classesByModule;
}
// Look for Scala test directories in all submodules
@@ -127,12 +331,15 @@ public class TestSparkSqlHudiPackageStructure {
for (File submodule : submodules) {
File scalaTestDir = new File(submodule, "src/test/scala/" +
PACKAGE_PATH);
if (scalaTestDir.exists() && scalaTestDir.isDirectory()) {
- classes.addAll(findScalaFilesRecursively(scalaTestDir,
BASE_PACKAGE));
+ List<String> classes = findScalaFilesRecursively(scalaTestDir,
BASE_PACKAGE);
+ if (!classes.isEmpty()) {
+ classesByModule.put(SPARK_DATASOURCE_DIR + "/" +
submodule.getName(), classes);
+ }
}
}
}
- return classes;
+ return classesByModule;
}
/**
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index 007fbd26ca79..9c88590cb72e 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -24,6 +24,7 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.HoodieSchemaException
+import org.apache.hudi.common.table.TableSchemaResolver
import org.apache.hudi.common.testutils.HoodieTestUtils
import org.apache.hudi.common.util.StringUtils
import org.apache.hudi.testutils.DataSourceTestUtils
@@ -977,6 +978,107 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
}
}
+ test("Test Spark 3.x schema-on-read reads of a variant table with a
committed internal schema") {
+ // #18021: hoodie.schema.on.read.enable resolves the schema through the
InternalSchema round
+ // trip, whose sentinel detection restores the VARIANT logical type - the
exact input
+ // HoodieSparkSchemaConverters rejects on Spark 3.x. Verified 2026-08-20:
every leg fails
+ // LOUDLY with the same actionable error as the plain auto-resolve path;
there is no silent
+ // wrong data and no obscure secondary failure. Notably that includes the
documented
+ // struct-DDL compat mode, which works on this same table with the conf
off (pinned below)
+ // but breaks once it is on, because internal-schema resolution overrides
the user's DDL.
+ // Real support is #18285; until then Spark 3.x compat-mode readers must
keep
+ // hoodie.schema.on.read.enable off - necessary, but only sufficient off a
Hive catalog:
+ // DefaultSource discards the user schema outright when
isUsingHiveCatalog, so under HMS the
+ // struct DDL below is never seen and the read throws with the conf off
too. This test runs
+ // on the in-memory catalog, so it pins the non-Hive half only.
+ assume(HoodieSparkUtils.isSpark3, "This test verifies Spark 3.x behavior
with schema-on-read")
+
+ withTempDir { tmpDir =>
+
HoodieTestUtils.extractZipToDirectory("variant_backward_compat/variant_schema_on_read_cow.zip",
tmpDir.toPath, getClass)
+ val tablePath =
tmpDir.toPath.resolve("variant_schema_on_read_cow").toString
+
+ // HoodieBaseHadoopFsRelationFactory (reached here via
+ // HoodieCopyOnWriteSnapshotHadoopFsRelationFactory, not
HoodieBaseRelation) swallows an
+ // internal-schema load failure into None and then falls back to
getTableSchema, which
+ // throws the same exception - so neither auto-resolve leg can, on its
own, prove the
+ // InternalSchema path ran. This assert only pins that the fixture still
carries a loadable
+ // internal schema; the conf-off/conf-on catalog-DDL pair below is what
discriminates,
+ // because with a DDL present the None fallback resolves to the user
schema and succeeds.
+ val schemaResolver = new TableSchemaResolver(createMetaClient(spark,
tablePath))
+ assert(schemaResolver.getTableInternalSchemaFromCommitMetadata.isPresent,
+ "fixture must carry a committed internal schema; regenerate it per the
README")
+
+ def assertVariantRejected(label: String)(f: => Unit): Unit = {
+ val ex = intercept[HoodieSchemaException](f)
+ assert(ex.getCause.getMessage.contains("VARIANT type is only supported
in Spark 4.0+"),
+ s"[$label] expected the actionable variant rejection, got:
${ex.getCause}")
+ }
+
+ assertVariantRejected("auto-resolve, plain") {
+ spark.read.format("hudi").load(tablePath).collect()
+ }
+ assertVariantRejected("auto-resolve, schema-on-read") {
+ spark.read.format("hudi").option("hoodie.schema.on.read.enable",
"true").load(tablePath).collect()
+ }
+
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v struct<value: binary, metadata: binary>,
+ | ts long,
+ | note string
+ |) using hudi
+ |location '$tablePath'
+ |tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ |)
+ """.stripMargin)
+ try {
+ // With the conf off the compat struct DDL works even though the table
carries an
+ // internal schema - pinning that the conf, not the committed internal
schema itself,
+ // is what breaks compat mode.
+ val rows = spark.sql(s"select id, v, note from $tableName order by
id").collect()
+ assert(rows.map(r => (r.getInt(0), r.getString(2))).toSeq == Seq((1,
null), (2, "n2")),
+ "[compat struct DDL, conf off] expected both rows readable through
the struct DDL")
+
+ // Pin the variant bytes, not just non-nullness, so a swapped or
structurally wrong
+ // value/metadata pair cannot pass as "no silent wrong data". The DDL
orders the struct
+ // as (value, metadata); Spark resolves the parquet fields by name.
+ // metadata: v1 header, 1-entry dictionary holding "key" (shared by
both rows).
+ val expectedMetadata = Array[Byte](0x01, 0x01, 0x00, 0x03, 0x6B, 0x65,
0x79)
+ // value: 1-field object, field id 0, data offsets 0..3, short-string
"v1" / "v2".
+ val expectedValues = Map(
+ 1 -> Array[Byte](0x02, 0x01, 0x00, 0x00, 0x03, 0x09, 0x76, 0x31),
+ 2 -> Array[Byte](0x02, 0x01, 0x00, 0x00, 0x03, 0x09, 0x76, 0x32))
+ rows.foreach { row =>
+ val id = row.getInt(0)
+ assert(!row.isNullAt(1), s"[compat struct DDL, conf off] variant
column null for id=$id")
+ val variant = row.getStruct(1)
+ assert(variant.size == 2, s"[compat struct DDL, conf off] expected
(value, metadata) for id=$id")
+ val valueBytes = variant.getAs[Array[Byte]](0)
+ val metadataBytes = variant.getAs[Array[Byte]](1)
+ assert(valueBytes.sameElements(expectedValues(id)),
+ s"[compat struct DDL, conf off] variant value bytes mismatch for
id=$id, got: "
+ + valueBytes.map("0x%02X".format(_)).mkString(", "))
+ assert(metadataBytes.sameElements(expectedMetadata),
+ s"[compat struct DDL, conf off] variant metadata bytes mismatch
for id=$id, got: "
+ + metadataBytes.map("0x%02X".format(_)).mkString(", "))
+ }
+
+ assertVariantRejected("compat struct DDL, schema-on-read") {
+ withSQLConf("hoodie.schema.on.read.enable" -> "true") {
+ spark.sql(s"select id, v, note from $tableName order by
id").collect()
+ }
+ }
+ } finally {
+ spark.sql(s"drop table $tableName")
+ }
+ }
+ }
+
test(s"Test Backward Compatibility: Read Spark 4.0 Variant Table in Spark
3.x") {
// This test only runs on Spark 3.x to verify backward compatibility
assume(HoodieSparkUtils.isSpark3, "This test verifies Spark 3.x can read
Spark 4.0 Variant tables")
