GH-43469: [Java] Change the default CompressionCodec.Factory to leverage compression support transparently

java/compression/src/main/java/module-info.java

@@ -15,11 +15,17 @@
  * limitations under the License.
  */
 
+import org.apache.arrow.vector.compression.CompressionCodec;
+
 module org.apache.arrow.compression {
   exports org.apache.arrow.compression;
 
   requires com.github.luben.zstd_jni;
   requires org.apache.arrow.memory.core;
   requires org.apache.arrow.vector;
   requires org.apache.commons.compress;
+
+  // Also defined under META-INF/services to support non-modular applications
+  provides CompressionCodec.Factory with
+      org.apache.arrow.compression.CommonsCompressionFactory;
 }

java/compression/src/main/resources/META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory

@@ -0,0 +1,15 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+org.apache.arrow.compression.CommonsCompressionFactory

java/compression/src/test/java/org/apache/arrow/compression/TestCompressionCodecServiceProvider.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.compression;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import org.apache.arrow.vector.compression.NoCompressionCodec;
+import org.junit.jupiter.api.Test;
+
+public class TestCompressionCodecServiceProvider {
+
+  /**
+   * When arrow-compression is in the classpath/module-path, {@link
+   * CompressionCodec.Factory#INSTANCE} should be able to handle all codec types.
+   */
+  @Test
+  public void testSupportedCompressionTypes() {
+    assertThrows( // no-compression doesn't support any actual compression types
+        IllegalArgumentException.class,
+        () -> checkAllCodecTypes(NoCompressionCodec.Factory.INSTANCE));
+    assertThrows( // commons-compression doesn't support the uncompressed type
+        IllegalArgumentException.class,
+        () -> checkAllCodecTypes(CommonsCompressionFactory.INSTANCE));
+    checkAllCodecTypes( // and the winner is...
+        CompressionCodec.Factory.INSTANCE); // combines the two above to support all types
+  }
+
+  private void checkAllCodecTypes(CompressionCodec.Factory factory) {
+    for (CompressionUtil.CodecType codecType : CompressionUtil.CodecType.values()) {
+      assertNotNull(factory.createCodec(codecType));
+    }
+  }
+}

java/vector/src/main/java/module-info.java

@@ -47,4 +47,6 @@
   requires org.apache.arrow.memory.core;
   requires org.apache.commons.codec;
   requires org.slf4j;
+
+  uses org.apache.arrow.vector.compression.CompressionCodec.Factory;
 }

java/vector/src/main/java/org/apache/arrow/vector/VectorLoader.java

