malinjawi commented on code in PR #12836:
URL: https://github.com/apache/gluten/pull/12836#discussion_r3862950138
##########
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:
Updated. The map is now named `nativeScanMetrics`, and the comment states
that format-specific metrics may be updated on the driver or executors while
driver-only aliases are excluded.
##########
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:
Updated. `SerializedDeletionVectorPayload` now clones the constructor input,
with a regression test covering mutation of the caller-owned array. The on-disk
deferred path is unaffected.
##########
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:
I verified the deserialization order and kept the logic unchanged, with a
clearer comment. `defaultReadObject()` deserializes the nested `SQLMetric`s
first, and Spark `AccumulatorV2.readObject` registers each one when
`TaskContext` exists. The wrapper then records that state to avoid registering
them twice. If no context exists, `registerForCurrentTask()` performs the
deferred registration. Resetting the flag to false would append duplicate
accumulators to `TaskMetrics`; the end-to-end tests also assert exactly one
reported read attempt.
--
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]
