voonhous commented on code in PR #19236:
URL: https://github.com/apache/hudi/pull/19236#discussion_r3559543949
##########
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:
Both fixed: now asserts `getLastClusteringInstant.get.isCompleted`
(isPresent alone accepts a REQUESTED/INFLIGHT replacecommit), and snapshots the
base files after the first commit then asserts the post-clustering set is
disjoint, so a skipped rewrite fails instead of the byte checks reading the
untouched originals.
--
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]
