Copilot commented on code in PR #12836:
URL: https://github.com/apache/gluten/pull/12836#discussion_r3861423801
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala:
##########
@@ -116,11 +116,14 @@ abstract class FileSourceScanExecTransformerBase(
disableBucketedScan)
with DatasourceScanTransformer {
+ /** Format-specific metrics that should be displayed with the native file
scan. */
+ protected def additionalScanMetrics: Map[String, SQLMetric] = Map.empty
+
// Executor-side metrics only (excludes driverMetricsAlias).
@transient private lazy val executorSideScanMetrics: Map[String, SQLMetric] =
BackendsApiManager.getMetricsApiInstance
.genFileSourceScanTransformerMetrics(sparkContext)
- .filter(m => !driverMetricsAlias.contains(m._1))
+ .filter(m => !driverMetricsAlias.contains(m._1)) ++ additionalScanMetrics
Review Comment:
The comment says this map contains \"Executor-side metrics only\", but
`additionalScanMetrics` can include metrics that are updated on the driver
(e.g., descriptor prep time/count). Please update the comment (or rename the
val) to reflect that this is the metric set attached to the native scan, not
necessarily executor-only in terms of update location.
##########
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java:
##########
@@ -112,7 +164,11 @@ public long deletionVectorCardinality() {
}
public byte[] serializedDeletionVector() {
- return serializedDeletionVector;
+ return deletionVectorPayload.materialize();
+ }
Review Comment:
`serializedDeletionVector()` previously returned already-available bytes; it
now may trigger deferred I/O and blocking work via `materialize()`. Since this
is a public API surface on `DeltaFileReadOptions`, please document this
behavior on the method (and/or consider exposing a more explicit API such as
`deletionVectorPayload()` or `materializeDeletionVector()`), so callers don’t
accidentally force remote reads on the driver.
##########
gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java:
##########
@@ -79,24 +79,76 @@ public enum RowIndexFilterType {
IF_NOT_CONTAINED
}
+ /**
+ * Serializable source for a deletion-vector payload.
+ *
+ * <p>The source travels inside a Spark input partition. Implementations may
therefore defer
+ * remote I/O until {@link #materialize()} is called while the split is
converted to protobuf on
+ * an executor. The returned byte array must not be modified: protobuf wraps
it without copying.
+ */
+ public interface DeletionVectorPayload extends Serializable {
+ byte[] materialize();
+
+ /** Returns whether the payload bytes are already resident in this object.
*/
+ boolean isMaterialized();
+ }
+
+ /** A payload source for inline DVs whose bytes are already present in Delta
metadata. */
+ public static final class SerializedDeletionVectorPayload implements
DeletionVectorPayload {
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] payload;
+
+ public SerializedDeletionVectorPayload(byte[] payload) {
+ this.payload = payload == null ? new byte[0] : payload;
Review Comment:
`SerializedDeletionVectorPayload` stores the caller-provided byte array by
reference, but the new class-level contract explicitly says the returned byte
array must not be modified (protobuf wraps without copying). To enforce this,
defensively copy the input (e.g., `payload.clone()`), otherwise external
mutation after construction can corrupt serialized splits/protobuf payloads.
##########
gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.gluten.delta
+
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.execution.metric.SQLMetric
+
+import java.io.ObjectInputStream
+
+/** Metrics updated while an executor materializes an on-disk deletion-vector
payload. */
+final case class DeletionVectorReadMetrics(
+ readTimeNanos: SQLMetric,
+ readBytes: SQLMetric,
+ readAttempts: SQLMetric) {
+
+ @transient @volatile private var registeredInTask = false
+
+ /**
+ * Spark deserializes RDD partitions before installing `TaskContext`, so
accumulators nested in a
+ * partition cannot register from `AccumulatorV2.readObject`. Register them
when deferred I/O
+ * first runs inside the task instead. The shared metrics object makes this
once-per-task even
+ * when a partition contains multiple deletion vectors.
+ */
+ def registerForCurrentTask(): Unit = {
+ if (!registeredInTask && TaskContext.get() != null) {
+ this.synchronized {
+ if (!registeredInTask) {
+ registeredInTask = TaskAccumulatorRegistry.registerForCurrentTask(
+ readTimeNanos,
+ readBytes,
+ readAttempts)
+ }
+ }
+ }
+ }
+
+ /** Avoid double registration when Spark deserializes this object after
installing TaskContext. */
+ private def readObject(input: ObjectInputStream): Unit = {
+ input.defaultReadObject()
+ registeredInTask = TaskContext.get() != null
Review Comment:
`readObject` sets `registeredInTask = true` whenever a `TaskContext` is
present, but it does not actually register the accumulators. If deserialization
happens after `TaskContext` is installed (which can occur depending on what is
being deserialized when), `registerForCurrentTask()` will then permanently skip
registration and metric updates may not be reported. Prefer making registration
idempotent (always attempt
`TaskAccumulatorRegistry.registerForCurrentTask(...)` under the existing
synchronization/flag) rather than inferring registration from `TaskContext`
presence.
##########
gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala:
##########
@@ -129,10 +151,21 @@ case class DeltaScanTransformer(
val tableRootPath = tahoe.path
splitInfos.zip(partitions).map {
case (localFiles: LocalFilesNode, (filePartition: FilePartition, _))
=>
- DeltaDeletionVectorScanInfo
- .normalize(filePartition.files.toSeq, tableRootPath)
+ val startedAt = System.nanoTime()
+ val normalized =
+ try {
+ DeltaDeletionVectorScanInfo.normalize(
+ filePartition.files.toSeq,
+ tableRootPath,
+ Some(deletionVectorReadMetrics))
+ } finally {
+ metrics("dvDescriptorPreparationTime").add(System.nanoTime() -
startedAt)
+ }
+ normalized
.map {
case (otherMetadataColumns, deltaReadOptions) =>
+ metrics("dvDescriptorCount")
+ .add(deltaReadOptions.count(_.hasDeletionVector()).toLong)
Review Comment:
This adds an extra full traversal over `deltaReadOptions` per partition.
Since the code will also iterate the same collection to build the local files
protobuf, consider computing the DV count during normalization (e.g., alongside
`scanInfos`) and returning it, or accumulate the count while iterating for
protobuf construction, to avoid redundant passes on large partitions.
--
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]
