wombatu-kun commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3558242255
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestLanceDataSource.scala:
##########
@@ -1443,6 +1481,114 @@ class TestLanceDataSource extends
HoodieSparkClientTestBase {
}
}
+ /**
+ * Clustering must preserve INLINE blob bytes under the DESCRIPTOR default.
Clustering rewrites
+ * ALL rows through
MultipleSparkJobExecutionStrategy.readRecordsForGroupAsRow, which reads base
+ * files through the same internal write-side reader context as compaction
+ * (SparkReaderContextFactory -> SparkFileFormatInternalRowReaderContext ->
SparkLanceReaderBase).
+ * Before SparkReaderContextFactory pinned {@code
hoodie.read.blob.inline.mode=CONTENT} for
+ * internal reads, the first clustering of a Lance table with INLINE blobs
read null {@code data}
+ * (the DESCRIPTOR default) and rewrote every row's blob with null bytes,
silently losing all
+ * blob content (#19232). This test bulk-inserts INLINE blobs, triggers
inline clustering,
+ * asserts a replacecommit actually completed, and verifies every row's
bytes survived.
+ */
+ @Test
+ def testBlobInlineClusteringRoundTrip(): Unit = {
+ val tableType = HoodieTableType.COPY_ON_WRITE
+ val tableName = "test_lance_blob_inline_cluster_cow"
+ val tablePath = s"$basePath/$tableName"
+
+ val payloadLen = 1024
+ val numRows = 5
+ val expectedPayloads: Seq[Array[Byte]] = (0 until numRows).map { i =>
+ (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray
+ }
+ val sparkSess = spark
+ import sparkSess.implicits._
+
+ val canonicalSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("payload", BlobType().asInstanceOf[StructType], nullable =
true,
+ BlobTestHelpers.blobMetadata)
+ ))
+ def asInlineDf(idToBytes: Seq[(Int, Array[Byte])]): DataFrame = {
+ val rawDf = idToBytes.toDF("id", "bytes")
+ .select($"id", BlobTestHelpers.inlineBlobStructCol("payload",
$"bytes"))
+ spark.createDataFrame(rawDf.rdd, canonicalSchema)
+ }
+
+ // First commit: bulk_insert ids 0..4 with the initial pattern into a base
file.
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(expectedPayloads.zipWithIndex.map { case (b, i) => (i, b) }),
+ saveMode = SaveMode.Overwrite,
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id"))
+
+ assertLanceBlobEncoding(tablePath)
+
+ // Second commit: a small bulk_insert that trips inline clustering
(max.commits=1). Clustering
+ // rewrites the existing base file's rows through
readRecordsForGroupAsRow, which must read the
+ // INLINE bytes as CONTENT, not the DESCRIPTOR default, or every rewritten
row loses its bytes.
+ val extraPayloads = (numRows until numRows + 2).map { i =>
+ (i, (0 until payloadLen).map(j => ((i + j) % 256).toByte).toArray)
+ }
+ writeDataframe(tableType, tableName, tablePath,
+ asInlineDf(extraPayloads),
+ operation = Some("bulk_insert"),
+ extraOptions = Map(PRECOMBINE_FIELD.key() -> "id",
+ "hoodie.clustering.inline" -> "true",
+ "hoodie.clustering.inline.max.commits" -> "1"))
+
+ // Assert a clustering (replacecommit) instant actually completed, so the
test cannot silently
+ // pass without a rewrite.
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .setBasePath(tablePath)
+ .build()
+ assertTrue(metaClient.getActiveTimeline.getLastClusteringInstant.isPresent,
Review Comment:
`getLastClusteringInstant` does not filter by instant state: it reads
`getCommitsTimeline()`, and `BaseHoodieTimeline.getTimelineOfActions` filters
`getInstantsAsStream()` by action only, so a REQUESTED or INFLIGHT
replacecommit satisfies `isPresent`. Nothing here checks completion;
`getLastClusteringInstant.get.isCompleted` would.
The second half of the claim needs more. A completed clustering instant
still does not prove ids 0..4 went through the rewrite: the first bulk-insert's
base file qualifies for the plan only via the
`hoodie.clustering.plan.strategy.small.file.limit` and single-group-clustering
defaults, and if either drifts the byte assertions below read the untouched
original bytes and pass. That is the masking mode of #19232, the same one
`testBlobInlineCompactionRoundTrip` now pins with its single-file-group assert.
Suggest capturing `fsView.getLatestBaseFiles("")` after the first commit and
asserting the post-clustering base files are disjoint from it.
--
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]
