felipepessoto commented on code in PR #13502:
URL: https://github.com/apache/gluten/pull/13502#discussion_r4485191447
##########
backends-velox/src-delta4/test/scala/org/apache/gluten/execution/VeloxDelta4WriteSuite.scala:
##########
@@ -121,46 +123,135 @@ class VeloxDelta4WriteSuite extends DeltaSQLCommandTest {
}
}
- test("a partitioned table falls back, since Delta writes the partition
columns") {
- withTempDir {
- dir =>
- val path = dir.getCanonicalPath
- createTable(path, partitioned = true)
- writeRows(path)
+ for (feature <- Seq(false, true); property <- Seq(None, Some(false),
Some(true))) {
+ test(s"native partition materialization: feature=$feature,
property=$property") {
+ withTempDir {
+ dir =>
+ val path = dir.getCanonicalPath
+ val properties = property.map(v =>
"delta.writePartitionColumnsToParquet" -> v.toString)
+ .toSeq ++ (if (feature) {
+ Seq("delta.feature.materializePartitionColumns" ->
"supported")
+ } else {
+ Seq.empty
+ })
+ createTable(path, partitioned = true, properties: _*)
+ writeRows(path)
- assertWrittenBy(path, velox = false, Seq("id", "value", "part"))
- footers(path).foreach {
- case (file, _) => assert(parse(file.stats) \ "minValues" \ "part" ==
JNothing)
- }
- assertRows(path)
+ val snapshot = DeltaLog.forTable(spark, path).update()
+
assert(snapshot.protocol.isFeatureSupported(MaterializePartitionColumnsTableFeature)
==
+ feature)
+
assert(snapshot.metadata.configuration.get("delta.writePartitionColumnsToParquet")
==
+ property.map(_.toString))
+ val materialize = feature || property.getOrElse(true)
+ val columns = if (materialize) Seq("id", "value", "part") else
Seq("id", "value")
+ assertWrittenBy(path, velox = true, columns)
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ footers(path).foreach {
+ case (file, _) =>
+ val stats = parse(file.stats)
+ for (field <- Seq("minValues", "maxValues", "nullCount")) {
+ assert(stats \ field \ "part" == JNothing, file.stats)
+ }
+ if (materialize) {
+ val filePath = DeltaFileOperations.absolutePath(path,
file.path).toString
+ val rows = spark.read.option("recursiveFileLookup", "true")
+ .parquet(filePath).select("part").collect()
+ assert(rows.nonEmpty)
+ assert(rows.forall(_.getInt(0).toString ==
file.partitionValues("part")))
+ }
+ }
+ }
+ assertRows(path)
+ }
}
}
- test("a partitioned table without partition columns in its files is written
natively") {
+ test("enabling materialization preserves reads of older files without
partition columns") {
withTempDir {
dir =>
val path = dir.getCanonicalPath
createTable(path, partitioned = true,
"delta.writePartitionColumnsToParquet" -> "false")
writeRows(path)
+ val oldFiles = footers(path).map(_._1.path).toSet
- assertWrittenBy(path, velox = true, Seq("id", "value"))
- assertRows(path)
+ sql(s"ALTER TABLE delta.`$path` SET TBLPROPERTIES " +
+ "('delta.feature.materializePartitionColumns' = 'supported')")
+ writeRows(path)
+
+ val files = footers(path)
+ assert(files.exists { case (file, _) => !oldFiles.contains(file.path)
})
+ files.foreach {
+ case (file, footer) =>
+ assert(writtenByVelox(footer), file.path)
+ val columns = if (oldFiles.contains(file.path)) {
+ Seq("id", "value")
+ } else {
+ Seq("id", "value", "part")
+ }
+ assert(fieldNames(footer) == columns, file.path)
+ }
+ val expected = (0 until 10).flatMap(i => Seq.fill(2)(Row(i.toLong,
i.toString, i % 2)))
+ for (gluten <- Seq("false", "true")) {
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> gluten) {
+
assert(spark.read.format("delta").load(path).orderBy("id").collect().toSeq ==
expected)
+ }
+ }
}
}
- test("materializePartitionColumns falls back") {
+ test("native updates and deletes materialize partition columns like vanilla
Delta") {
withTempDir {
- dir =>
- val path = dir.getCanonicalPath
- createTable(
- path,
- partitioned = true,
- "delta.writePartitionColumnsToParquet" -> "false",
- "delta.enableMaterializePartitionColumnsFeature" -> "true")
- writeRows(path)
-
- assertWrittenBy(path, velox = false, Seq("id", "value", "part"))
- assertRows(path)
+ nativeDir =>
+ withTempDir {
+ vanillaDir =>
+ val nativePath = nativeDir.getCanonicalPath
+ val vanillaPath = vanillaDir.getCanonicalPath
+ val paths = Seq(nativePath -> "true", vanillaPath -> "false")
+ paths.foreach {
+ case (path, native) =>
+ withSQLConf(VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key ->
native) {
+ createTable(
+ path,
+ partitioned = true,
+ "delta.feature.materializePartitionColumns" -> "supported",
+ "delta.writePartitionColumnsToParquet" -> "false",
+ "delta.enableDeletionVectors" -> "false"
+ )
+ writeRows(path)
+ }
+ }
+ val commands = Seq(
+ "UPDATE delta.`%s` SET part = part + 2, value = 'updated' WHERE
id = 0",
+ "DELETE FROM delta.`%s` WHERE id = 1")
+ commands.foreach {
+ command =>
+ val oldFiles = footers(nativePath).map(_._1.path).toSet
+ paths.foreach {
+ case (path, native) =>
+ withSQLConf(VeloxDeltaConfig.ENABLE_NATIVE_WRITE.key ->
native) {
+ sql(command.format(path)).collect()
+ }
+ }
+ val newFiles = footers(nativePath).filterNot {
+ case (file, _) => oldFiles.contains(file.path)
+ }
+ assert(newFiles.nonEmpty)
+ newFiles.foreach {
+ case (file, footer) =>
+ assert(writtenByVelox(footer), file.path)
+ assert(fieldNames(footer) == Seq("id", "value", "part"),
file.path)
+ val stats = parse(file.stats)
+ for (field <- Seq("minValues", "maxValues", "nullCount")) {
+ assert(stats \ field \ "part" == JNothing, file.stats)
+ }
+ }
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
Review Comment:
Added direct Parquet reads in e57998b8c, with `recursiveFileLookup=true` and
Gluten disabled for the independent read. Each rewritten file's stored `part`
values are compared with `AddFile.partitionValues`, and all active physical
rows are checked against explicit expected rows after UPDATE and DELETE, for
both native and vanilla writes. This now checks the changed partition value
itself, not just values reconstructed by Delta. The suite passes on Spark 4.1.
##########
backends-velox/src-delta4/main/scala/org/apache/spark/sql/delta/GlutenDeltaNativeWriteGate.scala:
##########
@@ -122,6 +120,18 @@ object GlutenDeltaNativeWriteGate {
private val cryptoFactoryClass =
EncryptionPropertiesFactory.CRYPTO_FACTORY_CLASS_PROPERTY_NAME
private val parquetV1 = WriterVersion.PARQUET_1_0
+ private def unsupportedPartitionColumns(
+ partitionColumns: Seq[Attribute],
+ materialize: Boolean): Option[String] =
+ partitionColumns.collectFirst {
+ case column if column.dataType == FloatType || column.dataType ==
DoubleType =>
+ s"${column.dataType.sql} partition columns are not supported by the
native Delta writer"
+ // PartitionIdGenerator cannot represent sub-millisecond timestamps.
Keep the existing
+ // materialized-timestamp fallback until splitting preserves their full
precision.
+ case column if materialize && column.dataType == TimestampType =>
Review Comment:
Removed the materialization condition in e57998b8c. TIMESTAMP partition
columns now fall back in both modes; timestamp data columns keep their existing
encoding checks. The docs call out that this is conservative even for
millisecond-only partition values. Added a physical-file regression with four
distinct microsecond partition values for both `writePartitionColumnsToParquet`
settings; both pass on Spark 4.1.
--
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]
