build: remove Guava dependency

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>

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
 }

pom.xml

@@ -95,7 +95,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
@@ -360,16 +359,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>
@@ -1392,15 +1381,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>

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>
@@ -126,23 +129,7 @@ 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>
+      <version>${guava.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -774,7 +761,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>
@@ -834,14 +820,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.

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);

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);
     }
   }
 

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;

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);
       }
     }
   }

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 = {