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 , 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. now clones the constructor input, with a regression test covering
mutation of the caller-owned array. The on-disk deferred path is unaffected.
##########
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:
Updated the method documentation to state that on-disk payload
materialization may perform blocking filesystem I/O, is intended for
executor-side split-to-protobuf conversion, and returns bytes that must not be
modified. The existing method name is retained for API compatibility.
##########
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:
I checked this path and kept the count where it is. It is a local
O(files-per-partition) traversal on the driver; protobuf construction occurs
later on the executor. Folding the count into normalization would widen the
shared Delta-version API for negligible cost. We can revisit it if profiling
shows this traversal is material.
--
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]
