You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "malinjawi (via GitHub)" <gi...@apache.org> on 2026/10/13 12:11:42 UTC
[PR] [GLUTEN-12569][VL] Enable Delta for Spark 4.2 with Delta 4.4.1 [gluten]
malinjawi opened a new pull request, #13271:
URL: https://github.com/apache/gluten/pull/13271
## What changes are proposed in this pull request?
The `spark-4.2` profile still used the Spark 4.1 Delta jar (`delta-spark_4.1` 4.1.0), with a note that Delta had no Spark 4.2 release. Delta 4.4.0 (2026-08-20) and 4.4.1 publish `delta-spark_4.2_2.13`, built against Spark 4.2.0. This PR switches to it and turns on Delta in the Spark 4.2 CI lane.
- **Build:** `pom.xml` and `tools/gluten-it/pom.xml` now use `delta-spark_4.2` 4.4.1 for `spark-4.2`. Its APIs still fit the `src-delta40` trees, so `delta.binary.version` stays 40.
- **Tests that no longer compile:** two unused helpers copied from Delta's tests don't compile against 4.4 and are removed: `getCheckpointDfForFilesContainingFileActions` and `TableCommitCoordinatorClientTestHelper.commit`.
- **Partition columns in data files:** Gluten's native partitioned write always leaves partition columns out of the data files.
- Some tables need them in the files: Iceberg-compat tables, and tables with the `materializePartitionColumns` writer feature (Delta 4.1+). Those now fall back to the vanilla Delta write.
- Gluten ignored the feature before this PR, so Spark 4.1 was already affected. Delta 4.0 has no class for the feature, so the check matches it by name.
- **Delta 4.4's new default:** Delta 4.4 also writes partition columns into the files when the feature is absent (`delta.writePartitionColumnsToParquet`, default true). Those writes stay native and leave the columns out.
- Delta reads partition values from the log, so results are the same.
- This is documented in `VeloxDelta.md`. Native support is a follow-up.
- **Deletion vectors without an offset:** Delta 4.4 allows on-disk deletion vectors with no offset (delta-io/delta#6436) and reads them from the start of the file. Gluten threw on them. It now uses offset 0, as Delta's `DeletionVectorStoredBitmap` does.
- **CI:** `-Pdelta` is added to `build-test-spark42` and `spark-test-spark42`, matching the Spark 4.1 lane.
- **Docs:** `VeloxDelta.md` gains Spark 4.1 and 4.2 rows.
Running Delta's own test suite on Spark 4.2 (`delta_spark_ut.yml`) is left to a follow-up. It needs v4.4.1 test patches and a separate known-failures baseline.
Part of #12569.
## How was this patch tested?
- **Compile:** `test-compile` with `-Pdelta -Pspark-ut` passes on Spark 3.5, 4.0, 4.1 and 4.2.
- **`DeltaDeletionVectorScanInfoSuite`:** a new test points an offset-less descriptor at a copy of a deletion vector that starts at byte 0. It checks the payload matches the original.
- Locally, the suite passes on Spark 4.2 (10/10).
- With the fix reverted, the new test fails with the old "missing its offset" error.
- It is skipped below Delta 4.4.
- **`GlutenDeltaNativePartitionWriteSuite`:** a new test writes to a partitioned table with `materializePartitionColumns` enabled. It checks that every data file's Parquet footer contains the partition column. It runs on the Spark 4.1 and 4.2 lanes and is skipped below Delta 4.1.
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: IBM Bob
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6174269978
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "felipepessoto (via GitHub)" <gi...@apache.org>.
felipepessoto commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6275093892
@malinjawi sorry I'm a bit late. I was out and I'm still catching up.
I think ideally that we need separate known-issues and patch files for each Delta version. In this PR: https://github.com/apache/gluten/pull/13129 I'm separating the patches, hopefully it will make easier to organize it.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta for Spark 4.2 with Delta 4.4.1 [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6167403438
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan merged PR #13271:
URL: https://github.com/apache/gluten/pull/13271
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta for Spark 4.2 with Delta 4.4.1 [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6167809143
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta for Spark 4.2 with Delta 4.4.1 [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6151268166
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6203525820
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on code in PR #13271:
URL: https://github.com/apache/gluten/pull/13271#discussion_r4314823137
##########
backends-velox/src-delta4/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala:
##########
@@ -0,0 +1,329 @@
+/*
+ * 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.delta
+
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.extension.columnar.transition.Transitions
+
+import org.apache.spark.sql.{AnalysisException, Dataset, SparkSession}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.delta.actions.{AddFile, FileAction}
+import org.apache.spark.sql.delta.constraints.{Constraint, Constraints, DeltaInvariantCheckerExec}
+import org.apache.spark.sql.delta.files.{DeltaFileFormatWriter, GlutenDeltaFileFormatWriter, TransactionalWrite}
+import org.apache.spark.sql.delta.hooks.AutoCompact
+import org.apache.spark.sql.delta.perf.{DeltaOptimizedWriterExec, GlutenDeltaOptimizedWriterExec}
+import org.apache.spark.sql.delta.schema.{EmptySchemaWriteException, InnerInvariantViolationException}
+import org.apache.spark.sql.delta.shims.VariantShreddingShims
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.TableParquetVersionOption
+import org.apache.spark.sql.execution.{SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, FileFormatWriter, WriteJobStatsTracker}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.ScalaExtensions.OptionExt
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.mutable.ListBuffer
+
+/**
+ * An [[OptimisticTransaction]] whose `writeFiles` writes with the Velox writer.
+ *
+ * `writeFiles` is a copy of `TransactionalWrite.writeFiles` from delta-io/delta v4.4.1
+ * (spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala). The copied code
+ * is kept verbatim between `format: off` and `format: on`; Gluten's changes are only the
+ * `GLUTEN-SPLICE` regions. Removing those regions gives back Delta's method, so on a Delta bump
+ * re-copy the upstream method and re-apply the splices.
+ */
+class GlutenOptimisticTransaction(delegate: OptimisticTransaction)
+ extends OptimisticTransaction(
+ delegate.deltaLog,
+ delegate.catalogTable,
+ delegate.snapshot
+ ) {
+
+ // scalastyle:off
+ // format: off
+ override def writeFiles(
+ inputData: Dataset[_],
+ writeOptions: Option[DeltaOptions],
+ isOptimize: Boolean,
+ additionalConstraints: Seq[Constraint]): Seq[FileAction] = {
+ // GLUTEN-SPLICE-BEGIN: schemas the native write path cannot handle use Delta's writeFiles.
+ GlutenDeltaNativeWriteGate.unsupportedSchema(inputData.schema).foreach { reason =>
+ logInfo(s"$reason; falling back to the vanilla Delta write path.")
+ return super.writeFiles(inputData, writeOptions, isOptimize, additionalConstraints)
+ }
+ // GLUTEN-SPLICE-END
+ hasWritten = true
+
+ val spark = inputData.sparkSession
+ val (data, partitionSchema) = performCDCPartition(inputData)
+ val outputPath = deltaLog.dataPath
+
+ val (queryExecution, output, generatedColumnConstraints, trackFromData) =
+ normalizeData(deltaLog, writeOptions, data)
+ // Use the track set from the transaction if set,
+ // otherwise use the track set from `normalizeData()`.
+ val trackIdentityHighWaterMarks = trackHighWaterMarks.getOrElse(trackFromData)
+
+ val partitioningColumns = getPartitioningColumns(partitionSchema, output)
+
+ val committer = getCommitter(outputPath)
+
+ val (statsDataSchema, _) = getStatsSchema(output, partitionSchema)
+
+ // If Statistics Collection is enabled, then create a stats tracker that will be injected during
+ // the FileFormatWriter.write call below and will collect per-file stats using
+ // StatisticsCollection
+ val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection(output, outputPath,
+ partitionSchema, data)
+
+
+ val constraints =
+ Constraints.getAll(metadata, spark) ++ generatedColumnConstraints ++ additionalConstraints
+ Constraints.validateCheckConstraints(spark, constraints, deltaLog, metadata.schema)
+
+ val identityTrackerOpt = IdentityColumn.createIdentityColumnStatsTracker(
+ spark,
+ deltaLog.newDeltaHadoopConf(),
+ outputPath,
+ metadata.schema,
+ statsDataSchema,
+ trackIdentityHighWaterMarks
+ )
+
+ SQLExecution.withNewExecutionId(queryExecution, Option("deltaTransactionalWrite")) {
+ val outputSpec = FileFormatWriter.OutputSpec(
+ outputPath.toString,
+ Map.empty,
+ output)
+
+ val empty2NullPlan = convertEmptyToNullIfNeeded(queryExecution.executedPlan,
+ partitioningColumns, constraints)
+ val checkInvariants = DeltaInvariantCheckerExec(spark, empty2NullPlan, constraints)
+ // No need to plan optimized write if the write command is OPTIMIZE, which aims to produce
+ // evenly-balanced data files already.
+ val physicalPlan = if (!isOptimize &&
+ shouldOptimizeWrite(writeOptions, spark.sessionState.conf)) {
+ DeltaOptimizedWriterExec(checkInvariants, metadata.partitionColumns, deltaLog)
+ } else {
+ checkInvariants
+ }
+
+ val statsTrackers: ListBuffer[WriteJobStatsTracker] = ListBuffer()
+
+ if (spark.conf.get(DeltaSQLConf.DELTA_HISTORY_METRICS_ENABLED)) {
+ val basicWriteJobStatsTracker = new BasicWriteJobStatsTracker(
+ new SerializableConfiguration(deltaLog.newDeltaHadoopConf()),
+ BasicWriteJobStatsTracker.metrics)
+ registerSQLMetrics(spark, basicWriteJobStatsTracker.driverSideMetrics)
+ statsTrackers.append(basicWriteJobStatsTracker)
+ }
+
+ // Iceberg spec requires partition columns in data files
+ val writePartitionColumns = IcebergCompat.isAnyEnabled(metadata) ||
+ protocol.isFeatureSupported(MaterializePartitionColumnsTableFeature) ||
+ DeltaConfigs.WRITE_PARTITION_COLUMNS_TO_PARQUET.fromMetaData(metadata).getOrElse(true)
+ // Retain only a minimal selection of Spark writer options to avoid any potential
+ // compatibility issues
+ val filteredOptions = (writeOptions match {
+ case None => Map.empty[String, String]
+ case Some(writeOptions) =>
+ writeOptions.options.filterKeys { key =>
+ key.equalsIgnoreCase(DeltaOptions.MAX_RECORDS_PER_FILE) ||
+ key.equalsIgnoreCase(DeltaOptions.COMPRESSION) ||
+ key.equalsIgnoreCase(DeltaOptions.PARQUET_VERSION) ||
+ key.equalsIgnoreCase(DeltaOptions.PARQUET_OUTPUT_TIMESTAMP_TYPE)
+ }.toMap
+ }) + (DeltaOptions.WRITE_PARTITION_COLUMNS -> writePartitionColumns.toString)
+ val options = filteredOptions ++
+ VariantShreddingShims.getVariantInferShreddingSchemaOptions(
+ DeltaConfigs.ENABLE_VARIANT_SHREDDING.fromMetaData(metadata)) ++
+ TableParquetVersionOption.getWriterOptions(
+ spark = spark,
+ writerOptions = filteredOptions,
+ tableProperties = metadata.configuration)
+
+ try {
+ // GLUTEN-SPLICE-BEGIN: use the Velox writer when it reproduces the files Delta writes.
+ // The job conf Delta passes to its writer below, where parquet-mr reads its settings.
+ // scalastyle:off deltahadoopconfiguration
+ val jobHadoopConf =
+ spark.sessionState.newHadoopConfWithOptions(metadata.configuration ++ deltaLog.options)
+ // scalastyle:on deltahadoopconfiguration
+ val nativeUnsupportedReason = GlutenDeltaNativeWriteGate.unsupportedWrite(
+ spark.sessionState.conf,
+ jobHadoopConf,
+ options,
+ DataTypeUtils.fromAttributes(output),
+ partitioningColumns,
+ metadata)
+ if (nativeUnsupportedReason.isEmpty) {
+ GlutenDeltaFileFormatWriter.write(
+ sparkSession = spark,
+ plan = glutenPhysicalPlan(spark, empty2NullPlan, constraints, isOptimize, writeOptions),
+ fileFormat = new GlutenDeltaParquetFileFormat(protocol, metadata),
+ committer = committer,
+ outputSpec = outputSpec,
+ hadoopConf = jobHadoopConf,
+ partitionColumns = partitioningColumns,
+ bucketSpec = None,
+ statsTrackers = optionalStatsTracker.toSeq
+ ++ statsTrackers
+ ++ identityTrackerOpt.toSeq,
+ options = options)
+ } else {
+ logInfo(s"${nativeUnsupportedReason.get}; writing with the vanilla Delta writer.")
+ // GLUTEN-SPLICE-END
+ DeltaFileFormatWriter.write(
+ sparkSession = spark,
+ plan = physicalPlan,
+ fileFormat = deltaLog.fileFormat(protocol, metadata), // TODO support changing formats.
+ committer = committer,
+ outputSpec = outputSpec,
+ // scalastyle:off deltahadoopconfiguration
+ hadoopConf =
+ spark.sessionState.newHadoopConfWithOptions(metadata.configuration ++ deltaLog.options),
+ // scalastyle:on deltahadoopconfiguration
+ partitionColumns = partitioningColumns,
+ bucketSpec = None,
+ statsTrackers = optionalStatsTracker.toSeq
+ ++ statsTrackers
+ ++ identityTrackerOpt.toSeq,
+ options = options)
Review Comment:
Thanks. Leaving this one as is on purpose: the vanilla branch is a verbatim copy of Delta 4.4.1’s `writeFiles`, so a Delta upgrade is just a re-copy. Reusing `jobHadoopConf` would mean editing Delta’s own line. It also wouldn’t change anything: the native path already uses the same conf the gate checked, and the fallback only builds one extra conf object.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6212347517
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6179253890
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6266965803
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6170523056
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "malinjawi (via GitHub)" <gi...@apache.org>.
malinjawi commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6247542277
cc: @zhouyuan @felipepessoto @zhztheplayer
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13271:
URL: https://github.com/apache/gluten/pull/13271#discussion_r4312331225
##########
backends-velox/src-delta4/main/scala/org/apache/spark/sql/delta/GlutenOptimisticTransaction.scala:
##########
@@ -0,0 +1,329 @@
+/*
+ * 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.delta
+
+import org.apache.gluten.backendsapi.velox.VeloxBatchType
+import org.apache.gluten.extension.columnar.transition.Transitions
+
+import org.apache.spark.sql.{AnalysisException, Dataset, SparkSession}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.delta.actions.{AddFile, FileAction}
+import org.apache.spark.sql.delta.constraints.{Constraint, Constraints, DeltaInvariantCheckerExec}
+import org.apache.spark.sql.delta.files.{DeltaFileFormatWriter, GlutenDeltaFileFormatWriter, TransactionalWrite}
+import org.apache.spark.sql.delta.hooks.AutoCompact
+import org.apache.spark.sql.delta.perf.{DeltaOptimizedWriterExec, GlutenDeltaOptimizedWriterExec}
+import org.apache.spark.sql.delta.schema.{EmptySchemaWriteException, InnerInvariantViolationException}
+import org.apache.spark.sql.delta.shims.VariantShreddingShims
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+import org.apache.spark.sql.delta.util.TableParquetVersionOption
+import org.apache.spark.sql.execution.{SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, FileFormatWriter, WriteJobStatsTracker}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.ScalaExtensions.OptionExt
+import org.apache.spark.util.SerializableConfiguration
+
+import scala.collection.mutable.ListBuffer
+
+/**
+ * An [[OptimisticTransaction]] whose `writeFiles` writes with the Velox writer.
+ *
+ * `writeFiles` is a copy of `TransactionalWrite.writeFiles` from delta-io/delta v4.4.1
+ * (spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala). The copied code
+ * is kept verbatim between `format: off` and `format: on`; Gluten's changes are only the
+ * `GLUTEN-SPLICE` regions. Removing those regions gives back Delta's method, so on a Delta bump
+ * re-copy the upstream method and re-apply the splices.
+ */
+class GlutenOptimisticTransaction(delegate: OptimisticTransaction)
+ extends OptimisticTransaction(
+ delegate.deltaLog,
+ delegate.catalogTable,
+ delegate.snapshot
+ ) {
+
+ // scalastyle:off
+ // format: off
+ override def writeFiles(
+ inputData: Dataset[_],
+ writeOptions: Option[DeltaOptions],
+ isOptimize: Boolean,
+ additionalConstraints: Seq[Constraint]): Seq[FileAction] = {
+ // GLUTEN-SPLICE-BEGIN: schemas the native write path cannot handle use Delta's writeFiles.
+ GlutenDeltaNativeWriteGate.unsupportedSchema(inputData.schema).foreach { reason =>
+ logInfo(s"$reason; falling back to the vanilla Delta write path.")
+ return super.writeFiles(inputData, writeOptions, isOptimize, additionalConstraints)
+ }
+ // GLUTEN-SPLICE-END
+ hasWritten = true
+
+ val spark = inputData.sparkSession
+ val (data, partitionSchema) = performCDCPartition(inputData)
+ val outputPath = deltaLog.dataPath
+
+ val (queryExecution, output, generatedColumnConstraints, trackFromData) =
+ normalizeData(deltaLog, writeOptions, data)
+ // Use the track set from the transaction if set,
+ // otherwise use the track set from `normalizeData()`.
+ val trackIdentityHighWaterMarks = trackHighWaterMarks.getOrElse(trackFromData)
+
+ val partitioningColumns = getPartitioningColumns(partitionSchema, output)
+
+ val committer = getCommitter(outputPath)
+
+ val (statsDataSchema, _) = getStatsSchema(output, partitionSchema)
+
+ // If Statistics Collection is enabled, then create a stats tracker that will be injected during
+ // the FileFormatWriter.write call below and will collect per-file stats using
+ // StatisticsCollection
+ val (optionalStatsTracker, _) = getOptionalStatsTrackerAndStatsCollection(output, outputPath,
+ partitionSchema, data)
+
+
+ val constraints =
+ Constraints.getAll(metadata, spark) ++ generatedColumnConstraints ++ additionalConstraints
+ Constraints.validateCheckConstraints(spark, constraints, deltaLog, metadata.schema)
+
+ val identityTrackerOpt = IdentityColumn.createIdentityColumnStatsTracker(
+ spark,
+ deltaLog.newDeltaHadoopConf(),
+ outputPath,
+ metadata.schema,
+ statsDataSchema,
+ trackIdentityHighWaterMarks
+ )
+
+ SQLExecution.withNewExecutionId(queryExecution, Option("deltaTransactionalWrite")) {
+ val outputSpec = FileFormatWriter.OutputSpec(
+ outputPath.toString,
+ Map.empty,
+ output)
+
+ val empty2NullPlan = convertEmptyToNullIfNeeded(queryExecution.executedPlan,
+ partitioningColumns, constraints)
+ val checkInvariants = DeltaInvariantCheckerExec(spark, empty2NullPlan, constraints)
+ // No need to plan optimized write if the write command is OPTIMIZE, which aims to produce
+ // evenly-balanced data files already.
+ val physicalPlan = if (!isOptimize &&
+ shouldOptimizeWrite(writeOptions, spark.sessionState.conf)) {
+ DeltaOptimizedWriterExec(checkInvariants, metadata.partitionColumns, deltaLog)
+ } else {
+ checkInvariants
+ }
+
+ val statsTrackers: ListBuffer[WriteJobStatsTracker] = ListBuffer()
+
+ if (spark.conf.get(DeltaSQLConf.DELTA_HISTORY_METRICS_ENABLED)) {
+ val basicWriteJobStatsTracker = new BasicWriteJobStatsTracker(
+ new SerializableConfiguration(deltaLog.newDeltaHadoopConf()),
+ BasicWriteJobStatsTracker.metrics)
+ registerSQLMetrics(spark, basicWriteJobStatsTracker.driverSideMetrics)
+ statsTrackers.append(basicWriteJobStatsTracker)
+ }
+
+ // Iceberg spec requires partition columns in data files
+ val writePartitionColumns = IcebergCompat.isAnyEnabled(metadata) ||
+ protocol.isFeatureSupported(MaterializePartitionColumnsTableFeature) ||
+ DeltaConfigs.WRITE_PARTITION_COLUMNS_TO_PARQUET.fromMetaData(metadata).getOrElse(true)
+ // Retain only a minimal selection of Spark writer options to avoid any potential
+ // compatibility issues
+ val filteredOptions = (writeOptions match {
+ case None => Map.empty[String, String]
+ case Some(writeOptions) =>
+ writeOptions.options.filterKeys { key =>
+ key.equalsIgnoreCase(DeltaOptions.MAX_RECORDS_PER_FILE) ||
+ key.equalsIgnoreCase(DeltaOptions.COMPRESSION) ||
+ key.equalsIgnoreCase(DeltaOptions.PARQUET_VERSION) ||
+ key.equalsIgnoreCase(DeltaOptions.PARQUET_OUTPUT_TIMESTAMP_TYPE)
+ }.toMap
+ }) + (DeltaOptions.WRITE_PARTITION_COLUMNS -> writePartitionColumns.toString)
+ val options = filteredOptions ++
+ VariantShreddingShims.getVariantInferShreddingSchemaOptions(
+ DeltaConfigs.ENABLE_VARIANT_SHREDDING.fromMetaData(metadata)) ++
+ TableParquetVersionOption.getWriterOptions(
+ spark = spark,
+ writerOptions = filteredOptions,
+ tableProperties = metadata.configuration)
+
+ try {
+ // GLUTEN-SPLICE-BEGIN: use the Velox writer when it reproduces the files Delta writes.
+ // The job conf Delta passes to its writer below, where parquet-mr reads its settings.
+ // scalastyle:off deltahadoopconfiguration
+ val jobHadoopConf =
+ spark.sessionState.newHadoopConfWithOptions(metadata.configuration ++ deltaLog.options)
+ // scalastyle:on deltahadoopconfiguration
+ val nativeUnsupportedReason = GlutenDeltaNativeWriteGate.unsupportedWrite(
+ spark.sessionState.conf,
+ jobHadoopConf,
+ options,
+ DataTypeUtils.fromAttributes(output),
+ partitioningColumns,
+ metadata)
+ if (nativeUnsupportedReason.isEmpty) {
+ GlutenDeltaFileFormatWriter.write(
+ sparkSession = spark,
+ plan = glutenPhysicalPlan(spark, empty2NullPlan, constraints, isOptimize, writeOptions),
+ fileFormat = new GlutenDeltaParquetFileFormat(protocol, metadata),
+ committer = committer,
+ outputSpec = outputSpec,
+ hadoopConf = jobHadoopConf,
+ partitionColumns = partitioningColumns,
+ bucketSpec = None,
+ statsTrackers = optionalStatsTracker.toSeq
+ ++ statsTrackers
+ ++ identityTrackerOpt.toSeq,
+ options = options)
+ } else {
+ logInfo(s"${nativeUnsupportedReason.get}; writing with the vanilla Delta writer.")
+ // GLUTEN-SPLICE-END
+ DeltaFileFormatWriter.write(
+ sparkSession = spark,
+ plan = physicalPlan,
+ fileFormat = deltaLog.fileFormat(protocol, metadata), // TODO support changing formats.
+ committer = committer,
+ outputSpec = outputSpec,
+ // scalastyle:off deltahadoopconfiguration
+ hadoopConf =
+ spark.sessionState.newHadoopConfWithOptions(metadata.configuration ++ deltaLog.options),
+ // scalastyle:on deltahadoopconfiguration
+ partitionColumns = partitioningColumns,
+ bucketSpec = None,
+ statsTrackers = optionalStatsTracker.toSeq
+ ++ statsTrackers
+ ++ identityTrackerOpt.toSeq,
+ options = options)
Review Comment:
The job Hadoop conf is constructed twice with the same inputs (`metadata.configuration ++ deltaLog.options`). Since `jobHadoopConf` is already computed for the gate and native writer, reuse it for the vanilla `DeltaFileFormatWriter.write` path as well. This reduces duplication and ensures the fallback path uses the exact same conf instance that was evaluated by the gate.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6198481000
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta for Spark 4.2 with Delta 4.4.1 [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6167500250
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [GLUTEN-12569][VL] Enable Delta on Spark 4.2: one Delta 4.4.1 release for Spark 4.x, following Delta's write path [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13271:
URL: https://github.com/apache/gluten/pull/13271#issuecomment-6169187393
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org