This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new c7713ce0ae [GLUTEN-10215][VL] Enforce file limits in native
partitioned Delta writes (#13160)
c7713ce0ae is described below
commit c7713ce0ae7704091f1945ccb1039952e2518688
Author: Mohammad Linjawi <[email protected]>
AuthorDate: Tue Oct 6 14:20:53 2026 +0300
[GLUTEN-10215][VL] Enforce file limits in native partitioned Delta writes
(#13160)
---
.../delta/files/GlutenDeltaFileFormatWriter.scala | 111 +++++++++++-----
.../delta/files/GlutenDeltaFileFormatWriter.scala | 106 +++++++++++----
gluten-ut/spark35/pom.xml | 11 ++
.../GlutenDeltaNativePartitionWriteSuite.scala | 144 +++++++++++++++++++++
gluten-ut/spark40/pom.xml | 30 +++++
gluten-ut/spark41/pom.xml | 30 +++++
6 files changed, 374 insertions(+), 58 deletions(-)
diff --git
a/backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
b/backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
index e9ea3e4753..2bcd7de021 100644
---
a/backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
+++
b/backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
@@ -18,6 +18,7 @@ package org.apache.spark.sql.delta.files
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.columnarbatch.VeloxColumnarBatches
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.execution._
import org.apache.gluten.execution.datasource.GlutenFormatFactory
@@ -45,6 +46,7 @@ import
org.apache.spark.sql.execution.datasources.FileFormatWriter._
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.vectorized.ColumnarBatch
import org.apache.spark.util.{SerializableConfiguration, Utils}
import org.apache.hadoop.conf.Configuration
@@ -590,43 +592,92 @@ object GlutenDeltaFileFormatWriter extends LoggingShims {
record match {
case carrierRow: BatchCarrierRow =>
carrierRow match {
- case placeholderRow: PlaceholderRow =>
+ case _: PlaceholderRow =>
// Do nothing.
case terminalRow: TerminalRow =>
- val numRows = terminalRow.batch().numRows()
- if (numRows > 0) {
- val blockStripes = GlutenFormatFactory.rowSplitter
- .splitBlockByPartitionAndBucket(
- terminalRow.batch(),
- partitionColIndice,
- isBucketed)
- val iter = blockStripes.iterator()
- while (iter.hasNext) {
- val blockStripe = iter.next()
- val headingRow = blockStripe.getHeadingRow
- beforeWrite(headingRow)
- val currentColumnBatch = blockStripe.getColumnarBatch
- val numRowsOfCurrentColumnarBatch =
currentColumnBatch.numRows()
- assert(numRowsOfCurrentColumnarBatch > 0)
- val currentTerminalRow =
terminalRow.withNewBatch(currentColumnBatch)
- currentWriter.write(currentTerminalRow)
- statsTrackers.foreach {
- tracker =>
- tracker.newRow(currentWriter.path, currentTerminalRow)
- for (_ <- 0 until numRowsOfCurrentColumnarBatch - 1) {
- tracker.newRow(currentWriter.path, new
PlaceholderRow())
- }
- }
- currentColumnBatch.close()
- }
- blockStripes.release()
- recordsInFile += numRows
- }
+ writePartitionedBatch(terminalRow)
}
case _ =>
beforeWrite(record)
writeRecord(record)
}
}
+
+ private def writeCurrentBatch(terminalRow: TerminalRow, rowCount: Int):
Unit = {
+ assert(rowCount > 0)
+ currentWriter.write(terminalRow)
+ statsTrackers.foreach(_.newRow(currentWriter.path, terminalRow))
+ recordsInFile += rowCount
+ }
+
+ private def writeCurrentBatchWithMaxRecords(
+ terminalRow: TerminalRow,
+ columnBatch: ColumnarBatch): Unit = {
+ val numRows = columnBatch.numRows()
+ var offset = 0
+ while (offset < numRows) {
+ val rowsRemaining = numRows - offset
+ val rowsToWrite = if (description.maxRecordsPerFile > 0) {
+ if (recordsInFile >= description.maxRecordsPerFile) {
+ renewCurrentWriterIfTooManyRecords(currentPartitionValues,
currentBucketId)
+ }
+ math.min(rowsRemaining.toLong, description.maxRecordsPerFile -
recordsInFile).toInt
+ } else {
+ rowsRemaining
+ }
+
+ assert(rowsToWrite > 0)
+ val batchToWrite =
+ if (offset == 0 && rowsToWrite == numRows) {
+ columnBatch
+ } else {
+ VeloxColumnarBatches.slice(columnBatch, offset, rowsToWrite)
+ }
+ Utils.tryWithSafeFinally {
+ writeCurrentBatch(terminalRow.withNewBatch(batchToWrite),
rowsToWrite)
+ } {
+ if (batchToWrite ne columnBatch) {
+ batchToWrite.close()
+ }
+ }
+ offset += rowsToWrite
+ }
+ }
+
+ private def writePartitionStripe(terminalRow: TerminalRow, blockStripe:
BlockStripe): Unit = {
+ val currentColumnBatch = blockStripe.getColumnarBatch
+ Utils.tryWithSafeFinally {
+ beforeWrite(blockStripe.getHeadingRow)
+ assert(currentColumnBatch.numRows() > 0)
+ writeCurrentBatchWithMaxRecords(terminalRow, currentColumnBatch)
+ } {
+ currentColumnBatch.close()
+ }
+ }
+
+ private def writePartitionedBatch(terminalRow: TerminalRow): Unit = {
+ val numRows = terminalRow.batch().numRows()
+ if (numRows > 0) {
+ val blockStripes = GlutenFormatFactory.rowSplitter
+ .splitBlockByPartitionAndBucket(
+ terminalRow.batch(),
+ partitionColIndice,
+ isBucketed)
+ val iter = blockStripes.iterator()
+ Utils.tryWithSafeFinally {
+ while (iter.hasNext) {
+ writePartitionStripe(terminalRow, iter.next())
+ }
+ } {
+ // Velox creates all stripe batches eagerly, and release() does not
close them.
+ // On failure, the current stripe has already been closed by
writePartitionStripe.
+ while (iter.hasNext) {
+ val remainingBatch = iter.next().getColumnarBatch
+ Utils.tryLogNonFatalError(remainingBatch.close())
+ }
+ blockStripes.release()
+ }
+ }
+ }
}
}
diff --git
a/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
b/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
index 73b9a8fdba..4d8c660687 100644
---
a/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
+++
b/backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/files/GlutenDeltaFileFormatWriter.scala
@@ -18,6 +18,7 @@ package org.apache.spark.sql.delta.files
import org.apache.gluten.backendsapi.BackendsApiManager
import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.columnarbatch.VeloxColumnarBatches
import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.execution._
import org.apache.gluten.execution.datasource.GlutenFormatFactory
@@ -46,6 +47,7 @@ import
org.apache.spark.sql.execution.datasources.FileFormatWriter._
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.vectorized.ColumnarBatch
import org.apache.spark.util.{SerializableConfiguration, Utils}
import org.apache.hadoop.conf.Configuration
@@ -583,42 +585,90 @@ object GlutenDeltaFileFormatWriter extends Logging {
record match {
case carrierRow: BatchCarrierRow =>
carrierRow match {
- case placeholderRow: PlaceholderRow =>
+ case _: PlaceholderRow =>
// Do nothing.
case terminalRow: TerminalRow =>
- val numRows = terminalRow.batch().numRows()
- if (numRows > 0) {
- val blockStripes = GlutenFormatFactory.rowSplitter
- .splitBlockByPartitionAndBucket(terminalRow.batch(),
partitionColIndice,
- isBucketed)
- val iter = blockStripes.iterator()
- while (iter.hasNext) {
- val blockStripe = iter.next()
- val headingRow = blockStripe.getHeadingRow
- beforeWrite(headingRow)
- val currentColumnBatch = blockStripe.getColumnarBatch
- val numRowsOfCurrentColumnarBatch =
currentColumnBatch.numRows()
- assert(numRowsOfCurrentColumnarBatch > 0)
- val currentTerminalRow =
terminalRow.withNewBatch(currentColumnBatch)
- currentWriter.write(currentTerminalRow)
- statsTrackers.foreach {
- tracker =>
- tracker.newRow(currentWriter.path, currentTerminalRow)
- for (_ <- 0 until numRowsOfCurrentColumnarBatch - 1) {
- tracker.newRow(currentWriter.path, new
PlaceholderRow())
- }
- }
- currentColumnBatch.close()
- }
- blockStripes.release()
- recordsInFile += numRows
- }
+ writePartitionedBatch(terminalRow)
}
case _ =>
beforeWrite(record)
writeRecord(record)
}
}
+
+ private def writeCurrentBatch(terminalRow: TerminalRow, rowCount: Int):
Unit = {
+ assert(rowCount > 0)
+ currentWriter.write(terminalRow)
+ statsTrackers.foreach(_.newRow(currentWriter.path, terminalRow))
+ recordsInFile += rowCount
+ }
+
+ private def writeCurrentBatchWithMaxRecords(
+ terminalRow: TerminalRow,
+ columnBatch: ColumnarBatch): Unit = {
+ val numRows = columnBatch.numRows()
+ var offset = 0
+ while (offset < numRows) {
+ val rowsRemaining = numRows - offset
+ val rowsToWrite = if (description.maxRecordsPerFile > 0) {
+ if (recordsInFile >= description.maxRecordsPerFile) {
+ renewCurrentWriterIfTooManyRecords(currentPartitionValues,
currentBucketId)
+ }
+ math.min(rowsRemaining.toLong, description.maxRecordsPerFile -
recordsInFile).toInt
+ } else {
+ rowsRemaining
+ }
+
+ assert(rowsToWrite > 0)
+ val batchToWrite =
+ if (offset == 0 && rowsToWrite == numRows) {
+ columnBatch
+ } else {
+ VeloxColumnarBatches.slice(columnBatch, offset, rowsToWrite)
+ }
+ Utils.tryWithSafeFinally {
+ writeCurrentBatch(terminalRow.withNewBatch(batchToWrite),
rowsToWrite)
+ } {
+ if (batchToWrite ne columnBatch) {
+ batchToWrite.close()
+ }
+ }
+ offset += rowsToWrite
+ }
+ }
+
+ private def writePartitionStripe(terminalRow: TerminalRow, blockStripe:
BlockStripe): Unit = {
+ val currentColumnBatch = blockStripe.getColumnarBatch
+ Utils.tryWithSafeFinally {
+ beforeWrite(blockStripe.getHeadingRow)
+ assert(currentColumnBatch.numRows() > 0)
+ writeCurrentBatchWithMaxRecords(terminalRow, currentColumnBatch)
+ } {
+ currentColumnBatch.close()
+ }
+ }
+
+ private def writePartitionedBatch(terminalRow: TerminalRow): Unit = {
+ val numRows = terminalRow.batch().numRows()
+ if (numRows > 0) {
+ val blockStripes = GlutenFormatFactory.rowSplitter
+ .splitBlockByPartitionAndBucket(terminalRow.batch(),
partitionColIndice, isBucketed)
+ val iter = blockStripes.iterator()
+ Utils.tryWithSafeFinally {
+ while (iter.hasNext) {
+ writePartitionStripe(terminalRow, iter.next())
+ }
+ } {
+ // Velox creates all stripe batches eagerly, and release() does not
close them.
+ // On failure, the current stripe has already been closed by
writePartitionStripe.
+ while (iter.hasNext) {
+ val remainingBatch = iter.next().getColumnarBatch
+ Utils.tryLogNonFatalError(remainingBatch.close())
+ }
+ blockStripes.release()
+ }
+ }
+ }
}
}
// spotless:on
diff --git a/gluten-ut/spark35/pom.xml b/gluten-ut/spark35/pom.xml
index 97fac48c18..ccd46169d7 100644
--- a/gluten-ut/spark35/pom.xml
+++ b/gluten-ut/spark35/pom.xml
@@ -296,5 +296,16 @@
</build>
</profile>
+ <profile>
+ <id>delta</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-delta</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
</profiles>
</project>
diff --git
a/gluten-ut/spark35/src-delta/test/scala/org/apache/spark/sql/execution/datasources/GlutenDeltaNativePartitionWriteSuite.scala
b/gluten-ut/spark35/src-delta/test/scala/org/apache/spark/sql/execution/datasources/GlutenDeltaNativePartitionWriteSuite.scala
new file mode 100644
index 0000000000..858d0e1444
--- /dev/null
+++
b/gluten-ut/spark35/src-delta/test/scala/org/apache/spark/sql/execution/datasources/GlutenDeltaNativePartitionWriteSuite.scala
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.utils.BackendTestUtils
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{GlutenQueryTest, Row}
+import org.apache.spark.sql.delta.DeltaLog
+import org.apache.spark.sql.delta.util.DeltaFileOperations
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
+import org.apache.spark.sql.execution.command.ExecutedCommandExec
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, LongType, StructType}
+import org.apache.spark.sql.util.QueryExecutionListener
+
+import org.json4s.{JInt, JNothing}
+import org.json4s.jackson.JsonMethods.parse
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.collection.JavaConverters._
+
+class GlutenDeltaNativePartitionWriteSuite extends GlutenQueryTest with
SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.plugins", "org.apache.gluten.GlutenPlugin")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.sql.shuffle.partitions", "1")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "1024MB")
+ .set("spark.ui.enabled", "false")
+ .set(GlutenConfig.GLUTEN_UI_ENABLED.key, "false")
+ .set("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
+ .set("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
+ .set("spark.gluten.sql.columnar.backend.velox.delta.enableNativeWrite",
"true")
+ .set("spark.databricks.delta.snapshotPartitions", "1")
+ .set("spark.databricks.delta.optimizeWrite.enabled", "false")
+ .set("spark.sql.ansi.enabled", "false")
+ .set(GlutenConfig.GLUTEN_ANSI_FALLBACK_ENABLED.key, "false")
+ .set("spark.sql.adaptive.enabled", "false")
+ .set("spark.sql.maxConcurrentOutputFileWriters", "0")
+ }
+
+ // Small batches cross partition boundaries; a large batch requires
splitting a single stripe.
+ for ((batchSize, limit, collectStats) <- Seq((4, 4, false), (16, 4, true),
(4, 0, true))) {
+ test(s"native Delta partitioned files: batch=$batchSize, limit=$limit,
stats=$collectStats") {
+ assume(BackendTestUtils.isVeloxBackendLoaded())
+ val schema = new StructType().add("id", LongType).add("part",
IntegerType)
+ val expected = Seq((null, 3), (Int.box(0), 4), (Int.box(1), 9)).flatMap {
+ case (part, count) =>
+ (0 until count).map(i => Row(if (i % 3 == 0) null else Long.box(i /
2), part))
+ }
+ withSQLConf(
+ GlutenConfig.COLUMNAR_MAX_BATCH_SIZE.key -> batchSize.toString,
+ "spark.databricks.delta.stats.collect" -> collectStats.toString,
+ "spark.sql.jsonGenerator.ignoreNullFields" -> "true"
+ ) {
+ withTempDir {
+ dir =>
+ val path = dir.getCanonicalPath
+ val plans = new ConcurrentLinkedQueue[SparkPlan]()
+ val listener = new QueryExecutionListener {
+ override def onSuccess(name: String, qe: QueryExecution,
duration: Long): Unit =
+ plans.add(qe.executedPlan)
+ override def onFailure(name: String, qe: QueryExecution, error:
Exception): Unit = {}
+ }
+ spark.listenerManager.register(listener)
+ try {
+ spark.createDataFrame(spark.sparkContext.parallelize(expected,
1), schema)
+ .write.format("delta").partitionBy("part")
+ .option("maxRecordsPerFile", limit.toString).save(path)
+ spark.sparkContext.listenerBus.waitUntilEmpty(10000)
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+ // This source also compiles for other backends, which do not
provide these classes.
+ assert(
+ plans.asScala.exists(_.exists {
+ case ExecutedCommandExec(command) =>
+ Set("GlutenDeltaLeafRunnableCommand",
"GlutenDeltaRunnableCommand")
+ .contains(command.getClass.getSimpleName)
+ case plan => plan.getClass.getSimpleName ==
"GlutenDeltaLeafV2CommandExec"
+ }),
+ s"Expected a native Delta
write:\n${plans.asScala.map(_.treeString).mkString}"
+ )
+
+ // Read the physical files with Spark to independently check
layout and per-file stats.
+ withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ val files = DeltaLog.forTable(spark,
path).update().allFiles.collect()
+ val fileSizes = files.toSeq.map {
+ file =>
+ val filePath = DeltaFileOperations.absolutePath(path,
file.path).toString
+ val rows =
spark.read.parquet(filePath).select("id").collect().toSeq
+ if (collectStats) {
+ val stats = parse(file.stats)
+ val values =
rows.filterNot(_.isNullAt(0)).map(_.getLong(0))
+ assert((stats \ "numRecords") == JInt(rows.size),
file.stats)
+ assert(
+ (stats \ "nullCount" \ "id") == JInt(rows.size -
values.size),
+ file.stats)
+ assert(
+ (stats \ "minValues" \ "id") ==
+ (if (values.isEmpty) JNothing else JInt(values.min)),
+ file.stats)
+ assert(
+ (stats \ "maxValues" \ "id") ==
+ (if (values.isEmpty) JNothing else JInt(values.max)),
+ file.stats)
+ } else {
+ assert(file.stats == null || file.stats.isEmpty)
+ }
+ file.partitionValues("part") -> rows.size
+ }.groupBy(_._1).map {
+ case (part, sizes) => part -> sizes.map(_._2).sorted
+ }
+ // Exact sizes detect premature rotation as well as files
exceeding the limit.
+ assert(fileSizes == Map(
+ (null, Seq(3)),
+ "0" -> Seq(4),
+ "1" -> (if (limit == 0) Seq(9) else Seq(1, 4, 4))))
+ checkAnswer(spark.read.format("delta").load(path).select("id",
"part"), expected)
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/gluten-ut/spark40/pom.xml b/gluten-ut/spark40/pom.xml
index e5cea2743b..179d479172 100644
--- a/gluten-ut/spark40/pom.xml
+++ b/gluten-ut/spark40/pom.xml
@@ -364,5 +364,35 @@
</plugins>
</build>
</profile>
+
+ <profile>
+ <id>delta</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-delta</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-delta-test-sources</id>
+ <configuration>
+ <sources combine.children="append">
+
<source>${project.basedir}/../spark35/src-delta/test/scala</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
</profiles>
</project>
diff --git a/gluten-ut/spark41/pom.xml b/gluten-ut/spark41/pom.xml
index 4d75307c29..dbb761d93f 100644
--- a/gluten-ut/spark41/pom.xml
+++ b/gluten-ut/spark41/pom.xml
@@ -365,5 +365,35 @@
</plugins>
</build>
</profile>
+
+ <profile>
+ <id>delta</id>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.gluten</groupId>
+ <artifactId>gluten-delta</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-delta-test-sources</id>
+ <configuration>
+ <sources combine.children="append">
+
<source>${project.basedir}/../spark35/src-delta/test/scala</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
</profiles>
</project>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
