voonhous opened a new issue, #19556:
URL: https://github.com/apache/hudi/issues/19556
**Describe the problem you faced**
On a COW table with a VARIANT column whose parquet files carry the shredded
layout,
inline clustering rewrites carried-over rows with `v = NULL`, silently
losing the variant
values. The replacecommit completes cleanly and no error is raised; the
nulls are physical
in the new base files, so once the cleaner removes the pre-clustering files
the data is
unrecoverable.
Only rows that go through the clustering rewrite lose data. Rows whose file
group is not
in the clustering plan keep their values, which is what proves the write
side (not the
reader) dropped them.
Shredding is enabled by default
(`hoodie.parquet.variant.write.shredding.enabled=true`).
A control experiment with the identical scenario and shredding disabled
passes, so the
loss is specific to reading the shredded layout inside the clustering
rewrite, not to
variant clustering in general.
**To Reproduce**
Spark SQL on master under the spark4.1 profile (the shredding schema is
forced via the
test-only config to deterministically produce shredded files):
```sql
create table t (id int, v variant, ts long) using hudi
location '/tmp/variant_cluster_repro'
tblproperties (
primaryKey = 'id', type = 'cow', preCombineField = 'ts',
hoodie.parquet.variant.write.shredding.enabled = 'true',
hoodie.parquet.variant.force.shredding.schema.for.test = 'key string',
hoodie.index.type = 'INMEMORY',
hoodie.clustering.inline = 'true',
hoodie.clustering.inline.max.commits = '2'
);
insert into t values (1, parse_json('{"key":"value1"}'), 1000),
(2, parse_json('{"key":"value2"}'), 1000);
-- second commit trips inline clustering
insert into t values (3, parse_json('{"key":"value3"}'), 1000),
(4, parse_json('{"key":"value4"}'), 1000);
select id, cast(v as string), ts from t order by id;
```
Observed (rows 1 and 2 went through the clustering rewrite; 3 and 4 did not):
```
Expected: [1,{"key":"value1"},1000], [2,{"key":"value2"},1000],
[3,{"key":"value3"},1000], [4,{"key":"value4"},1000]
Got: [1,null,1000], [2,null,1000],
[3,{"key":"value3"},1000], [4,{"key":"value4"},1000]
```
Test-form repro: add the method below to
`org.apache.spark.sql.hudi.dml.schema.TestVariantDataType` (it mirrors the
SQL above and
asserts a COMPLETED replacecommit exists so the failure cannot be a
clustering no-op):
```scala
test("Test COW clustering preserves VARIANT values") {
assume(HoodieSparkUtils.gteqSpark4_1, "Shredded variant base-file read
requires Spark 4.1 or higher")
withRecordType()(withTempDir { tmp =>
val tableName = generateTableName
val tablePath = tmp.getCanonicalPath
spark.sql(
s"""
|create table $tableName (id int, v variant, ts long) using hudi
| location '$tablePath'
| tblproperties (
| primaryKey = 'id', type = 'cow', preCombineField = 'ts',
| hoodie.parquet.variant.write.shredding.enabled = 'true',
| hoodie.parquet.variant.force.shredding.schema.for.test = 'key
string',
| hoodie.index.type = 'INMEMORY',
| hoodie.clustering.inline = 'true',
| hoodie.clustering.inline.max.commits = '2'
| )
""".stripMargin)
spark.sql(s"insert into $tableName values " +
"(1, parse_json('{\"key\":\"value1\"}'), 1000), " +
"(2, parse_json('{\"key\":\"value2\"}'), 1000)")
spark.sql(s"insert into $tableName values " +
"(3, parse_json('{\"key\":\"value3\"}'), 1000), " +
"(4, parse_json('{\"key\":\"value4\"}'), 1000)")
val metaClient = createMetaClient(spark, tablePath)
val lastClustering =
metaClient.getActiveTimeline.getLastClusteringInstant
assert(lastClustering.isPresent && lastClustering.get.isCompleted,
"A COMPLETED clustering (replacecommit) instant must exist after
inline clustering")
checkAnswer(s"select id, cast(v as string), ts from $tableName order by
id")(
Seq(1, "{\"key\":\"value1\"}", 1000),
Seq(2, "{\"key\":\"value2\"}", 1000),
Seq(3, "{\"key\":\"value3\"}", 1000),
Seq(4, "{\"key\":\"value4\"}", 1000)
)
})
}
```
```bash
export JAVA_HOME=$(/usr/libexec/java_home -v 17)
mvn test -Dspark4.1 -pl hudi-spark-datasource/hudi-spark -am \
-Dtest=none -Dsurefire.failIfNoSpecifiedTests=false \
-DwildcardSuites='org.apache.spark.sql.hudi.dml.schema.TestVariantDataType'
```
The identical test with `hoodie.parquet.variant.write.shredding.enabled =
'false'` (and no
force schema) passes.
**Expected behavior**
Variant values survive clustering regardless of the physical layout of the
input files.
**Suspected cause (differentially verified, code path unverified)**
A shredded parquet file stores the data in `typed_value`; the `value` field
is physically
null. User queries reconstruct that via Spark 4.1's `PushVariantIntoScan`
catalyst rule.
Clustering reads the file groups through
`MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow`, i.e. the
internal write-side
reader stack (`SparkReaderContextFactory` ->
`SparkFileFormatInternalRowReaderContext`),
which does not go through the catalyst optimizer -- so nothing reconstructs
`typed_value`,
the read yields `{metadata, value=null}`, and the row writer persists the
nulls into the
clustered base files.
This is the VARIANT twin of #19232 (compaction/clustering dropping INLINE
blob bytes):
same internal write-side reader path, same silent-loss shape. The fix likely
mirrors it --
engage shredded-variant reconstruction (or the variant projection overlay)
on the internal
reader context used by clustering.
Notes on adjacent paths, from the same test suite run:
* MOR compaction with shredded variants passes ("Test Query Log Only MOR
Table With
VARIANT column triggers compaction") -- compaction merges from log records
via a
different flow.
* Unshredded clustering passes.
**Environment Description**
* Hudi version: master (1.3.0-SNAPSHOT, c4b38935db04)
* Spark version: 4.1 profile (Scala 2.13, JDK 17)
* Table type: COPY_ON_WRITE, parquet base files
* Storage: local FS
* Running on Docker: no
--
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]
