From 38feaade7c89e5aa4cc25dc2cae3ef478be0c4cf Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Wed, 14 Oct 2026 16:46:23 -0600
Subject: [PATCH 1/3] build: remove Guava dependency

---
 .../adding_a_new_operator.md                  |  8 +--
 pom.xml                                       | 20 -------
 spark/pom.xml                                 | 31 -----------
 .../CometBypassMergeSortShuffleWriter.java    | 35 +++---------
 .../shuffle/CometDiskBlockWriter.java         |  4 +-
 .../shuffle/CometUnsafeShuffleWriter.java     | 39 ++++++++-----
 .../spark/sql/comet/CometBatchScanExec.scala  |  6 +-
 .../comet/CometBroadcastExchangeExec.scala    |  5 +-
 .../spark/sql/comet/CometCoalesceExec.scala   |  6 +-
 .../sql/comet/CometCollectLimitExec.scala     |  6 +-
 .../sql/comet/CometCsvNativeScanExec.scala    |  6 +-
 .../comet/CometIcebergNativeScanExec.scala    |  8 +--
 .../sql/comet/CometLocalTableScanExec.scala   |  6 +-
 .../spark/sql/comet/CometLocalTopKExec.scala  |  6 +-
 .../spark/sql/comet/CometNativeScanExec.scala |  6 +-
 .../spark/sql/comet/CometRangeExec.scala      |  6 +-
 .../spark/sql/comet/CometSampleExec.scala     |  6 +-
 .../spark/sql/comet/CometWindowExec.scala     |  6 +-
 .../sql/comet/CometWindowGroupLimitExec.scala |  6 +-
 .../shuffle/CometShuffleExchangeExec.scala    |  5 +-
 .../apache/spark/sql/comet/operators.scala    | 30 +++++-----
 .../comet/parquet/ParquetReadSuite.scala      |  5 +-
 .../sql/comet/CometPlanEqualitySuite.scala    | 55 ++++++++++++++++---
 23 files changed, 141 insertions(+), 170 deletions(-)

diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md
index 15c4e9169b6..88272332882 100644
--- a/docs/source/contributor-guide/adding_a_new_operator.md
+++ b/docs/source/contributor-guide/adding_a_new_operator.md
@@ -226,7 +226,7 @@ The validation workflow in `CometExecRule.isOperatorEnabled`:
 #### Simple Example (Filter)
 
 ```scala
-import com.google.common.base.Objects
+import java.util.Objects
 
 object CometFilterExec extends CometOperatorSerde[FilterExec] {
 
@@ -284,14 +284,14 @@ case class CometFilterExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, condition, child)
+  override def hashCode(): Int = Objects.hash(output, condition, child)
 }
 ```
 
 #### More Complex Example (Project)
 
 ```scala
-import com.google.common.base.Objects
+import java.util.Objects
 
 object CometProjectExec extends CometOperatorSerde[ProjectExec] {
 
@@ -348,7 +348,7 @@ case class CometProjectExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, projectList, child)
+  override def hashCode(): Int = Objects.hash(output, projectList, child)
 
   override protected def outputExpressions: Seq[NamedExpression] = projectList
 }
diff --git a/pom.xml b/pom.xml
index a020f0719f9..279bbd67aa7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -92,7 +92,6 @@ under the License.
          release that exists for each of their Scala versions (2.13.16, 2.13.17 and 2.13.18). -->
     <semanticdb.version>4.13.10</semanticdb.version>
     <slf4j.version>2.0.17</slf4j.version>
-    <guava.version>33.2.1-jre</guava.version>
     <testcontainers.version>1.21.4</testcontainers.version>
     <amazon-awssdk-v2.version>2.31.51</amazon-awssdk-v2.version>
     <!-- AWS SDK v1, present at runtime on Hadoop 3.3.x (spark-3.x profiles). Used at provided
@@ -334,16 +333,6 @@ under the License.
         <scope>${spark.maven.scope}</scope>
       </dependency>
 
-      <!-- Shaded deps marked as provided. These are promoted to compile scope
-           in the modules where we want the shaded classes to appear in the
-           associated jar. -->
-      <dependency>
-        <groupId>com.google.guava</groupId>
-        <artifactId>guava</artifactId>
-        <version>${guava.version}</version>
-      </dependency>
-      <!-- End of shaded deps -->
-
       <!-- Test dependencies -->
       <dependency>
         <groupId>org.apache.spark</groupId>
@@ -1276,15 +1265,6 @@ under the License.
                         <ignoreClass>javax.annotation.meta.TypeQualifierNickname</ignoreClass>
                       </ignoreClasses>
                     </dependency>
-                    <dependency>
-                      <groupId>com.google.guava</groupId>
-                      <artifactId>guava</artifactId>
-                      <ignoreClasses>
-                        <ignoreClass>com.google.thirdparty.publicsuffix.TrieParser</ignoreClass>
-                        <ignoreClass>com.google.thirdparty.publicsuffix.PublicSuffixPatterns</ignoreClass>
-                        <ignoreClass>com.google.thirdparty.publicsuffix.PublicSuffixType</ignoreClass>
-                      </ignoreClasses>
-                    </dependency>
                     <dependency>
                       <groupId>org.apache.datasketches</groupId>
                       <artifactId>datasketches-memory</artifactId>
diff --git a/spark/pom.xml b/spark/pom.xml
index 18c0ac60914..1a61581fca1 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -123,28 +123,6 @@ under the License.
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>com.google.guava</groupId>
-      <artifactId>guava</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>*</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-      </exclusions>
-    </dependency>
-    <!-- failureaccess: Guava's companion artifact (split out at Guava 27).
-         Guava's transitive deps are wildcard-excluded above to keep the
-         shaded production JAR small, which strips failureaccess too. At
-         test runtime any Guava call into futures (e.g. via Hadoop or
-         Spark internals) needs InternalFutureFailureAccess on the
-         classpath — explicitly pull it in here at test scope. -->
-    <dependency>
-      <groupId>com.google.guava</groupId>
-      <artifactId>failureaccess</artifactId>
-      <version>1.0.2</version>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.codehaus.jackson</groupId>
       <artifactId>jackson-mapper-asl</artifactId>
@@ -720,7 +698,6 @@ under the License.
                   <include>org.apache.arrow:*</include>
                   <!-- Relocate Protobuf since Spark uses 2.5.0 while Comet uses 3.x -->
                   <include>com.google.protobuf:protobuf-java</include>
-                  <include>com.google.guava:guava</include>
                 </includes>
               </artifactSet>
               <filters>
@@ -780,14 +757,6 @@ under the License.
                   <pattern>com.google.protobuf</pattern>
                   <shadedPattern>${comet.shade.packageName}.protobuf</shadedPattern>
                 </relocation>
-                <relocation>
-                  <pattern>com.google.common</pattern>
-                  <shadedPattern>${comet.shade.packageName}.guava</shadedPattern>
-                </relocation>
-                <relocation>
-                  <pattern>com.google.thirdparty</pattern>
-                  <shadedPattern>${comet.shade.packageName}.guava.thirdparty</shadedPattern>
-                </relocation>
               </relocations>
               <transformers>
                 <!-- arrow-compression ships META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory.
diff --git a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometBypassMergeSortShuffleWriter.java b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometBypassMergeSortShuffleWriter.java
index 24929648b1d..be09b970e90 100644
--- a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometBypassMergeSortShuffleWriter.java
+++ b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometBypassMergeSortShuffleWriter.java
@@ -61,8 +61,6 @@
 import org.apache.spark.storage.TempShuffleBlockId;
 import org.apache.spark.util.Utils;
 
-import com.google.common.io.Closeables;
-
 import org.apache.comet.CometConf$;
 import org.apache.comet.Native;
 
@@ -364,36 +362,19 @@ private long[] writePartitionedData(ShuffleMapOutputWriter mapOutputWriter) thro
 
   private void writePartitionedDataWithChannel(File file, WritableByteChannelWrapper outputChannel)
       throws IOException {
-    boolean copyThrewException = true;
-    try {
-      FileInputStream in = new FileInputStream(file);
-      try (FileChannel inputChannel = in.getChannel()) {
-        Utils.copyFileStreamNIO(inputChannel, outputChannel.channel(), 0L, inputChannel.size());
-        copyThrewException = false;
-      } finally {
-        Closeables.close(in, copyThrewException);
-      }
-    } finally {
-      Closeables.close(outputChannel, copyThrewException);
+    try (FileInputStream in = new FileInputStream(file);
+        WritableByteChannelWrapper ignored = outputChannel) {
+      FileChannel inputChannel = in.getChannel();
+      Utils.copyFileStreamNIO(inputChannel, outputChannel.channel(), 0L, inputChannel.size());
     }
   }
 
   private void writePartitionedDataWithStream(File file, ShufflePartitionWriter writer)
       throws IOException {
-    boolean copyThrewException = true;
-    FileInputStream in = new FileInputStream(file);
-    OutputStream outputStream;
-    try {
-      outputStream = blockManager.serializerManager().wrapForEncryption(writer.openStream());
-
-      try {
-        Utils.copyStream(in, outputStream, false, false);
-        copyThrewException = false;
-      } finally {
-        Closeables.close(outputStream, copyThrewException);
-      }
-    } finally {
-      Closeables.close(in, copyThrewException);
+    try (FileInputStream in = new FileInputStream(file);
+        OutputStream outputStream =
+            blockManager.serializerManager().wrapForEncryption(writer.openStream())) {
+      Utils.copyStream(in, outputStream, false, false);
     }
   }
 
diff --git a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometDiskBlockWriter.java b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometDiskBlockWriter.java
index 839a69bd765..307104fb2e4 100644
--- a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometDiskBlockWriter.java
+++ b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometDiskBlockWriter.java
@@ -44,8 +44,6 @@
 import org.apache.spark.unsafe.Platform;
 import org.apache.spark.unsafe.UnsafeAlignedOffset;
 
-import com.google.common.annotations.VisibleForTesting;
-
 import org.apache.comet.CometConf$;
 import org.apache.comet.Native;
 
@@ -70,7 +68,7 @@ public final class CometDiskBlockWriter {
 
   private final TaskContext taskContext;
 
-  @VisibleForTesting static final int DEFAULT_INITIAL_SER_BUFFER_SIZE = 1024 * 1024;
+  static final int DEFAULT_INITIAL_SER_BUFFER_SIZE = 1024 * 1024;
 
   // Copied from Spark `org.apache.spark.shuffle.sort.PackedRecordPointer.MAXIMUM_PAGE_SIZE_BYTES`
   static final int MAXIMUM_PAGE_SIZE_BYTES = 1 << 27;
diff --git a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometUnsafeShuffleWriter.java b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometUnsafeShuffleWriter.java
index 42ffb603b98..024fdd8b39c 100644
--- a/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometUnsafeShuffleWriter.java
+++ b/spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/CometUnsafeShuffleWriter.java
@@ -19,6 +19,7 @@
 
 package org.apache.spark.sql.comet.execution.shuffle;
 
+import java.io.Closeable;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -74,10 +75,6 @@
 import org.apache.spark.util.TaskCompletionListener;
 import org.apache.spark.util.Utils;
 
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.io.ByteStreams;
-import com.google.common.io.Closeables;
-
 import org.apache.comet.CometConf;
 import org.apache.comet.Native;
 
@@ -102,7 +99,7 @@
 @Private
 public class CometUnsafeShuffleWriter<K, V> extends ShuffleWriter<K, V> {
 
-  @VisibleForTesting static final int DEFAULT_INITIAL_SER_BUFFER_SIZE = 1024 * 1024;
+  static final int DEFAULT_INITIAL_SER_BUFFER_SIZE = 1024 * 1024;
   private static final Logger logger = LoggerFactory.getLogger(CometUnsafeShuffleWriter.class);
   private static final ClassTag<Object> OBJECT_CLASS_TAG = ClassTag$.MODULE$.Object();
   private final BlockManager blockManager;
@@ -201,6 +198,22 @@ private static OutputStream openStreamUnchecked(ShufflePartitionWriter writer) {
     }
   }
 
+  /** Close a resource without masking an exception already thrown by the copy operation. */
+  private static void close(Closeable closeable, boolean copyThrewException) throws IOException {
+    if (closeable == null) {
+      return;
+    }
+    try {
+      closeable.close();
+    } catch (IOException e) {
+      if (copyThrewException) {
+        logger.warn("Suppressing exception thrown while closing a shuffle resource", e);
+      } else {
+        throw e;
+      }
+    }
+  }
+
   private void updatePeakMemoryUsed() {
     // sorter can be null if this writer is closed
     if (sorter != null) {
@@ -218,7 +231,6 @@ public long getPeakMemoryUsedBytes() {
   }
 
   /** This convenience method should only be called in test code. */
-  @VisibleForTesting
   public void write(Iterator<Product2<K, V>> records) throws IOException {
     write(asScala(records));
   }
@@ -300,7 +312,6 @@ private void open() {
     serOutputStream = serializer.serializeStream(serBuffer);
   }
 
-  @VisibleForTesting
   void closeAndWriteOutput() throws IOException {
     assert (sorter != null);
     updatePeakMemoryUsed();
@@ -326,7 +337,6 @@ void closeAndWriteOutput() throws IOException {
     mapStatus = MapStatusHelper.apply(blockManager.shuffleServerId(), partitionLengths, mapId);
   }
 
-  @VisibleForTesting
   /** Serializes records and inserts into external sorter */
   void insertRecordIntoSorter(Product2<K, V> record) throws IOException {
     assert (sorter != null);
@@ -344,7 +354,6 @@ void insertRecordIntoSorter(Product2<K, V> record) throws IOException {
         serBuffer.getBuf(), Platform.BYTE_ARRAY_OFFSET, serializedRecordSize, partitionId);
   }
 
-  @VisibleForTesting
   void forceSorterToSpill() throws IOException {
     assert (sorter != null);
     sorter.spill();
@@ -496,16 +505,16 @@ private void mergeSpillsWithFileStream(SpillInfo[] spills, ShuffleMapOutputWrite
               try {
                 partitionInputStream =
                     new LimitedInputStream(spillInputStreams[i], partitionLengthInSpill, false);
-                ByteStreams.copy(partitionInputStream, partitionOutput);
+                partitionInputStream.transferTo(partitionOutput);
                 copySpillThrewException = false;
               } finally {
-                Closeables.close(partitionInputStream, copySpillThrewException);
+                close(partitionInputStream, copySpillThrewException);
               }
             }
           }
           copyThrewException = false;
         } finally {
-          Closeables.close(partitionOutput, copyThrewException);
+          close(partitionOutput, copyThrewException);
         }
         long numBytesWritten = writer.getNumBytesWritten();
         writeMetrics.incBytesWritten(numBytesWritten);
@@ -515,7 +524,7 @@ private void mergeSpillsWithFileStream(SpillInfo[] spills, ShuffleMapOutputWrite
       // To avoid masking exceptions that caused us to prematurely enter the finally block, only
       // throw exceptions during cleanup if threwException == false.
       for (InputStream stream : spillInputStreams) {
-        Closeables.close(stream, threwException);
+        close(stream, threwException);
       }
     }
   }
@@ -571,7 +580,7 @@ private void mergeSpillsWithTransferTo(SpillInfo[] spills, ShuffleMapOutputWrite
             writeMetrics.incWriteTime(System.nanoTime() - writeStartTime);
           }
         } finally {
-          Closeables.close(resolvedChannel, copyThrewException);
+          close(resolvedChannel, copyThrewException);
         }
         long numBytes = writer.getNumBytesWritten();
         writeMetrics.incBytesWritten(numBytes);
@@ -582,7 +591,7 @@ private void mergeSpillsWithTransferTo(SpillInfo[] spills, ShuffleMapOutputWrite
       // throw exceptions during cleanup if threwException == false.
       for (int i = 0; i < spills.length; i++) {
         assert (spillInputChannelPositions[i] == spills[i].file.length());
-        Closeables.close(spillInputChannels[i], threwException);
+        close(spillInputChannels[i], threwException);
       }
     }
   }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala
index 4f807dc1c79..73a07a100b8 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometBatchScanExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.collection.mutable.ListBuffer
 
 import org.apache.spark.rdd._
@@ -34,8 +36,6 @@ import org.apache.spark.sql.execution.metric._
 import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
 import org.apache.spark.sql.vectorized._
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{DataTypeSupport, MetricsSupport}
 import org.apache.comet.iceberg.CometIcebergNativeScanMetadata
 
@@ -108,7 +108,7 @@ case class CometBatchScanExec(
   }
 
   override def hashCode(): Int = {
-    Objects.hashCode(wrappedScan, runtimeFilters)
+    Objects.hash(wrappedScan, runtimeFilters)
   }
 
   override def doCanonicalize(): CometBatchScanExec = {
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala
index 615f7bada26..8066f817004 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
 import java.util.UUID
 import java.util.concurrent.{Future, TimeoutException, TimeUnit}
 
@@ -42,8 +43,6 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.util.{SparkFatalException, ThreadUtils}
 import org.apache.spark.util.io.ChunkedByteBuffer
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.serde.OperatorOuterClass
 import org.apache.comet.serde.operator.CometSink
@@ -251,7 +250,7 @@ case class CometBroadcastExchangeExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(child)
+  override def hashCode(): Int = Objects.hash(child)
 
   override def stringArgs: Iterator[Any] = Iterator(output, child)
 
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometCoalesceExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometCoalesceExec.scala
index 7f947cdff2d..192bcc76930 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometCoalesceExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometCoalesceExec.scala
@@ -19,14 +19,14 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, SinglePartition, UnknownPartitioning}
 import org.apache.spark.sql.execution.{CoalesceExec, SparkPlan, UnaryExecNode}
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.serde.OperatorOuterClass
 import org.apache.comet.serde.operator.CometSink
@@ -85,5 +85,5 @@ case class CometCoalesceExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(numPartitions: java.lang.Integer, child)
+  override def hashCode(): Int = Objects.hash(numPartitions: java.lang.Integer, child)
 }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala
index 4d560ef8d3f..8b27274b6ce 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometCollectLimitExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.serializer.Serializer
 import org.apache.spark.sql.catalyst.InternalRow
@@ -29,8 +31,6 @@ import org.apache.spark.sql.execution.{CollectLimitExec, ColumnarToRowExec, Spar
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics, SQLShuffleReadMetricsReporter, SQLShuffleWriteMetricsReporter}
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.isCometShuffleEnabled
 import org.apache.comet.serde.{Compatible, OperatorOuterClass, SupportLevel, Unsupported}
@@ -136,5 +136,5 @@ case class CometCollectLimitExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(limit: java.lang.Integer, offset: java.lang.Integer, child)
+    Objects.hash(limit: java.lang.Integer, offset: java.lang.Integer, child)
 }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala
index dc92648b8d5..346b3a5d118 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometCsvNativeScanExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.catalyst.csv.CSVOptions
@@ -29,8 +31,6 @@ import org.apache.spark.sql.execution.datasources.FilePartition
 import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.objectstore.NativeConfig
 import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
@@ -71,7 +71,7 @@ case class CometCsvNativeScanExec(
   }
 
   override def hashCode(): Int = {
-    Objects.hashCode(output, serializedPlanOpt, originalPlan)
+    Objects.hash(output, serializedPlanOpt, originalPlan)
   }
 }
 
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala
index c944d3af8cd..36f53b3f22c 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.{Partition, TaskContext}
@@ -33,8 +35,6 @@ import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.util.AccumulatorV2
 
-import com.google.common.base.Objects
-
 import org.apache.comet.iceberg.CometIcebergNativeScanMetadata
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.operator.CometIcebergNativeScan
@@ -381,13 +381,13 @@ case class CometIcebergNativeScanExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       metadataLocation,
       output.asJava,
       serializedPlanOpt,
       runtimeFilters,
       // Ascribe java.lang.Integer: Scala 2.12 rejects the implicit Int -> Object conversion into
-      // the Object... varargs of Guava's Objects.hashCode.
+      // the Object... varargs of java.util.Objects.hash.
       scanHashCode: java.lang.Integer)
 }
 
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTableScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTableScanExec.scala
index 82d5e417c14..d5f01597503 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTableScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTableScanExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.collection.mutable.ListBuffer
 import scala.reflect.ClassTag
 
@@ -33,8 +35,6 @@ import org.apache.spark.sql.execution.{LeafExecNode, LocalTableScanExec}
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.types.{DataType, StructType}
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry, DataTypeSupport}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.OperatorOuterClass.Operator
@@ -120,7 +120,7 @@ case class CometLocalTableScanExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(originalPlan, originalPlan.schema, output)
+  override def hashCode(): Int = Objects.hash(originalPlan, originalPlan.schema, output)
 }
 
 object CometLocalTableScanExec extends CometSink[LocalTableScanExec] with DataTypeSupport {
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTopKExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTopKExec.scala
index ec3c85191a3..3fc9b540d30 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTopKExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometLocalTopKExec.scala
@@ -19,14 +19,14 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
 import org.apache.spark.sql.catalyst.plans.physical.Partitioning
 import org.apache.spark.sql.execution.{SparkPlan, TakeOrderedAndProjectExec}
 import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.types.{ByteType, IntegerType, LongType, ShortType}
 
-import com.google.common.base.Objects
-
 import org.apache.comet.CometConf
 import org.apache.comet.serde.OperatorOuterClass.{Operator, Sort}
 import org.apache.comet.serde.QueryPlanSerde.exprToProto
@@ -110,7 +110,7 @@ case class CometLocalTopKExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, Int.box(limit), sortOrder, Boolean.box(dynamicFilterEnabled), child)
+    Objects.hash(output, Int.box(limit), sortOrder, Boolean.box(dynamicFilterEnabled), child)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.baselineMetrics(sparkContext) ++ Map(
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala
index 8ea48c1002d..9e1c141e989 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometNativeScanExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.{Partition, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.SparkSession
@@ -35,8 +37,6 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 import org.apache.spark.util.SerializableConfiguration
 import org.apache.spark.util.collection._
 
-import com.google.common.base.Objects
-
 import org.apache.comet.parquet.CometParquetUtils
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.QueryPlanSerde.exprToProto
@@ -334,7 +334,7 @@ case class CometNativeScanExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(originalPlan, serializedPlanOpt, partitionFilters, dataFilters)
+    Objects.hash(originalPlan, serializedPlanOpt, partitionFilters, dataFilters)
 
   private val driverMetricKeys =
     Set(
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometRangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometRangeExec.scala
index 3cb049398ed..7435a57f3e7 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometRangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometRangeExec.scala
@@ -19,11 +19,11 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.execution.{RangeExec, SparkPlan}
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.serde.{CometOperatorSerde, Compatible, OperatorOuterClass, SupportLevel, Unsupported}
 import org.apache.comet.serde.OperatorOuterClass.Operator
@@ -66,7 +66,7 @@ case class CometRangeExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(originalPlan, output)
+  override def hashCode(): Int = Objects.hash(originalPlan, output)
 }
 
 object CometRangeExec extends CometOperatorSerde[RangeExec] {
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometSampleExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometSampleExec.scala
index dcdae2c779d..08140564a5a 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometSampleExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometSampleExec.scala
@@ -19,12 +19,12 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
 import org.apache.spark.sql.catalyst.plans.physical.Partitioning
 import org.apache.spark.sql.execution.{SampleExec, SparkPlan}
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass, SupportLevel, Unsupported}
@@ -119,7 +119,7 @@ case class CometSampleExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       lowerBound: java.lang.Double,
       upperBound: java.lang.Double,
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
index e7d3bd480fe..ba53f65a30e 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
@@ -19,6 +19,8 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Cast, CumeDist, CurrentRow, DenseRank, Divide, Expression, Lag, Lead, Literal, MakeDecimal, NamedExpression, NthValue, NTile, PercentRank, RangeFrame, Rank, RowFrame, RowNumber, SortOrder, SpecialFrameBoundary, SpecifiedWindowFrame, UnboundedFollowing, UnboundedPreceding, UnscaledValue, WindowExpression}
@@ -30,8 +32,6 @@ import org.apache.spark.sql.execution.window.WindowExec
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{ArrayType, DataType, DateType, Decimal, DecimalType, DoubleType, LongType, MapType, NumericType, StructType}
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.{AggSerde, CometOperatorSerde, CometSortOrder, LiteralOuterClass, OperatorOuterClass}
@@ -755,5 +755,5 @@ case class CometWindowExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, windowExpression, partitionSpec, orderSpec, child)
+    Objects.hash(output, windowExpression, partitionSpec, orderSpec, child)
 }
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowGroupLimitExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowGroupLimitExec.scala
index ee3cbd61c63..5e33553a21a 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowGroupLimitExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowGroupLimitExec.scala
@@ -19,14 +19,14 @@
 
 package org.apache.spark.sql.comet
 
+import java.util.Objects
+
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, SortOrder}
 import org.apache.spark.sql.catalyst.plans.physical.Partitioning
 import org.apache.spark.sql.execution.SparkPlan
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass}
@@ -190,7 +190,7 @@ case class CometWindowGroupLimitExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       partitionSpec,
       orderSpec,
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
index 3216b6a4301..68d3ebc3a98 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffleExchangeExec.scala
@@ -19,6 +19,7 @@
 
 package org.apache.spark.sql.comet.execution.shuffle
 
+import java.util.Objects
 import java.util.function.Supplier
 
 import scala.concurrent.Future
@@ -48,8 +49,6 @@ import org.apache.spark.util.MutablePair
 import org.apache.spark.util.collection.unsafe.sort.{PrefixComparators, RecordComparator}
 import org.apache.spark.util.random.XORShiftRandom
 
-import com.google.common.base.Objects
-
 import org.apache.comet.{CometConf, CometExplainInfo}
 import org.apache.comet.CometConf.{COMET_SHUFFLE_ENABLED, COMET_SHUFFLE_MODE}
 import org.apache.comet.CometSparkSessionExtensions.{cometCelebornShuffleFallbackReason, hasFallbackReason, isCometCelebornShuffleManagerEnabled, isCometShuffleManagerEnabled, isSpark40Plus, withFallbackReasons}
@@ -324,7 +323,7 @@ case class CometShuffleExchangeExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(outputPartitioning, shuffleOrigin, shuffleType, advisoryPartitionSize, child)
+    Objects.hash(outputPartitioning, shuffleOrigin, shuffleType, advisoryPartitionSize, child)
 
   override def stringArgs: Iterator[Any] =
     Iterator(outputPartitioning, shuffleOrigin, shuffleType, child) ++ Iterator(s"[plan_id=$id]")
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index fdb07c9acff..06abd4875d3 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -20,6 +20,7 @@
 package org.apache.spark.sql.comet
 
 import java.util.Locale
+import java.util.Objects
 
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
@@ -52,7 +53,6 @@ import org.apache.spark.unsafe.Platform
 import org.apache.spark.util.SerializableConfiguration
 import org.apache.spark.util.io.ChunkedByteBuffer
 
-import com.google.common.base.Objects
 import com.google.protobuf.CodedOutputStream
 
 import org.apache.comet.{CometConf, CometExecIterator, CometRuntimeException, ConfigEntry, ContribServices}
@@ -1430,7 +1430,7 @@ case class CometProjectExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, projectList, child)
+  override def hashCode(): Int = Objects.hash(output, projectList, child)
 
   override protected def outputExpressions: Seq[NamedExpression] = projectList
 }
@@ -1491,7 +1491,7 @@ case class CometFilterExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, condition, child)
+  override def hashCode(): Int = Objects.hash(output, condition, child)
 
   override def verboseStringWithOperatorId(): String = {
     s"""
@@ -1570,7 +1570,7 @@ case class CometSortExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, sortOrder, child)
+  override def hashCode(): Int = Objects.hash(output, sortOrder, child)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.baselineMetrics(sparkContext) ++
@@ -1639,7 +1639,7 @@ case class CometLocalLimitExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, limit: java.lang.Integer, child)
+  override def hashCode(): Int = Objects.hash(output, limit: java.lang.Integer, child)
 }
 
 object CometGlobalLimitExec extends CometOperatorSerde[GlobalLimitExec] {
@@ -1702,7 +1702,7 @@ case class CometGlobalLimitExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, limit: java.lang.Integer, offset: java.lang.Integer, child)
+    Objects.hash(output, limit: java.lang.Integer, offset: java.lang.Integer, child)
 }
 
 object CometExpandExec extends CometOperatorSerde[ExpandExec] {
@@ -1767,7 +1767,7 @@ case class CometExpandExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, projections, child)
+  override def hashCode(): Int = Objects.hash(output, projections, child)
 
   // TODO: support native Expand metrics
   override lazy val metrics: Map[String, SQLMetric] = Map.empty
@@ -1838,7 +1838,7 @@ case class CometMergeRowsExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       isSourceRowPresent,
       isTargetRowPresent,
@@ -1966,7 +1966,7 @@ case class CometExplodeExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, generator, generatorOutput, outer: java.lang.Boolean, child)
+    Objects.hash(output, generator, generatorOutput, outer: java.lang.Boolean, child)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.baselineMetrics(sparkContext) ++
@@ -2051,7 +2051,7 @@ case class CometUnionExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, children)
+  override def hashCode(): Int = Objects.hash(output, children)
 }
 
 /**
@@ -2685,7 +2685,7 @@ abstract class CometBaseAggregateExec
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       groupingExpressions,
       aggregateExpressions,
@@ -2902,7 +2902,7 @@ case class CometBroadcastNestedLoopJoinExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, joinType, condition, buildSide, left, right)
+    Objects.hash(output, joinType, condition, buildSide, left, right)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.joinMetrics(sparkContext)
@@ -3123,7 +3123,7 @@ case class CometHashJoinExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, leftKeys, rightKeys, joinType, condition, buildSide, left, right)
+    Objects.hash(output, leftKeys, rightKeys, joinType, condition, buildSide, left, right)
 
   override lazy val metrics: Map[String, SQLMetric] = {
     val joinMetrics = CometMetricNode.joinMetrics(sparkContext)
@@ -3278,7 +3278,7 @@ case class CometBroadcastHashJoinExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       leftKeys,
       rightKeys,
@@ -3483,7 +3483,7 @@ case class CometSortMergeJoinExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(output, leftKeys, rightKeys, joinType, condition, left, right)
+    Objects.hash(output, leftKeys, rightKeys, joinType, condition, left, right)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.sortMergeJoinMetrics(sparkContext)
diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
index 2a595398367..b8a40e266de 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -47,8 +47,6 @@ import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
-import com.google.common.primitives.UnsignedLong
-
 import org.apache.comet.CometConf
 import org.apache.comet.CometSparkSessionExtensions.{isSpark40Plus, isSpark41Plus}
 import org.apache.comet.vector.CometVector
@@ -1407,8 +1405,7 @@ abstract class ParquetReadSuite extends CometTestBase {
           checkAnswer(
             df,
             (0 until 10).map(n =>
-              Row(
-                new BigDecimal(UnsignedLong.fromLongBits(n + Long.MaxValue).bigIntegerValue()))))
+              Row(new BigDecimal(BigInteger.valueOf(n).add(BigInteger.valueOf(Long.MaxValue))))))
         }
       }
     }
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanEqualitySuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanEqualitySuite.scala
index 52a5e1fe953..fda69e10f5e 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanEqualitySuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/CometPlanEqualitySuite.scala
@@ -19,7 +19,11 @@
 
 package org.apache.spark.sql.comet
 
+import java.io.File
 import java.lang.reflect.Modifier
+import java.net.URL
+import java.nio.file.{Files, Paths}
+import java.util.jar.JarFile
 
 import scala.collection.JavaConverters._
 import scala.reflect.runtime.universe.runtimeMirror
@@ -29,8 +33,6 @@ import org.scalatest.funsuite.AnyFunSuite
 import org.apache.xbean.asm9.{ClassReader, Opcodes}
 import org.apache.xbean.asm9.tree.{ClassNode, FieldInsnNode, MethodInsnNode}
 
-import com.google.common.reflect.ClassPath
-
 import org.apache.comet.CometSparkSessionExtensions.isSpark42Plus
 import org.apache.comet.util.ClassLoaders
 
@@ -76,6 +78,48 @@ class CometPlanEqualitySuite extends AnyFunSuite {
       .map(_.name.decodedName.toString)
       .toSet
 
+  private def productionClasses(
+      loader: ClassLoader,
+      location: URL,
+      packageName: String): Seq[Class[_]] = {
+    val packagePath = packageName.replace('.', '/') + "/"
+    val root = Paths.get(location.toURI)
+    val classNames = if (Files.isDirectory(root)) {
+      val stream = Files.walk(root.resolve(packagePath))
+      try {
+        stream
+          .iterator()
+          .asScala
+          .filter(path => Files.isRegularFile(path) && path.toString.endsWith(".class"))
+          .map { path =>
+            root
+              .relativize(path)
+              .toString
+              .stripSuffix(".class")
+              .replace(File.separatorChar, '.')
+          }
+          .toSeq
+      } finally {
+        stream.close()
+      }
+    } else {
+      val jar = new JarFile(root.toFile)
+      try {
+        jar
+          .entries()
+          .asScala
+          .filter(entry =>
+            !entry.isDirectory && entry.getName.startsWith(packagePath) &&
+              entry.getName.endsWith(".class"))
+          .map(_.getName.stripSuffix(".class").replace('/', '.'))
+          .toSeq
+      } finally {
+        jar.close()
+      }
+    }
+    classNames.map(name => Class.forName(name, false, loader))
+  }
+
   /**
    * A structural guard: reads of constructor accessors/fields must occur in equals. This does not
    * prove those reads implement a correct comparison, nor detect semantics never captured as a
@@ -119,12 +163,7 @@ class CometPlanEqualitySuite extends AnyFunSuite {
     val loader = classOf[CometNativeExec].getClassLoader
     val productionLocation =
       classOf[CometNativeExec].getProtectionDomain.getCodeSource.getLocation
-    val classes = ClassPath
-      .from(loader)
-      .getAllClasses
-      .asScala
-      .filter(_.getName.startsWith("org.apache.spark.sql.comet."))
-      .map(_.load())
+    val classes = productionClasses(loader, productionLocation, "org.apache.spark.sql.comet")
       .filter { cls =>
         !Modifier.isAbstract(cls.getModifiers) &&
         cls.getProtectionDomain.getCodeSource.getLocation == productionLocation &&

From 255d7d9afb0089fcc936d21ee7ae4792ea7b1026 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Thu, 15 Oct 2026 16:24:52 -0600
Subject: [PATCH 2/3] fix: keep Guava removal compatible with latest main

---
 contrib/delta-spark/pom.xml                                    | 3 +++
 spark/src/main/java/org/apache/comet/udf/CometUdfBridge.java   | 3 ---
 .../test/scala/org/apache/comet/parquet/ParquetReadSuite.scala | 3 ++-
 3 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/contrib/delta-spark/pom.xml b/contrib/delta-spark/pom.xml
index 172a04ed27f..77c29895fc8 100644
--- a/contrib/delta-spark/pom.xml
+++ b/contrib/delta-spark/pom.xml
@@ -36,6 +36,9 @@
          jni.dir (basedir/../native) would point at contrib/native. -->
     <jni.dir>${project.basedir}/../../native/target/debug</jni.dir>
     <maven.deploy.skip>false</maven.deploy.skip>
+    <!-- Delta 4.x still requires Guava at test runtime. Keep the version local so the core Comet
+         modules do not expose or package Guava. -->
+    <guava.version>33.2.1-jre</guava.version>
   </properties>
 
   <dependencies>
diff --git a/spark/src/main/java/org/apache/comet/udf/CometUdfBridge.java b/spark/src/main/java/org/apache/comet/udf/CometUdfBridge.java
index 3ab34fe0496..4bf7e9c0aec 100644
--- a/spark/src/main/java/org/apache/comet/udf/CometUdfBridge.java
+++ b/spark/src/main/java/org/apache/comet/udf/CometUdfBridge.java
@@ -32,8 +32,6 @@
 import org.apache.spark.comet.CometTaskContextShim;
 import org.apache.spark.util.TaskCompletionListener;
 
-import com.google.common.annotations.VisibleForTesting;
-
 import org.apache.comet.util.ClassLoaders;
 
 /**
@@ -315,7 +313,6 @@ public static void releasePlan(long taskAttemptId, long planId) {
   }
 
   /** The instance of {@code udfClassName} that task {@code taskAttemptId} holds, or null. */
-  @VisibleForTesting
   public static CometUDF instanceFor(long taskAttemptId, String udfClassName) {
     ConcurrentHashMap<String, CometUDF> perTask = INSTANCES.get(taskAttemptId);
     return perTask == null ? null : perTask.get(udfClassName);
diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
index 20646219625..4f0f8d907c7 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -1407,7 +1407,8 @@ abstract class ParquetReadSuite extends CometTestBase {
           checkAnswer(
             df,
             (0 until 10).map(n =>
-              Row(new BigDecimal(BigInteger.valueOf(n).add(BigInteger.valueOf(Long.MaxValue))))))
+              Row(new BigDecimal(
+                BigInteger.valueOf(n.toLong).add(BigInteger.valueOf(Long.MaxValue))))))
         }
       }
     }

From 12275be605a40c2aa3b84f050d38c57f4f6cdbad Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Thu, 15 Oct 2026 17:17:35 -0600
Subject: [PATCH 3/3] Add Guava to Spark test runtime

---
 spark/pom.xml | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/spark/pom.xml b/spark/pom.xml
index ebd43c9fc4c..b1035c5fd7e 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -36,6 +36,9 @@ under the License.
   <properties>
     <!-- The only published module: the parent pom skips deploying every module -->
     <maven.deploy.skip>false</maven.deploy.skip>
+    <!-- Hive's test-only catalog integration still uses Guava at runtime. Keep the version local
+         so Guava is not exposed by or packaged into the published Comet artifact. -->
+    <guava.version>33.2.1-jre</guava.version>
   </properties>
 
   <dependencies>
@@ -123,6 +126,12 @@ under the License.
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>${guava.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.codehaus.jackson</groupId>
       <artifactId>jackson-mapper-asl</artifactId>