felipepessoto commented on code in PR #13136:
URL: https://github.com/apache/gluten/pull/13136#discussion_r4405699005
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala:
##########
@@ -97,6 +150,219 @@ class GlutenFileMetadataStructSuite extends
FileMetadataStructSuite with GlutenS
assert(executedPlan.exists(plan => plan.getClass == tag.runtimeClass))
}
+ private def checkMetadataAnswer(
+ df: DataFrame,
+ expected: Seq[Row],
+ hasRowIndex: Boolean = false): Unit = {
+ checkAnswer(df, expected)
+ val settings = BackendsApiManager.getSettings
+ if (
+ settings.supportNativeMetadataColumns() &&
+ (!hasRowIndex || settings.supportNativeRowIndexColumn())
+ ) {
+ checkOperatorMatch[FileSourceScanExecTransformer](df)
+ } else {
+ checkOperatorMatch[FileSourceScanExec](df)
+ }
+ }
+
+ metadataColumns.zipWithIndex.foreach {
+ case (metadataColumn, index) =>
+ metadataColumnsNativeTest(
+ s"incompatible same-name data field for $metadataColumn",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ val expected =
Seq.fill(collisionRows0.size)(Row(f0(metadataColumn))) ++
+ Seq.fill(collisionRows1.size)(Row(f1(metadataColumn)))
+ checkMetadataAnswer(df.select(metadataColumn), expected)
+
checkMetadataAnswer(df.select(Column(metadataColumn).as("aliased_metadata")),
expected)
+
+ val combined = df.select(
+ Column(metadataColumn.stripPrefix("_metadata.")).as("user_value"),
+ Column(metadataColumn).as("metadata_value"))
+ checkAnswer(
+ combined,
+ collisionRows0.map(row => Row(row.get(index), f0(metadataColumn)))
++
+ collisionRows1.map(row => Row(row.get(index),
f1(metadataColumn))))
+ checkOperatorMatch[FileSourceScanExec](combined)
+ }
+ }
+
+ metadataColumnsNativeTest(
+ "metadata-only projection of empty Parquet files",
+ schemaWithMetadataCollisions,
+ Seq.empty,
+ Seq.empty) {
+ (df, _, _) =>
+ assert(df.inputFiles.length == 2)
+ df.inputFiles.foreach {
+ file =>
+ val input =
+ HadoopInputFile.fromPath(new Path(new URI(file)),
spark.sessionState.newHadoopConf())
+ Utils.tryWithResource(ParquetFileReader.open(input)) {
+ reader =>
+ val footer = reader.getFooter
+ assert(footer.getBlocks.asScala.map(_.getRowCount).sum == 0L)
+ assert(
+ footer.getFileMetaData.getSchema
+ .asGroupType()
+ .getType("file_name")
+ .asPrimitiveType()
+ .getPrimitiveTypeName == PrimitiveTypeName.INT64)
+ }
+ }
+ checkMetadataAnswer(df.select(METADATA_FILE_NAME), Seq.empty)
+ }
+
+ metadataColumnsNativeTest(
+ "metadata constants with a regular sibling",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ val selected = df.select((metadataColumns.map(Column(_)) :+
Column("id")): _*)
+ val expected = Seq((collisionRows0, f0), (collisionRows1, f1)).flatMap {
+ case (rows, metadata) =>
+ rows.map(row => Row.fromSeq(metadataColumns.map(metadata) :+
row.getLong(6)))
Review Comment:
Updated in 27e9bf905: `idColumnIndex` is derived from
`schemaWithMetadataCollisions.fieldIndex("id")`, and all repeated `getLong(6)`
lookups use it. The fixture data and ordering are unchanged. The full metadata
suite passed all 95 tests.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
