diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md
index b2beae1740b..18c336524a7 100644
--- a/docs/source/contributor-guide/adding_a_new_operator.md
+++ b/docs/source/contributor-guide/adding_a_new_operator.md
@@ -229,7 +229,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] {
 
@@ -287,14 +287,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] {
 
@@ -351,7 +351,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 93d99760ea1..afe4e46a1f0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -364,15 +364,15 @@ 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. -->
+      <!-- Comet does not use Guava itself. Hive's test-only catalog integration and Delta 4.x
+           still need it at test runtime, so it is managed here at test scope and neither
+           exposed by nor packaged into the published artifact. -->
       <dependency>
         <groupId>com.google.guava</groupId>
         <artifactId>guava</artifactId>
         <version>${guava.version}</version>
+        <scope>test</scope>
       </dependency>
-      <!-- End of shaded deps -->
 
       <!-- Test dependencies -->
       <dependency>
@@ -1412,15 +1412,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 e2c51bf3c5f..5aad564e26d 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -130,23 +130,6 @@ under the License.
     <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>
@@ -733,7 +716,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>
@@ -794,14 +776,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/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/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 96a7733dc06..da1438dbc0d 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;
 
@@ -370,36 +368,21 @@ 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);
+    // The channel is declared first so that it is closed even if opening the file fails, and it is
+    // closed after the input, as Spark's BypassMergeSortShuffleWriter does.
+    try (WritableByteChannelWrapper ignored = outputChannel;
+        FileInputStream in = new FileInputStream(file)) {
+      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 cbd9fa63792..59b00afae79 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 175c46a4b14..2878ec5515f 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}
 
@@ -44,8 +45,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
@@ -272,7 +271,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 ce6fc589093..4f40b66ca90 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.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.objectstore.NativeConfig
@@ -76,7 +76,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 e3466c71d27..91e4a3f9e76 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
@@ -408,13 +408,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 16ecb208819..d69ff6a5f5a 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
@@ -119,7 +119,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 8c8456b074a..3a274824b08 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
@@ -320,7 +320,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 54391580a44..50a2e878148 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
 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}
@@ -117,7 +117,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 3c3b2df03d5..bb80bd7f5b0 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, FrameType, 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}
@@ -847,5 +847,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 1242e549084..402a7120da7 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}
@@ -191,7 +191,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 3a1a431cbd2..a1825a19a6b 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.collection.mutable.ListBuffer
@@ -49,8 +50,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
 import org.apache.comet.CometConf.{COMET_SHUFFLE_ENABLED, COMET_SHUFFLE_MODE}
 import org.apache.comet.CometSparkSessionExtensions.{cometCelebornShuffleFallbackReason, hasFallbackReason, isCometCelebornShuffleManagerEnabled, isCometShuffleManagerEnabled, isSpark40Plus, withFallbackReasons}
@@ -334,7 +333,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 d07490ee914..0243a9ddd9b 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.annotation.tailrec
 import scala.collection.mutable
@@ -55,7 +56,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}
@@ -1499,7 +1499,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
 }
@@ -1564,7 +1564,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"""
@@ -1643,7 +1643,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) ++
@@ -1712,7 +1712,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] {
@@ -1775,7 +1775,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] {
@@ -1840,7 +1840,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
@@ -1911,7 +1911,7 @@ case class CometMergeRowsExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       isSourceRowPresent,
       isTargetRowPresent,
@@ -2039,7 +2039,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) ++
@@ -2124,7 +2124,7 @@ case class CometUnionExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, children)
+  override def hashCode(): Int = Objects.hash(output, children)
 }
 
 /**
@@ -2787,7 +2787,7 @@ abstract class CometBaseAggregateExec
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       groupingExpressions,
       aggregateExpressions,
@@ -3185,7 +3185,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)
@@ -3326,7 +3326,7 @@ case class CometCartesianProductExec(
     }
   }
 
-  override def hashCode(): Int = Objects.hashCode(output, left, right)
+  override def hashCode(): Int = Objects.hash(output, left, right)
 
   override lazy val metrics: Map[String, SQLMetric] =
     CometMetricNode.joinMetrics(sparkContext)
@@ -3480,7 +3480,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)
@@ -3635,7 +3635,7 @@ case class CometBroadcastHashJoinExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       leftKeys,
       rightKeys,
@@ -3864,7 +3864,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] = {
     val joinMetrics = CometMetricNode.sortMergeJoinMetrics(sparkContext)
diff --git a/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/CometArrowEvalPythonExec.scala b/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/CometArrowEvalPythonExec.scala
index ef7e8608cb2..afc629b0298 100644
--- a/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/CometArrowEvalPythonExec.scala
+++ b/spark/src/main/spark-4.1+/org/apache/spark/sql/comet/CometArrowEvalPythonExec.scala
@@ -20,6 +20,7 @@
 package org.apache.spark.sql.comet
 
 import java.nio.charset.StandardCharsets
+import java.util.Objects
 
 import scala.jdk.CollectionConverters._
 
@@ -31,7 +32,6 @@ import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
 import org.apache.spark.sql.execution.python.ArrowEvalPythonExec
 import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType, StringType, TimestampNTZType}
 
-import com.google.common.base.Objects
 import com.google.protobuf.ByteString
 
 import org.apache.comet.{CometConf, ConfigEntry, NativeBase}
@@ -243,7 +243,7 @@ case class CometArrowEvalPythonExec(
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(
+    Objects.hash(
       output,
       resultAttrs,
       udfs,
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 56743367a4a..763792225f5 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -50,8 +50,6 @@ import org.apache.spark.sql.functions.col
 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.{isSpark35Plus, isSpark40Plus, isSpark41Plus}
 import org.apache.comet.vector.CometVector
@@ -1579,8 +1577,8 @@ 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.toLong).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..995f6dc8013 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, '.')
+          }
+          .toVector
+      } 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('/', '.'))
+          .toVector
+      } 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 &&