wombatu-kun commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3868223755
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HiveHoodieReaderContext.java:
##########
@@ -148,6 +153,45 @@ private ClosableIterator<ArrayWritable>
getFileRecordIterator(StoragePath filePa
fileSchema = dataSchema;
}
+ // Fail fast on shredded variant columns: this reader hands the file to a
plain
+ // parquet-avro read at the requested {metadata, value} projection, so a
file whose variant
+ // group carries typed_value would come back with silent nulls (the typed
rows keep their
+ // payload in typed_value, which the projection drops). Detection is
shape-based on the
+ // footer schema and anchored on the requested column being a variant, so
plain user structs
+ // of the same shape are left alone. toShreddedReadSchema recurses through
structs, array
+ // elements and map values, matching the row writer, which shreds nested
variants too.
+ // Columns not requested stay readable: the flagged columns are checked
against Hive's read
+ // column names, not requiredSchema, which can be wider than the query --
a CUSTOM merge reads
+ // the whole table schema for merging (no merger overrides
isProjectionCompatible), so
+ // `select id` arrives here asking for the variant column too. Hive writes
the full name list
+ // for `select *` and none for count(*). So does a read whose nested
column paths
+ // (hive.io.file.readNestedColumn.paths) all miss the shredded group:
Hive's parquet reader
+ // materializes only the paths it is given, and the mask rewrite below
already handles the
+ // compacted projection such a read comes back in.
+ if (isParquetOrOrc && requiredSchema.getType() == HoodieSchemaType.RECORD)
{
+ HoodieSchema shreddedReadSchema =
VariantSchemaUtils.toShreddedReadSchema(requiredSchema, fileSchema);
+ if (shreddedReadSchema != requiredSchema) {
+ List<String> shreddedPaths = new ArrayList<>();
+ collectShreddedVariantPaths(requiredSchema, shreddedReadSchema, "",
shreddedPaths);
+ Configuration conf = storage.getConf().unwrapAs(Configuration.class);
+ Set<String> requestedColumns =
Arrays.stream(HoodieColumnProjectionUtils.getReadColumnNames(conf))
+ .map(name -> name.trim().toLowerCase(Locale.ROOT))
+ .collect(Collectors.toSet());
+ List<String> offendingColumns =
HoodieColumnProjectionUtils.columnsReadingShreddedPaths(conf, shreddedPaths)
+ .stream()
+ .filter(requestedColumns::contains)
Review Comment:
`setSchemas` overwrites `hive.io.file.readcolumn.names` with
`requiredSchema`'s fields on the per-file conf, so under a CUSTOM merge - where
`FileGroupReaderSchemaHandler` hands back the whole table schema - the variant
is still materialized at `{metadata, value}` and reaches the merger as nulls
while this check no longer fires. Throwing when a shredded file meets a merger
that is not projection-compatible would keep that read from being served
silently.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunClusteringProcedure.scala:
##########
@@ -227,19 +246,38 @@ class RunClusteringProcedure extends BaseProcedure
prunedPartitions.map(partitionPath =>
partitionPath.getPath).toSet.mkString(",")
}
+ /**
+ * Validates the already-normalised (comma-separated, trimmed) order column
list against the
+ * table schema with its metadata fields, which is what the partitioners
sort on at execution
+ * time (the execution strategy adds them to the schema it hands the
partitioners), so a
+ * `_hoodie_*` column is accepted. A dotted path names a nested field: the
partitioners resolve
+ * it (getNestedFieldVal on the record path, Column(name) on the row path)
but the shared
+ * check only walks top-level names, so it is resolved here and its leaf
checked under the path.
+ */
private def validateOrderColumns(orderColumns: String, metaClient:
HoodieTableMetaClient): Unit = {
if (orderColumns == null) {
throw new HoodieClusteringException("Order columns is null")
}
val tableSchemaResolver = new TableSchemaResolver(metaClient)
- val fields = tableSchemaResolver.getTableSchema(false)
- .getFields.asScala.map(_.name().toLowerCase)
- orderColumns.split(",").foreach(col => {
+ val tableSchema = tableSchemaResolver.getTableSchema(true)
+ val fields = tableSchema.getFields.asScala.map(_.name().toLowerCase)
+ val (nestedColumns, topLevelColumns) =
orderColumns.split(",").partition(_.contains("."))
+ topLevelColumns.foreach(col => {
if (!fields.contains(col.toLowerCase)) {
throw new HoodieClusteringException("Order column not exist:" + col)
}
})
+ // The same validation the partitioners apply at execution time (see
+ // SortUtils.validateSortableColumns), surfaced here before the job is
submitted.
+ SortUtils.validateSortableColumns(topLevelColumns, tableSchema)
+ nestedColumns.foreach { col =>
+ val leaf = tableSchema.getNestedField(col)
Review Comment:
`getNestedField` is a parquet-path resolver: it matches each segment exactly
(so `S.level` is reported as not existing while
`RowCustomColumnsSortPartitioner` resolves it) and walks accessor levels (so
`s.tags` is rejected as a MAP but `s.tags.key_value.value` passes). Walking the
segments yourself, case-insensitively and descending only into RECORD, would
fit what this check needs.
##########
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:
With no preCombineField there is no ordering field at all, so
`RecordContext.getOrderingValue` returns the default for both records and
`shouldKeepNewerRecord` keeps the newer one under EVENT_TIME too - the leg
passes identically under either mode. Keeping `preCombineField = 'ts'` and
setting `hoodie.record.merge.mode = 'COMMIT_TIME_ORDERING'` would make the
lower-ts update actually discriminate.
--
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]
