voonhous commented on code in PR #19687: URL: https://github.com/apache/hudi/pull/19687#discussion_r3871028864
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala: ########## @@ -0,0 +1,1208 @@ +/* + * 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 +import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType +import org.apache.hudi.core.io.storage.VariantShreddingInferenceFileWriter +import org.apache.hudi.testutils.DataSourceTestUtils + +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase +import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase.getLastCommitMetadata + +/** + * Mixed-layout variant shredding matrix: files with DIFFERENT typed_value layouts in one table, + * shredded/unshredded splits between base and log files, and rows inside one file that fell back + * to the residual value column, driven through compaction, clustering, merges and every Spark + * read mode. Complements [[TestVariantDataType]], whose shredded tests force ONE layout per + * table. + * + * Layouts are toggled per commit or table service through session confs (session hoodie.* confs + * override tblproperties for SQL DML and for the run_compaction/run_clustering procedures alike). + * Every test is gated on Spark 4.1+, which is exactly the set of profiles that register #18961's + * per-file shredding-schema inferrer (pinned by TestVariantDataType's "A shredding-schema + * inferrer is registered for every Spark version that ships one"), so an [[Inferred]] leg here + * always infers rather than silently degrading to an unshredded write. + * + * Deliberately not covered here: + * - Custom payloads: FileGroupRecordBuffer.getProjectedTransformer short-circuits the variant + * log-block projection when payload classes are present (#18674), so that is a real, + * explicitly UNTESTED variant branch; PartialUpdateMode and the CUSTOM merge mode are + * likewise unreached (EVENT_TIME ordering is swept throughout, COMMIT_TIME in the one leg + * that drops preCombineField). + * - Multi-writer OCC: conflict resolution is key/instant based and never inspects layouts; the + * mixed-file outcomes it can produce are the same ones pinned here. + */ +class TestVariantShreddingMixedLayouts extends HoodieSparkSqlTestBase with VariantShreddingTestSupport { + + import VariantShreddingTestSupport._ + import VariantShreddingTestSupport.VariantShape._ + + private val SPARK_4_1_GATE = "Shredded variant read-back requires Spark 4.1 or higher" + + /** One insert commit per layout; returns the completed instant of each commit, in order. */ + private def seedMixedLayoutTable(tableName: String, + tablePath: String, + layouts: Seq[(WriteLayout, Seq[(Range, VariantShape)])]): Seq[String] = { + layouts.map { case (layout, segments) => + withWriteLayout(layout) { + spark.sql(s"insert into $tableName ${variantSourceSql(segments)}") + } + latestCompletedInstant(tablePath) + } + } + + /** scheduleAndExecute compaction; the options carry the NUM_COMMITS trigger so one delta commit suffices. */ + private def runCompaction(tableName: String): Unit = { + spark.sql(s"call run_compaction(op => 'scheduleandexecute', table => '$tableName', " + + "options => 'hoodie.compact.inline.max.delta.commits=1')") + } + + private def runClustering(tableName: String, rowWriter: Boolean): Unit = { + spark.sql(s"call run_clustering(table => '$tableName', " + + s"options => 'hoodie.datasource.write.row.writer.enable=$rowWriter')") + } + + // ----------------------------------------------------------------------------------------------- + // A. Mixed records inside one file + // ----------------------------------------------------------------------------------------------- + + test("Forced shredding: non-matching rows fall back to the residual in the same file") { + assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE) + + withVariantTable("same-file mix", "cow") { (tableName, tablePath, leg) => + // One insert, one file: rows 0-9 match the forced schema exactly; 10-14 conflict on the + // type of a (string into a bigint slot -> per-field residual); 15-19 carry disjoint keys + // (root residual); 20-22 are root scalars and 23 a JSON null (no object typed_value); + // 24 is a SQL NULL variant. + val segments = Seq( + (0 until 10, ObjA), + (10 until 15, ObjAConflict), + (15 until 20, ObjB), + (20 until 23, RootScalar), + (23 until 24, JsonNull), + (24 until 25, SqlNull)) + withWriteLayout(Forced("a bigint, b string")) { + spark.sql(s"insert into $tableName ${variantSourceSql(segments)}") + } + + val files = listDataParquetFiles(tablePath) + assert(files.size == 1, s"[$leg] expected exactly one data file, got $files") + assertVariantLayout(tablePath, shredded = true, leg) + + // Physical placement per the shredding spec: objects always materialize typed_value; + // unmatched FIELDS go to the per-field residual, unmatched KEYS to the root residual; + // non-objects (scalars, arrays, JSON null) live entirely in the root residual. + val stats = inspectVariantRows(files.head) + assert(stats.rows == 25, s"[$leg] rows: $stats") + assert(stats.nullVariants == 1, s"[$leg] null variants: $stats") + assert(stats.rootTyped == 20, s"[$leg] object rows with typed_value: $stats") + assert(stats.rootResidual == 9, s"[$leg] root residual rows (ObjB 5 + scalars 3 + json null 1): $stats") + assert(stats.fieldTyped("a") == 10, s"[$leg] typed a: $stats") + assert(stats.fieldResidual("a") == 5, s"[$leg] residual a (type conflict): $stats") + assert(stats.fieldTyped("b") == 15, s"[$leg] typed b: $stats") + + assertVariantSegments(tableName, leg, Seq(("v", segments))) + + // Update rows served from the typed slot and from the residual: the AVRO record type + // reconstructs both through HoodieVariantReconstruction, SPARK natively. + withWriteLayout(Forced("a bigint, b string")) { + spark.sql(s"""update $tableName set v = parse_json('{"a":100,"b":"bu"}'), ts = 1001 where id = 20""") + spark.sql(s"""update $tableName set v = parse_json('{"a":101,"b":"bv"}'), ts = 1001 where id = 5""") + } + checkAnswer(s"select id, cast(v as string), ts from $tableName where id in (5, 12, 20) order by id")( + Seq(5, """{"a":101,"b":"bv"}""", 1001), + Seq(12, """{"a":"s12","b":"b12"}""", 1000), + Seq(20, """{"a":100,"b":"bu"}""", 1001) + ) + assertVariantLayout(tablePath, shredded = true, leg) + } + } + + // ----------------------------------------------------------------------------------------------- + // B. Mixed files inside one table + // ----------------------------------------------------------------------------------------------- + + test("Each commit keeps its own layout; snapshot, time travel, incremental and RO read them all") { + assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE) + + // Read-mode test: layouts are writer-side and every layout is written identically by both + // record types, so the sweep would only re-run the same reads. SPARK pinned. + withVariantTable("mixed-files", "cow", props = Seq(NEW_FILE_GROUP_PER_COMMIT), + recordTypes = Seq(HoodieRecordType.SPARK)) { (tableName, tablePath, leg) => + // Four commits, four layouts, one file each (small.file.limit=0 keeps every commit in its + // own file group). The last commit infers {c, d} from its own ObjB rows. + val instants = seedMixedLayoutTable(tableName, tablePath, Seq( + (Unshredded, Seq((0 until 2, ObjA))), + (Forced("a bigint, b string"), Seq((2 until 4, ObjA))), + (Forced("b string"), Seq((4 until 6, ObjA))), + (Inferred, Seq((6 until 8, ObjB))))) + + assertLayoutsByInstant(baseLayouts(tablePath), leg)( + instants(0) -> None, + instants(1) -> Some(Seq("a", "b")), + instants(2) -> Some(Seq("b")), + instants(3) -> Some(Seq("c", "d"))) + + // Snapshot reads every layout. + assertVariantSegments(tableName, leg, Seq(("v", Seq( + (0 until 6, ObjA), (6 until 8, ObjB))))) + + // Time travel at the second commit sees only the first two layouts. + checkAnswer(s"select id, cast(v as string) from $tableName timestamp as of '${instants(1)}' order by id")( + Seq(0, """{"a":0,"b":"b0"}"""), + Seq(1, """{"a":1,"b":"b1"}"""), + Seq(2, """{"a":2,"b":"b2"}"""), + Seq(3, """{"a":3,"b":"b3"}""") + ) + + // Incremental over the full range returns the latest state of all eight keys, values + // intact (a count alone would pass even if v reconstructed as all-null). + val incRows = incrementalIdAndVariant(tablePath) + assert(incRows.length == 8, s"[$leg] incremental over the full range should see all rows") + incRows.foreach { row => + val id = row.getInt(0) + val expected = if (id < 6) s"""{"a":$id,"b":"b$id"}""" else s"""{"c":$id,"d":true}""" + assert(row.getString(1) == expected, + s"[$leg] incremental id=$id: expected $expected, got ${row.getString(1)}") + } + + // Read-optimized on COW equals the snapshot, values intact. + checkAnswer(s"select id, cast(v as string) from hudi_query('$tableName', 'read_optimized') " + + "where id in (0, 6) order by id")( + Seq(0, """{"a":0,"b":"b0"}"""), + Seq(6, """{"c":6,"d":true}""") + ) + } + } + + test("Small-file bin-pack rewrites the file under the layout of the incoming commit") { + assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE) + + // Default small.file.limit on purpose: each insert bin-packs into the first file group + // and rewrites it (HoodieConcatHandle -> HoodieMergeHelper on the AVRO record type). + // The value round-trip of that merge is owned by TestVariantDataType's small-file test; + // this one exists for the per-instant LAYOUT pin below. + withVariantTable("bin-pack layout flip", "cow") { (tableName, tablePath, leg) => + withWriteLayout(Forced("a bigint, b string")) { + spark.sql(s"""insert into $tableName values (1, parse_json('{"a":1,"b":"b1"}'), 1000)""") + } + val instant1 = latestCompletedInstant(tablePath) + withWriteLayout(Unshredded) { + spark.sql(s"""insert into $tableName values (2, parse_json('{"a":2,"b":"b2"}'), 1000)""") + } + val instant2 = latestCompletedInstant(tablePath) + withWriteLayout(Forced("a bigint")) { + spark.sql(s"""insert into $tableName values (3, parse_json('{"a":3,"b":"b3"}'), 1000)""") + } + val instant3 = latestCompletedInstant(tablePath) + + assertSingleFileGroup(tablePath, leg) + // The rewrite re-derives the layout from the CURRENT write config; the input file's + // layout is never consulted. Older file versions keep their own layouts. + assertLayoutsByInstant(baseLayouts(tablePath), leg)( + instant1 -> Some(Seq("a", "b")), + instant2 -> None, + instant3 -> Some(Seq("a"))) + + checkAnswer(s"select id, cast(v as string), ts from $tableName order by id")( + Seq(1, """{"a":1,"b":"b1"}""", 1000), + Seq(2, """{"a":2,"b":"b2"}""", 1000), + Seq(3, """{"a":3,"b":"b3"}""", 1000) + ) + } + } + + // ----------------------------------------------------------------------------------------------- + // C. MOR compaction over base/log layout splits + // ----------------------------------------------------------------------------------------------- + + test("MOR compaction merges logs of three layouts and re-derives the base layout per service run") { + assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE) + + // INMEMORY sends MOR inserts to log files; compaction runs via the procedure so each run + // can happen under its own layout confs. + withVariantTable("compaction layout split", "mor", props = Seq( + "hoodie.index.type = 'INMEMORY'", "hoodie.compact.inline = 'false'")) { (tableName, tablePath, leg) => + withWriteLayout(Forced("a bigint, b string")) { + spark.sql(s"""insert into $tableName values (1, parse_json('{"a":1,"b":"b1"}'), 1000)""") + } + val instant1 = latestCompletedInstant(tablePath) + withWriteLayout(Unshredded) { + spark.sql(s"""insert into $tableName values (2, parse_json('{"a":2,"b":"b2"}'), 1000), """ + + """(3, parse_json('{"a":3,"b":"b3"}'), 1000), (4, parse_json('{"a":4,"b":"b4"}'), 1000)""") + } + val instant2 = latestCompletedInstant(tablePath) + withWriteLayout(Inferred) { + spark.sql(s"""insert into $tableName values (5, parse_json('{"c":5,"d":true}'), 1000), """ + + """(6, parse_json('{"c":6,"d":true}'), 1000)""") + } + val instant3 = latestCompletedInstant(tablePath) + + assertResult(true)(DataSourceTestUtils.isLogFileOnly(tablePath)) + // On the default table version the data logs are native parquet, each with the layout of + // its own commit. (The SPARK withRecordType leg sets the parquet log block format, the + // AVRO leg avro blocks, but write version >= 10 writes native log FILES either way.) + assertLayoutsByInstant(nativeLogLayouts(tablePath), leg)( + instant1 -> Some(Seq("a", "b")), + instant2 -> None, + instant3 -> Some(Seq("c", "d"))) + + // Merge-on-read snapshot over the three-layout split, before any base file exists. + checkAnswer(s"select id, cast(v as string) from $tableName order by id")( + Seq(1, """{"a":1,"b":"b1"}"""), + Seq(2, """{"a":2,"b":"b2"}"""), + Seq(3, """{"a":3,"b":"b3"}"""), + Seq(4, """{"a":4,"b":"b4"}"""), + Seq(5, """{"c":5,"d":true}"""), + Seq(6, """{"c":6,"d":true}""") + ) + + // Compaction 1 under Inferred: reads all three log layouts, infers the base layout from + // the merged rows. + withWriteLayout(Inferred) { + runCompaction(tableName) + } + assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath)) + assertCompactionCount(tablePath, 1, leg) + val base1 = baseLayouts(tablePath) + assertAllShredded(base1, shredded = true, s"$leg compacted base under Inferred") + // 6 rows: a and b on 4 (66 percent), c and d on 2 (33 percent) - all clear the 10 + // percent inference bar. + base1.foreach(l => assert(l.typedFields.toSet == Set("a", "b", "c", "d"), + s"[$leg] inferred typed_value should carry all four keys: ${l.typedFields}")) + checkAnswer(s"select id, cast(v as string) from $tableName where id in (1, 5) order by id")( + Seq(1, """{"a":1,"b":"b1"}"""), + Seq(5, """{"c":5,"d":true}""") + ) + checkAnswer(s"select id, cast(v as string) from hudi_query('$tableName', 'read_optimized') " + + "where id in (1, 5) order by id")( + Seq(1, """{"a":1,"b":"b1"}"""), + Seq(5, """{"c":5,"d":true}""") + ) + + // Round 2: updates under two further layouts, compaction under Unshredded. The service + // reads a shredded base plus mixed logs and must strip typed_value on the way out. + withWriteLayout(Forced("a bigint")) { + spark.sql(s"""update $tableName set v = parse_json('{"a":22,"b":"b22"}'), ts = 1001 where id = 2""") + } + withWriteLayout(Unshredded) { + spark.sql(s"""update $tableName set v = parse_json('{"a":33,"b":"b33"}'), ts = 1001 where id = 3""") + } + // A delete block (no data column) between the differently-shredded logs: the merged read + // and the following compaction must step over it without a layout to anchor on. + withWriteLayout(Forced("a bigint")) { + spark.sql(s"delete from $tableName where id = 6") + } + // Merge-on-read over shredded base + {a}-shredded log + unshredded log + delete block. + checkAnswer(s"select id, cast(v as string) from $tableName where id in (2, 3, 5) order by id")( + Seq(2, """{"a":22,"b":"b22"}"""), + Seq(3, """{"a":33,"b":"b33"}"""), + Seq(5, """{"c":5,"d":true}""") + ) + withWriteLayout(Unshredded) { + runCompaction(tableName) + } + assertCompactionCount(tablePath, 2, leg) + val compact2Instant = latestCompletedInstant(tablePath) + val base2 = baseLayouts(tablePath).filter(_.instantTime == compact2Instant) + assertAllShredded(base2, shredded = false, s"$leg base of the compaction under Unshredded") + checkAnswer(s"select id, cast(v as string) from $tableName where id in (2, 3) order by id")( + Seq(2, """{"a":22,"b":"b22"}"""), + Seq(3, """{"a":33,"b":"b33"}""") + ) + + // Round 3: compaction under Inferred again, this time reading an UNSHREDDED base plus a + // shredded log. + withWriteLayout(Inferred) { + spark.sql(s"""update $tableName set v = parse_json('{"a":44,"b":"b44"}'), ts = 1001 where id = 4""") + runCompaction(tableName) + } + assertCompactionCount(tablePath, 3, leg) + val compact3Instant = latestCompletedInstant(tablePath) + val base3 = baseLayouts(tablePath).filter(_.instantTime == compact3Instant) + assertAllShredded(base3, shredded = true, s"$leg base of the second compaction under Inferred") + + checkAnswer(s"select id, cast(v as string) from $tableName order by id")( + Seq(1, """{"a":1,"b":"b1"}"""), + Seq(2, """{"a":22,"b":"b22"}"""), + Seq(3, """{"a":33,"b":"b33"}"""), + Seq(4, """{"a":44,"b":"b44"}"""), + Seq(5, """{"c":5,"d":true}""") + ) + // Incremental over the full range sees the latest value of every LIVE key (id 6 deleted), + // values intact - a bare count would pass with v all-null. + val incRows = incrementalIdAndVariant(tablePath) + assert(incRows.map(r => (r.getInt(0), r.getString(1))).toSeq == Seq( + (1, """{"a":1,"b":"b1"}"""), + (2, """{"a":22,"b":"b22"}"""), + (3, """{"a":33,"b":"b33"}"""), + (4, """{"a":44,"b":"b44"}"""), + (5, """{"c":5,"d":true}""") + ), s"[$leg] incremental over the full range, got: ${incRows.mkString(", ")}") + } + } + + test("COMMIT_TIME ordering lets a lower-ts update win across layouts, in the log merge and after compaction") { + assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE) + + // Every other table in the suite carries preCombineField = 'ts' and merges under EVENT_TIME + // ordering. Without it the table config resolves to COMMIT_TIME ordering, where the later commit + // wins whatever its ts: the update below carries a LOWER ts than the row it replaces and must + // still win - under EVENT_TIME the read would keep {"a":1}. Pinned once in the log merge over Review Comment: Done in 84bea41: the leg keeps `preCombineField = 'ts'` and sets `hoodie.record.merge.mode = 'COMMIT_TIME_ORDERING'`, so the ts 500 update wins over the ts 1000 row only because of the mode (with the property removed the leg fails on {"a":1}, the rest of the suite unchanged). The `preCombine` scaffold parameter is gone with its only caller, and the header names the property instead of the dropped field. -- 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]
