shangxinli commented on code in PR #18665:
URL: https://github.com/apache/hudi/pull/18665#discussion_r4371338259
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestBlobDataType.scala:
##########
@@ -282,4 +283,346 @@ class TestBlobDataType extends HoodieSparkSqlTestBase {
"Expected at least one .clean instant on the timeline after
compaction")
})
}
+
+ test("Test Query Log Only MOR Table With BLOB INLINE column triggers
compaction (Lance)") {
+ assume(System.getProperty("lance.skip.tests") != "true",
+ "Lance tests disabled via -Dlance.skip.tests=true")
+ // Covers log-only MOR ingest of INLINE blobs on a Lance base format, the
inline
+ // compaction trigger, and the post-compaction inline read shape.
+
+ withRecordType()(withTempDir { tmp =>
+ val tablePath = new File(tmp, "hudi").getCanonicalPath
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | data blob,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | type = 'mor',
+ | preCombineField = 'ts',
+ | hoodie.index.type = 'INMEMORY',
+ | hoodie.compact.inline = 'true',
+ | hoodie.clean.commits.retained = '1',
+ | 'hoodie.table.base.file.format' = 'LANCE'
+ | )
+ """.stripMargin)
+
+ // Verify the LANCE config was actually persisted to hoodie.properties.
+ assertResult(HoodieFileFormat.LANCE)(
+ createMetaClient(spark, tablePath).getTableConfig.getBaseFileFormat)
+
+ spark.sql(s"insert into $tableName values (1,
${inlineBlobLiteral("01")}, 1000)")
+ spark.sql(s"insert into $tableName values (2,
${inlineBlobLiteral("02")}, 1000)")
+ spark.sql(s"insert into $tableName values (3,
${inlineBlobLiteral("03")}, 1000)")
+ assertResult(true)(DataSourceTestUtils.isLogFileOnly(tablePath))
+
+ spark.sql(
+ s"""
+ |merge into $tableName h0
+ |using (
+ | select 1 as id, ${inlineBlobLiteral("11")} as data, 1001L as ts
+ |) s0
+ | on h0.id = s0.id
+ | when matched then update set *
+ |""".stripMargin)
+ assertResult(true)(DataSourceTestUtils.isLogFileOnly(tablePath))
+
+ spark.sql(
+ s"""
+ |merge into $tableName h0
+ |using (
+ | select 4 as id, ${inlineBlobLiteral("04")} as data, 1000L as ts
+ |) s0
+ | on h0.id = s0.id
+ | when not matched then insert *
+ |""".stripMargin)
+
+ assertResult(false)(DataSourceTestUtils.isLogFileOnly(tablePath))
+
+ val bytesById = spark.sql(
+ s"select id, read_blob(data) as bytes from $tableName order by id"
+ ).collect().map(r => r.getInt(0) -> r.getAs[Array[Byte]]("bytes")).toMap
+ assertResult(4)(bytesById.size)
+ assert(bytesById(1).sameElements(Array(0x11.toByte)))
+ assert(bytesById(2).sameElements(Array(0x02.toByte)))
+ assert(bytesById(3).sameElements(Array(0x03.toByte)))
+ assert(bytesById(4).sameElements(Array(0x04.toByte)))
+
+ spark.sql(s"select id, data from $tableName order by
id").collect().foreach { row =>
+ val blob = row.getStruct(1)
+
assertResult("INLINE")(blob.getString(blob.fieldIndex(HoodieSchema.Blob.TYPE)))
+
assert(!blob.isNullAt(blob.fieldIndex(HoodieSchema.Blob.INLINE_DATA_FIELD)))
+ // Lance materializes the `reference` struct as non-null with all-null
leaves for
+ // INLINE rows (vs. a null struct on Parquet). `type` is the canonical
INLINE
+ // discriminator per RFC-100; tolerate either shape and just check the
leaves.
+ val refIdx = blob.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE)
+ if (!blob.isNullAt(refIdx)) {
+ val ref = blob.getStruct(refIdx)
+
assert(ref.isNullAt(ref.fieldIndex(HoodieSchema.Blob.EXTERNAL_REFERENCE_PATH)))
+ }
+ }
+
+ val blobField = spark.table(tableName).schema.find(_.name == "data").get
+ assert(blobField.metadata.contains(HoodieSchema.TYPE_METADATA_FIELD),
+ s"Expected BLOB type metadata on data field after compaction, " +
+ s"got: ${blobField.metadata}")
+ assertResult(HoodieSchemaType.BLOB.name())(
+ blobField.metadata.getString(HoodieSchema.TYPE_METADATA_FIELD))
+
+ // 6th commit drives an auto-clean that retires the now-superseded
log-only slice.
+ spark.sql(
+ s"""
+ |merge into $tableName h0
+ |using (
+ | select 2 as id, ${inlineBlobLiteral("22")} as data, 1002L as ts
+ |) s0
+ | on h0.id = s0.id
+ | when matched then update set *
+ |""".stripMargin)
+ val updatedBytesById = spark.sql(
+ s"select id, read_blob(data) as bytes from $tableName order by id"
+ ).collect().map(r => r.getInt(0) -> r.getAs[Array[Byte]]("bytes")).toMap
+ assert(updatedBytesById(2).sameElements(Array(0x22.toByte)))
+
+ val metaClient = createMetaClient(spark, tablePath)
+ metaClient.reloadActiveTimeline()
+ assert(metaClient.getActiveTimeline.getCleanerTimeline.countInstants() >
0,
+ "Expected at least one .clean instant on the timeline after
compaction")
+ })
+ }
+
+ test("Test Query Log Only MOR Table With BLOB OUT_OF_LINE column triggers
compaction (Lance)") {
+ assume(System.getProperty("lance.skip.tests") != "true",
+ "Lance tests disabled via -Dlance.skip.tests=true")
+ // Lance writer has no BLOB handling today (RFC-100 Phase 2). Expected to
fail
Review Comment:
The comment says this is expected to fail until Lance gets BLOB support, but
the test body below runs a full set of unconditional assertions on the written
and compacted data with no exception handling, so either the comment is stale
or this test is currently red.
##########
hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/command/InsertIntoHoodieTableCommand.scala:
##########
@@ -168,7 +168,13 @@ object InsertIntoHoodieTableCommand extends Logging with
ProvidesHoodieConfig wi
// since such columns wouldn't be otherwise specified w/in the query
itself and therefore couldn't be matched
// positionally for example
val expectedQueryColumns =
catalogTable.tableSchemaWithoutMetaFields.filterNot(f =>
staticPartitionValues.contains(f.name))
- val coercedQueryOutput =
coerceQueryOutputColumns(StructType(expectedQueryColumns), cleanedQuery,
catalogTable, conf)
+ // Canonicalize partial BLOB struct expressions before TableOutputResolver
runs,
+ // so OUT_OF_LINE-style `{type, reference}` literals (which Spark's
positional
+ // struct check would otherwise reject for not aligning with the canonical
+ // `{type, data, reference}` layout) are accepted on the SQL ingest path.
+ val canonicalizedQuery =
InsertBlobCanonicalizer.padPartialBlobsAgainstTarget(
Review Comment:
padPartialBlobsAgainstTarget only gets wired into the INSERT INTO path here.
MergeIntoHoodieTableCommand resolves its insert output with its own
castIfNeeded logic and never calls this, so a partial out-of-line blob struct
written through a merge insert clause skips canonicalization entirely.
--
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]