@@ -50,7 +50,7 @@ public class VectorLoader {
    * @param root the root to add vectors to based on schema
    */
   public VectorLoader(VectorSchemaRoot root) {
-    this(root, NoCompressionCodec.Factory.INSTANCE);
+    this(root, CompressionCodec.Factory.INSTANCE);
   }
 
   /**

java/vector/src/main/java/org/apache/arrow/vector/compression/CompressionCodec.java

@@ -16,6 +16,9 @@
  */
 package org.apache.arrow.vector.compression;
 
+import java.util.EnumMap;
+import java.util.Map;
+import java.util.ServiceLoader;
 import org.apache.arrow.memory.ArrowBuf;
 import org.apache.arrow.memory.BufferAllocator;
 
@@ -51,11 +54,52 @@ public interface CompressionCodec {
 
   /** Factory to create compression codec. */
   interface Factory {
+    /**
+     * This combines all the available factories registered as service providers in the module path.
+     * For each {@link CompressionUtil.CodecType compression codec type}, it will use whatever
+     * factory supports it, i.e. doesn't throw on `createCodec(type)`. If multiple factories
+     * registered as service providers support the same codec type, the first one encountered while
+     * iterating over the {@link ServiceLoader} will be selected. A codec type that is not supported
+     * by any registered service provider will fall back to {@link
+     * NoCompressionCodec.Factory#INSTANCE} for backwards compatibility.
+     */
+    Factory INSTANCE = bestEffort();
 
     /** Creates the codec based on the codec type. */
     CompressionCodec createCodec(CompressionUtil.CodecType codecType);
 
     /** Creates the codec based on the codec type and compression level. */
     CompressionCodec createCodec(CompressionUtil.CodecType codecType, int compressionLevel);
+
+    private static Factory bestEffort() {
+      final ServiceLoader<Factory> serviceLoader = ServiceLoader.load(Factory.class);
+      final Map<CompressionUtil.CodecType, Factory> factories =
+          new EnumMap<>(CompressionUtil.CodecType.class);
+      for (Factory factory : serviceLoader) {
+        for (CompressionUtil.CodecType codecType : CompressionUtil.CodecType.values()) {
+          try {
+            factory.createCodec(codecType); // will throw if not supported
+            factories.putIfAbsent(codecType, factory);
+          } catch (Throwable ignored) {
+          }
+        }
+      }
+
+      final Factory fallback = NoCompressionCodec.Factory.INSTANCE;
+      return new Factory() {
+        @Override
+        public CompressionCodec createCodec(CompressionUtil.CodecType codecType) {
+          return factories.getOrDefault(codecType, fallback).createCodec(codecType);
+        }
+
+        @Override
+        public CompressionCodec createCodec(
+            CompressionUtil.CodecType codecType, int compressionLevel) {
+          return factories
+              .getOrDefault(codecType, fallback)
+              .createCodec(codecType, compressionLevel);
+        }
+      };
+    }
   }
 }

java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java

@@ -27,7 +27,6 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.util.VisibleForTesting;
 import org.apache.arrow.vector.compression.CompressionCodec;
-import org.apache.arrow.vector.compression.NoCompressionCodec;
 import org.apache.arrow.vector.ipc.message.ArrowBlock;
 import org.apache.arrow.vector.ipc.message.ArrowDictionaryBatch;
 import org.apache.arrow.vector.ipc.message.ArrowFooter;
@@ -64,7 +63,7 @@ public ArrowFileReader(
   }
 
   public ArrowFileReader(SeekableReadChannel in, BufferAllocator allocator) {
-    this(in, allocator, NoCompressionCodec.Factory.INSTANCE);
+    this(in, allocator, CompressionCodec.Factory.INSTANCE);
   }
 
   public ArrowFileReader(SeekableByteChannel in, BufferAllocator allocator) {

java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java

@@ -28,7 +28,6 @@
 import org.apache.arrow.vector.VectorLoader;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.compression.CompressionCodec;
-import org.apache.arrow.vector.compression.NoCompressionCodec;
 import org.apache.arrow.vector.dictionary.Dictionary;
 import org.apache.arrow.vector.dictionary.DictionaryProvider;
 import org.apache.arrow.vector.ipc.message.ArrowDictionaryBatch;
@@ -50,7 +49,7 @@ public abstract class ArrowReader implements DictionaryProvider, AutoCloseable {
   private final CompressionCodec.Factory compressionFactory;
 
   protected ArrowReader(BufferAllocator allocator) {
-    this(allocator, NoCompressionCodec.Factory.INSTANCE);
+    this(allocator, CompressionCodec.Factory.INSTANCE);
   }
 
   protected ArrowReader(BufferAllocator allocator, CompressionCodec.Factory compressionFactory) {

java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowStreamReader.java

@@ -25,7 +25,6 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.compression.CompressionCodec;
-import org.apache.arrow.vector.compression.NoCompressionCodec;
 import org.apache.arrow.vector.ipc.message.ArrowDictionaryBatch;
 import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
 import org.apache.arrow.vector.ipc.message.MessageChannelReader;
@@ -65,7 +64,7 @@ public ArrowStreamReader(
    * @param allocator to allocate new buffers
    */
   public ArrowStreamReader(MessageChannelReader messageReader, BufferAllocator allocator) {
-    this(messageReader, allocator, NoCompressionCodec.Factory.INSTANCE);
+    this(messageReader, allocator, CompressionCodec.Factory.INSTANCE);
   }
 
   /**