Skip to content

HADOOP-17125. Using snappy-java in SnappyCodec #2297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
steveloughran merged 24 commits into from
Oct 6, 2020
Merged

HADOOP-17125. Using snappy-java in SnappyCodec #2297

Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
23da513
SnappyCodec with java-snappy
dbtsai Jul 6, 2020
3aa3f55
rebase master
dbtsai Aug 31, 2020
65fd9f4
Reset compressedDirectBuf and uncompressedDirectBuf.
viirya Sep 10, 2020
9c0f08b
Revert some debugging code.
viirya Sep 10, 2020
f52dd20
Remove snappy native code.
viirya Sep 10, 2020
87903a9
Remove snappy compilation.
viirya Sep 11, 2020
5adcf53
Fix limit parameter.
viirya Sep 11, 2020
40cc18a
Remove require.snappy.
viirya Sep 13, 2020
0e44d4b
trigger CI
viirya Sep 15, 2020
666a37b
Add compatibility test.
viirya Sep 15, 2020
9de4712
Check snappy library and remove useless code.
viirya Sep 17, 2020
0ed518d
Revert trailing whitespace.
viirya Sep 17, 2020
712749c
For review comment.
viirya Sep 17, 2020
2b5a212
Address review comments.
viirya Sep 24, 2020
1cb398b
Take safer approach.
viirya Sep 24, 2020
aa6a6d5
For review comments.
viirya Sep 25, 2020
19edba2
Update BUILDING and NativeLibraries.
viirya Sep 25, 2020
beec931
Merge remote-tracking branch 'upstream/trunk' into java-snappy
viirya Sep 25, 2020
fc525fa
Make snappy-java as compile scope.
viirya Sep 29, 2020
562b80d
For review comment.
viirya Sep 29, 2020
0600169
Revert Snappy description in BUILDING.txt.
viirya Sep 29, 2020
fd71a20
Fix style issue.
viirya Oct 1, 2020
7dc320d
Fix another style...
viirya Oct 2, 2020
5685b0b
Address comments: throwing AssertionError and exclude jobTokenPasswor…
viirya Oct 5, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 5 additions & 17 deletions hadoop-common-project/hadoop-common/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,11 @@
<artifactId>wildfly-openssl-java</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.xerial.snappy</groupId>
Comment thread
steveloughran marked this conversation as resolved.
<artifactId>snappy-java</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>

<build>
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Expand Down
29 changes: 0 additions & 29 deletions hadoop-common-project/hadoop-common/src/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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}
Expand All @@ -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
Expand Down
2 changes: 1 addition & 1 deletion ...ommon-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/Decompressor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
40 changes: 1 addition & 39 deletions ...common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/SnappyCodec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -56,37 +55,6 @@ public Configuration getConf() {
return conf;
}

/**
* Are the native snappy libraries loaded &amp; 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}.
Expand Down Expand Up @@ -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);
Expand All @@ -133,7 +100,6 @@ public CompressionOutputStream createOutputStream(OutputStream out,
*/
@Override
public Class<? extends Compressor> getCompressorType() {
checkNativeCodeLoaded();
return SnappyCompressor.class;
}

Expand All @@ -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);
Expand Down Expand Up @@ -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));
Expand All @@ -192,7 +156,6 @@ public CompressionInputStream createInputStream(InputStream in,
*/
@Override
public Class<? extends Decompressor> getDecompressorType() {
checkNativeCodeLoaded();
return SnappyDecompressor.class;
}

Expand All @@ -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);
Expand All @@ -215,7 +177,7 @@ public Decompressor createDecompressor() {
*/
@Override
public DirectDecompressor createDirectDecompressor() {
return isNativeCodeLoaded() ? new SnappyDirectDecompressor() : null;
return new SnappyDirectDecompressor();
}

/**
Expand Down
50 changes: 25 additions & 25 deletions ...ct/hadoop-common/src/main/java/org/apache/hadoop/io/compress/snappy/SnappyCompressor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@

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;
import org.xerial.snappy.SnappyLoader;

/**
* A {@link Compressor} based on the snappy compression algorithm.
Expand All @@ -48,30 +49,21 @@ 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;

@dbtsai dbtsai Sep 17, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we need to check if the snappy class is available for SnappyCompressor too.

@viirya viirya Sep 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added.

}

/**
* Creates a new compressor.
*
* @param directBufferSize size of the direct buffer to be used.
*/
public SnappyCompressor(int directBufferSize) {
// `snappy-java` is provided scope. We need to check if it is available.
try {
SnappyLoader.getVersion();
} catch (Throwable t) {
throw new RuntimeException("snappy-java library is not available: " +
"SnappyCompressor has not been loaded. You need to add " +
"snappy-java.jar to your CLASSPATH", t);
}

this.directBufferSize = directBufferSize;

uncompressedDirectBuf = ByteBuffer.allocateDirect(directBufferSize);
Expand Down Expand Up @@ -225,7 +217,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

Expand Down Expand Up @@ -291,9 +283,17 @@ 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;
uncompressedDirectBuf.limit(uncompressedDirectBuf.capacity()).position(0);

@sunchao sunchao Sep 25, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: this seems unnecessary as clear is called shortly after at the call site?

@viirya viirya Sep 25, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems so, I remember I added this to fix test failure. It might be SnappyDecompressor, I think, then I copied to SnappyCompressor. Deleted this and see what Jenkins tells.

return size;
}
}
}
Loading