malinjawi commented on code in PR #13502:
URL: https://github.com/apache/gluten/pull/13502#discussion_r4480125924
##########
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:
The splitter fails on sub-millisecond timestamps whether or not the column
is materialized, so the non-materialized case still breaks. Drop `materialize
&&`, or reword the reason?
##########
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:
This compares rows read through Delta, so it never reads the physical `part`
column. Could we check it from the files too?
--
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]
