diff --git a/java/compression/src/main/java/module-info.java b/java/compression/src/main/java/module-info.java index 6bf989e4c142..113a1dba9d45 100644 --- a/java/compression/src/main/java/module-info.java +++ b/java/compression/src/main/java/module-info.java @@ -15,6 +15,8 @@ * limitations under the License. */ +import org.apache.arrow.vector.compression.CompressionCodec; + module org.apache.arrow.compression { exports org.apache.arrow.compression; @@ -22,4 +24,8 @@ 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; } diff --git a/java/compression/src/main/resources/META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory b/java/compression/src/main/resources/META-INF/services/org.apache.arrow.vector.compression.CompressionCodec$Factory new file mode 100644 index 000000000000..ccdcef9aed96 --- /dev/null +++ b/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 diff --git a/java/compression/src/test/java/org/apache/arrow/compression/TestCompressionCodecServiceProvider.java b/java/compression/src/test/java/org/apache/arrow/compression/TestCompressionCodecServiceProvider.java new file mode 100644 index 000000000000..795e05d7cb12 --- /dev/null +++ b/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)); + } + } +} diff --git a/java/vector/src/main/java/module-info.java b/java/vector/src/main/java/module-info.java index 73af2d1b67ef..fdea2bd06726 100644 --- a/java/vector/src/main/java/module-info.java +++ b/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; } diff --git a/java/vector/src/main/java/org/apache/arrow/vector/VectorLoader.java b/java/vector/src/main/java/org/apache/arrow/vector/VectorLoader.java index c076161bc21d..ecd3fb91241b 100644 --- a/java/vector/src/main/java/org/apache/arrow/vector/VectorLoader.java +++ b/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); } /** diff --git a/java/vector/src/main/java/org/apache/arrow/vector/compression/CompressionCodec.java b/java/vector/src/main/java/org/apache/arrow/vector/compression/CompressionCodec.java index 2de8ff246591..dd62108a84a6 100644 --- a/java/vector/src/main/java/org/apache/arrow/vector/compression/CompressionCodec.java +++ b/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 serviceLoader = ServiceLoader.load(Factory.class); + final Map 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); + } + }; + } } } diff --git a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java b/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java index 982651b2ff3d..7cac0a15a198 100644 --- a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowFileReader.java +++ b/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) { diff --git a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java b/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java index 15ade38cd3d6..7f4addf2d0de 100644 --- a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java +++ b/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) { diff --git a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowStreamReader.java b/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowStreamReader.java index 660c6a5f8986..69811dc71727 100644 --- a/java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowStreamReader.java +++ b/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); } /**