HADOOP-17125. Using snappy-java in SnappyCodec

BUILDING.txt

@@ -82,7 +82,7 @@ Installing required packages for clean install of Ubuntu 14.04 LTS Desktop:
 
 Optional packages:
 
-* Snappy compression
+* Snappy compression (only used for hadoop-mapreduce-client-nativetask)
   $ sudo apt-get install snappy libsnappy-dev
 * Intel ISA-L library for erasure coding
   Please refer to https://01.org/intel%C2%AE-storage-acceleration-library-open-source-version
@@ -168,7 +168,8 @@ Maven build goals:
 
    Snappy is a compression library that can be utilized by the native code.
    It is currently an optional component, meaning that Hadoop can be built with
-   or without this dependency.
+   or without this dependency. Snappy library as optional dependency is only
+   used for hadoop-mapreduce-client-nativetask.
 
   * Use -Drequire.snappy to fail the build if libsnappy.so is not found.
     If this option is not specified and the snappy library is missing,

dev-support/bin/dist-copynativelibs

@@ -111,9 +111,6 @@ for i in "$@"; do
     --openssllibbundle=*)
       OPENSSLLIBBUNDLE=${i#*=}
     ;;
-    --snappybinbundle=*)
-      SNAPPYBINBUNDLE=${i#*=}
-    ;;
     --snappylib=*)
       SNAPPYLIB=${i#*=}
     ;;
@@ -176,8 +173,6 @@ if [[ -d "${BIN_DIR}" ]] ; then
     exit 1
   fi
 
-  bundle_native_bin "${SNAPPYBINBUNDLE}" "${SNAPPYLIBBUNDLE}" "snappy.lib" "snappy" "${SNAPPYLIB}"
-
   bundle_native_bin "${ZSTDBINBUNDLE}" "${ZSTDLIBBUNDLE}" "zstd.lib" "zstd" "${ZSTDLIB}"
 
   bundle_native_bin "${OPENSSLBINBUNDLE}" "${OPENSSLLIBBUNDLE}" "openssl.lib" "crypto" "${OPENSSLLIB}"

hadoop-common-project/hadoop-common/pom.xml

@@ -363,6 +363,11 @@
       <artifactId>wildfly-openssl-java</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.xerial.snappy</groupId>
+      <artifactId>snappy-java</artifactId>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -642,10 +647,6 @@
       </activation>
       <properties>
         <require.bzip2>false</require.bzip2>
-        <snappy.prefix></snappy.prefix>
-        <snappy.lib></snappy.lib>
-        <snappy.include></snappy.include>
-        <require.snappy>false</require.snappy>
         <zstd.prefix></zstd.prefix>
         <zstd.lib></zstd.lib>
         <zstd.include></zstd.include>
@@ -699,11 +700,7 @@
                     <GENERATED_JAVAH>${project.build.directory}/native/javah</GENERATED_JAVAH>
                     <JVM_ARCH_DATA_MODEL>${sun.arch.data.model}</JVM_ARCH_DATA_MODEL>
                     <REQUIRE_BZIP2>${require.bzip2}</REQUIRE_BZIP2>
-                    <REQUIRE_SNAPPY>${require.snappy}</REQUIRE_SNAPPY>
                     <REQUIRE_ZSTD>${require.zstd}</REQUIRE_ZSTD>
-                    <CUSTOM_SNAPPY_PREFIX>${snappy.prefix}</CUSTOM_SNAPPY_PREFIX>
-                    <CUSTOM_SNAPPY_LIB>${snappy.lib} </CUSTOM_SNAPPY_LIB>
-                    <CUSTOM_SNAPPY_INCLUDE>${snappy.include} </CUSTOM_SNAPPY_INCLUDE>
                     <CUSTOM_ZSTD_PREFIX>${zstd.prefix}</CUSTOM_ZSTD_PREFIX>
                     <CUSTOM_ZSTD_LIB>${zstd.lib} </CUSTOM_ZSTD_LIB>
                     <CUSTOM_ZSTD_INCLUDE>${zstd.include} </CUSTOM_ZSTD_INCLUDE>
@@ -758,14 +755,9 @@
         </os>
       </activation>
       <properties>
-        <snappy.prefix></snappy.prefix>
-        <snappy.lib></snappy.lib>
-        <snappy.include></snappy.include>
         <require.isal>false</require.isal>
         <isal.prefix></isal.prefix>
         <isal.lib></isal.lib>
-        <require.snappy>false</require.snappy>
-        <bundle.snappy.in.bin>true</bundle.snappy.in.bin>
         <zstd.prefix></zstd.prefix>
         <zstd.lib></zstd.lib>
         <zstd.include></zstd.include>
@@ -865,10 +857,6 @@
                     <argument>/nologo</argument>
                     <argument>/p:Configuration=Release</argument>
                     <argument>/p:OutDir=${project.build.directory}/bin/</argument>
-                    <argument>/p:CustomSnappyPrefix=${snappy.prefix}</argument>
-                    <argument>/p:CustomSnappyLib=${snappy.lib}</argument>
-                    <argument>/p:CustomSnappyInclude=${snappy.include}</argument>
-                    <argument>/p:RequireSnappy=${require.snappy}</argument>
                     <argument>/p:CustomZstdPrefix=${zstd.prefix}</argument>
                     <argument>/p:CustomZstdLib=${zstd.lib}</argument>
                     <argument>/p:CustomZstdInclude=${zstd.include}</argument>

hadoop-common-project/hadoop-common/src/CMakeLists.txt

@@ -67,33 +67,6 @@ else()
 endif()
 set(CMAKE_FIND_LIBRARY_SUFFIXES ${STORED_CMAKE_FIND_LIBRARY_SUFFIXES})
 
-# Require snappy.
-set(STORED_CMAKE_FIND_LIBRARY_SUFFIXES ${CMAKE_FIND_LIBRARY_SUFFIXES})
-hadoop_set_find_shared_library_version("1")
-find_library(SNAPPY_LIBRARY
-    NAMES snappy
-    PATHS ${CUSTOM_SNAPPY_PREFIX} ${CUSTOM_SNAPPY_PREFIX}/lib
-          ${CUSTOM_SNAPPY_PREFIX}/lib64 ${CUSTOM_SNAPPY_LIB})
-set(CMAKE_FIND_LIBRARY_SUFFIXES ${STORED_CMAKE_FIND_LIBRARY_SUFFIXES})
-find_path(SNAPPY_INCLUDE_DIR
-    NAMES snappy.h
-    PATHS ${CUSTOM_SNAPPY_PREFIX} ${CUSTOM_SNAPPY_PREFIX}/include
-          ${CUSTOM_SNAPPY_INCLUDE})
-if(SNAPPY_LIBRARY AND SNAPPY_INCLUDE_DIR)
-    get_filename_component(HADOOP_SNAPPY_LIBRARY ${SNAPPY_LIBRARY} NAME)
-    set(SNAPPY_SOURCE_FILES
-        "${SRC}/io/compress/snappy/SnappyCompressor.c"
-        "${SRC}/io/compress/snappy/SnappyDecompressor.c")
-    set(REQUIRE_SNAPPY ${REQUIRE_SNAPPY}) # Stop warning about unused variable.
-    message(STATUS "Found Snappy: ${SNAPPY_LIBRARY}")
-else()
-    set(SNAPPY_INCLUDE_DIR "")
-    set(SNAPPY_SOURCE_FILES "")
-    if(REQUIRE_SNAPPY)
-        message(FATAL_ERROR "Required snappy library could not be found.  SNAPPY_LIBRARY=${SNAPPY_LIBRARY}, SNAPPY_INCLUDE_DIR=${SNAPPY_INCLUDE_DIR}, CUSTOM_SNAPPY_INCLUDE_DIR=${CUSTOM_SNAPPY_INCLUDE_DIR}, CUSTOM_SNAPPY_PREFIX=${CUSTOM_SNAPPY_PREFIX}, CUSTOM_SNAPPY_INCLUDE=${CUSTOM_SNAPPY_INCLUDE}")
-    endif()
-endif()
-
 # Require zstandard
 SET(STORED_CMAKE_FIND_LIBRARY_SUFFIXES ${CMAKE_FIND_LIBRARY_SUFFIXES})
 hadoop_set_find_shared_library_version("1")
@@ -253,7 +226,6 @@ include_directories(
     ${JNI_INCLUDE_DIRS}
     ${ZLIB_INCLUDE_DIRS}
     ${BZIP2_INCLUDE_DIR}
-    ${SNAPPY_INCLUDE_DIR}
     ${ISAL_INCLUDE_DIR}
     ${ZSTD_INCLUDE_DIR}
     ${OPENSSL_INCLUDE_DIR}
@@ -269,7 +241,6 @@ hadoop_add_dual_library(hadoop
     ${SRC}/io/compress/lz4/lz4.c
     ${SRC}/io/compress/lz4/lz4hc.c
     ${ISAL_SOURCE_FILES}
-    ${SNAPPY_SOURCE_FILES}
     ${ZSTD_SOURCE_FILES}
     ${OPENSSL_SOURCE_FILES}
     ${SRC}/io/compress/zlib/ZlibCompressor.c

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/Decompressor.java

@@ -92,7 +92,7 @@ public interface Decompressor {
    * {@link #needsInput()} should be called in order to determine if more
    * input data is required.
    * 
-   * @param b Buffer for the compressed data
+   * @param b Buffer for the uncompressed data
    * @param off Start offset of the data
    * @param len Size of the buffer
    * @return The actual number of bytes of uncompressed data.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/SnappyCodec.java

@@ -28,7 +28,6 @@
 import org.apache.hadoop.io.compress.snappy.SnappyDecompressor;
 import org.apache.hadoop.io.compress.snappy.SnappyDecompressor.SnappyDirectDecompressor;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.hadoop.util.NativeCodeLoader;
 
 /**
  * This class creates snappy compressors/decompressors.
@@ -56,37 +55,6 @@ public Configuration getConf() {
     return conf;
   }
 
-  /**
-   * Are the native snappy libraries loaded & initialized?
-   */
-  public static void checkNativeCodeLoaded() {
-    if (!NativeCodeLoader.buildSupportsSnappy()) {
-      throw new RuntimeException("native snappy library not available: " +
-          "this version of libhadoop was built without " +
-          "snappy support.");
-    }
-    if (!NativeCodeLoader.isNativeCodeLoaded()) {
-      throw new RuntimeException("Failed to load libhadoop.");
-    }
-    if (!SnappyCompressor.isNativeCodeLoaded()) {
-      throw new RuntimeException("native snappy library not available: " +
-          "SnappyCompressor has not been loaded.");
-    }
-    if (!SnappyDecompressor.isNativeCodeLoaded()) {
-      throw new RuntimeException("native snappy library not available: " +
-          "SnappyDecompressor has not been loaded.");
-    }
-  }
-
-  public static boolean isNativeCodeLoaded() {
-    return SnappyCompressor.isNativeCodeLoaded() && 
-        SnappyDecompressor.isNativeCodeLoaded();
-  }
-
-  public static String getLibraryName() {
-    return SnappyCompressor.getLibraryName();
-  }
-
   /**
    * Create a {@link CompressionOutputStream} that will write to the given
    * {@link OutputStream}.
@@ -115,7 +83,6 @@ public CompressionOutputStream createOutputStream(OutputStream out)
   public CompressionOutputStream createOutputStream(OutputStream out,
                                                     Compressor compressor)
       throws IOException {
-    checkNativeCodeLoaded();
     int bufferSize = conf.getInt(
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_DEFAULT);
@@ -133,7 +100,6 @@ public CompressionOutputStream createOutputStream(OutputStream out,
    */
   @Override
   public Class<? extends Compressor> getCompressorType() {
-    checkNativeCodeLoaded();
     return SnappyCompressor.class;
   }
 
@@ -144,7 +110,6 @@ public Class<? extends Compressor> getCompressorType() {
    */
   @Override
   public Compressor createCompressor() {
-    checkNativeCodeLoaded();
     int bufferSize = conf.getInt(
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_DEFAULT);
@@ -179,7 +144,6 @@ public CompressionInputStream createInputStream(InputStream in)
   public CompressionInputStream createInputStream(InputStream in,
                                                   Decompressor decompressor)
       throws IOException {
-    checkNativeCodeLoaded();
     return new BlockDecompressorStream(in, decompressor, conf.getInt(
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_DEFAULT));
@@ -192,7 +156,6 @@ public CompressionInputStream createInputStream(InputStream in,
    */
   @Override
   public Class<? extends Decompressor> getDecompressorType() {
-    checkNativeCodeLoaded();
     return SnappyDecompressor.class;
   }
 
@@ -203,7 +166,6 @@ public Class<? extends Decompressor> getDecompressorType() {
    */
   @Override
   public Decompressor createDecompressor() {
-    checkNativeCodeLoaded();
     int bufferSize = conf.getInt(
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_KEY,
         CommonConfigurationKeys.IO_COMPRESSION_CODEC_SNAPPY_BUFFERSIZE_DEFAULT);
@@ -215,7 +177,7 @@ public Decompressor createDecompressor() {
    */
   @Override
   public DirectDecompressor createDirectDecompressor() {
-    return isNativeCodeLoaded() ? new SnappyDirectDecompressor() : null;
+    return new SnappyDirectDecompressor();
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/snappy/SnappyCompressor.java

@@ -24,9 +24,9 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.compress.Compressor;
-import org.apache.hadoop.util.NativeCodeLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.xerial.snappy.Snappy;
 
 /**
  * A {@link Compressor} based on the snappy compression algorithm.
@@ -48,24 +48,6 @@ public class SnappyCompressor implements Compressor {
   private long bytesRead = 0L;
   private long bytesWritten = 0L;
 
-  private static boolean nativeSnappyLoaded = false;
-
-  static {
-    if (NativeCodeLoader.isNativeCodeLoaded() &&
-        NativeCodeLoader.buildSupportsSnappy()) {
-      try {
-        initIDs();
-        nativeSnappyLoaded = true;
-      } catch (Throwable t) {
-        LOG.error("failed to load SnappyCompressor", t);
-      }
-    }
-  }
-
-  public static boolean isNativeCodeLoaded() {
-    return nativeSnappyLoaded;
-  }
-
   /**
    * Creates a new compressor.
    *
@@ -225,7 +207,7 @@ public int compress(byte[] b, int off, int len)
     }
 
     // Compress data
-    n = compressBytesDirect();
+    n = compressDirectBuf();
     compressedDirectBuf.limit(n);
     uncompressedDirectBuf.clear(); // snappy consumes all buffer input
 
@@ -291,9 +273,16 @@ public long getBytesWritten() {
   public void end() {
   }
 
-  private native static void initIDs();
-
-  private native int compressBytesDirect();
-
-  public native static String getLibraryName();
+  private int compressDirectBuf() throws IOException {
+    if (uncompressedDirectBufLen == 0) {
+      return 0;
+    } else {
+      // Set the position and limit of `uncompressedDirectBuf` for reading
+      uncompressedDirectBuf.limit(uncompressedDirectBufLen).position(0);
+      int size = Snappy.compress((ByteBuffer) uncompressedDirectBuf,
+              (ByteBuffer) compressedDirectBuf);
+      uncompressedDirectBufLen = 0;
+      return size;
+    }
+  }
 }