This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 9728c02f fix: address failure caused by method signature change in
SPARK-48791 (#693)
9728c02f is described below
commit 9728c02fe4c3d4e63ed368eb2c2abba873f10372
Author: Parth Chandra <[email protected]>
AuthorDate: Sat Jul 20 06:33:38 2024 -0700
fix: address failure caused by method signature change in SPARK-48791 (#693)
* fix: address failure caused by method signature change in SPARK-48791
* fix build
* remove shim
* add comment
---
.../java/org/apache/comet/parquet/BatchReader.java | 30 +++++++++++++++++++++-
1 file changed, 29 insertions(+), 1 deletion(-)
diff --git a/common/src/main/java/org/apache/comet/parquet/BatchReader.java
b/common/src/main/java/org/apache/comet/parquet/BatchReader.java
index 4b63f84e..11c6d14d 100644
--- a/common/src/main/java/org/apache/comet/parquet/BatchReader.java
+++ b/common/src/main/java/org/apache/comet/parquet/BatchReader.java
@@ -21,6 +21,8 @@ package org.apache.comet.parquet;
import java.io.Closeable;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
@@ -33,6 +35,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import scala.Option;
+import scala.collection.Seq;
+import scala.collection.mutable.Buffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,6 +61,7 @@ import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;
import org.apache.spark.TaskContext;
import org.apache.spark.TaskContext$;
+import org.apache.spark.executor.TaskMetrics;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.comet.parquet.CometParquetReadSupport;
import org.apache.spark.sql.execution.datasources.PartitionedFile;
@@ -345,7 +350,7 @@ public class BatchReader extends RecordReader<Void,
ColumnarBatch> implements Cl
// Note that this tries to get thread local TaskContext object, if this is
called at other
// thread, it won't update the accumulator.
if (taskContext != null) {
- Option<AccumulatorV2<?, ?>> accu =
taskContext.taskMetrics().externalAccums().lastOption();
+ Option<AccumulatorV2<?, ?>> accu =
getTaskAccumulator(taskContext.taskMetrics());
if (accu.isDefined() &&
accu.get().getClass().getSimpleName().equals("NumRowGroupsAcc")) {
@SuppressWarnings("unchecked")
AccumulatorV2<Integer, Integer> intAccum = (AccumulatorV2<Integer,
Integer>) accu.get();
@@ -632,4 +637,27 @@ public class BatchReader extends RecordReader<Void,
ColumnarBatch> implements Cl
}
}
}
+
+ // Signature of externalAccums changed from returning a Buffer to returning
a Seq. If comet is
+ // expecting a Buffer but the Spark version returns a Seq or vice versa, we
get a
+ // method not found exception.
+ @SuppressWarnings("unchecked")
+ private Option<AccumulatorV2<?, ?>> getTaskAccumulator(TaskMetrics
taskMetrics) {
+ Method externalAccumsMethod;
+ try {
+ externalAccumsMethod =
TaskMetrics.class.getDeclaredMethod("externalAccums");
+ externalAccumsMethod.setAccessible(true);
+ String returnType = externalAccumsMethod.getReturnType().getName();
+ if (returnType.equals("scala.collection.mutable.Buffer")) {
+ return ((Buffer<AccumulatorV2<?, ?>>)
externalAccumsMethod.invoke(taskMetrics))
+ .lastOption();
+ } else if (returnType.equals("scala.collection.Seq")) {
+ return ((Seq<AccumulatorV2<?, ?>>)
externalAccumsMethod.invoke(taskMetrics)).lastOption();
+ } else {
+ return Option.apply(null); // None
+ }
+ } catch (NoSuchMethodException | InvocationTargetException |
IllegalAccessException e) {
+ return Option.apply(null); // None
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
