This is an automated email from the ASF dual-hosted git repository.
dkulp pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/master by this push:
new d81bffd AVRO-2355 [java] Add level and checksum options to ZStandard
compression
d81bffd is described below
commit d81bffd41c348d8ffe11506fe26c66c58a37a580
Author: Scott Carey <[email protected]>
AuthorDate: Wed Mar 20 17:19:43 2019 -0700
AVRO-2355 [java] Add level and checksum options to ZStandard compression
---
.../java/org/apache/avro/file/CodecFactory.java | 29 +++++++++++++--
.../java/org/apache/avro/file/ZstandardCodec.java | 35 ++++++++++++++----
.../java/org/apache/avro/file/ZstandardLoader.java | 43 ++++++++++++++++++++++
.../test/java/org/apache/avro/TestDataFile.java | 5 ++-
.../org/apache/avro/file/TestZstandardCodec.java | 35 ++++++++++++++++++
lang/java/mapred/pom.xml | 1 -
.../avro/mapreduce/TestAvroKeyOutputFormat.java | 2 +-
7 files changed, 135 insertions(+), 15 deletions(-)
diff --git
a/lang/java/avro/src/main/java/org/apache/avro/file/CodecFactory.java
b/lang/java/avro/src/main/java/org/apache/avro/file/CodecFactory.java
index 25fa669..da1442f 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/file/CodecFactory.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/file/CodecFactory.java
@@ -74,9 +74,29 @@ public abstract class CodecFactory {
return new BZip2Codec.Option();
}
- /** zstandard codec. */
- public static CodecFactory zstandardCodec() {
- return new ZstandardCodec.Option();
+ /**
+ * zstandard codec, with specific compression level.
+ *
+ * @param level The compression level should be between -5 and 22, inclusive.
+ * Negative levels are 'fast' modes akin to lz4 or snappy,
levels
+ * above 9 are generally for archival purposes, and levels
above 18
+ * use a lot of memory.
+ */
+ public static CodecFactory zstandardCodec(int level) {
+ return new ZstandardCodec.Option(level, false);
+ }
+
+ /**
+ * zstandard codec, with specific compression level.
+ *
+ * @param level The compression level should be between -5 and 22,
+ * inclusive. Negative levels are 'fast' modes akin to
lz4 or
+ * snappy, levels above 9 are generally for archival
+ * purposes, and levels above 18 use a lot of memory.
+ * @param useChecksum if true, will include a checksum with each data block
+ */
+ public static CodecFactory zstandardCodec(int level, boolean useChecksum) {
+ return new ZstandardCodec.Option(level, useChecksum);
}
/** Creates internal Codec. */
@@ -90,13 +110,14 @@ public abstract class CodecFactory {
public static final int DEFAULT_DEFLATE_LEVEL = Deflater.DEFAULT_COMPRESSION;
public static final int DEFAULT_XZ_LEVEL = XZCodec.DEFAULT_COMPRESSION;
+ public static final int DEFAULT_ZSTANDARD_LEVEL = 3;
static {
addCodec(DataFileConstants.NULL_CODEC, nullCodec());
addCodec(DataFileConstants.DEFLATE_CODEC,
deflateCodec(DEFAULT_DEFLATE_LEVEL));
addCodec(DataFileConstants.BZIP2_CODEC, bzip2Codec());
addCodec(DataFileConstants.XZ_CODEC, xzCodec(DEFAULT_XZ_LEVEL));
- addCodec(DataFileConstants.ZSTANDARD_CODEC, zstandardCodec());
+ addCodec(DataFileConstants.ZSTANDARD_CODEC,
zstandardCodec(DEFAULT_ZSTANDARD_LEVEL));
addCodec(DataFileConstants.SNAPPY_CODEC, snappyCodec());
}
diff --git
a/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardCodec.java
b/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardCodec.java
index 90b8681..95e803d 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardCodec.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardCodec.java
@@ -24,22 +24,38 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
-import
org.apache.commons.compress.compressors.zstandard.ZstdCompressorInputStream;
-import
org.apache.commons.compress.compressors.zstandard.ZstdCompressorOutputStream;
import org.apache.commons.compress.utils.IOUtils;
public class ZstandardCodec extends Codec {
static class Option extends CodecFactory {
+ private final int compressionLevel;
+ private final boolean useChecksum;
+
+ Option(int compressionLevel, boolean useChecksum) {
+ this.compressionLevel = compressionLevel;
+ this.useChecksum = useChecksum;
+ }
@Override
protected Codec createInstance() {
- return new ZstandardCodec();
+ return new ZstandardCodec(compressionLevel, useChecksum);
}
}
+ private final int compressionLevel;
+ private final boolean useChecksum;
private ByteArrayOutputStream outputBuffer;
+ /**
+ * Create a ZstandardCodec instance with the given compressionLevel and
checksum
+ * option
+ **/
+ public ZstandardCodec(int compressionLevel, boolean useChecksum) {
+ this.compressionLevel = compressionLevel;
+ this.useChecksum = useChecksum;
+ }
+
@Override
public String getName() {
return DataFileConstants.ZSTANDARD_CODEC;
@@ -48,7 +64,7 @@ public class ZstandardCodec extends Codec {
@Override
public ByteBuffer compress(ByteBuffer data) throws IOException {
ByteArrayOutputStream baos = getOutputBuffer(data.remaining());
- try (OutputStream outputStream = new ZstdCompressorOutputStream(baos)) {
+ try (OutputStream outputStream = ZstandardLoader.output(baos,
compressionLevel, useChecksum)) {
outputStream.write(data.array(), computeOffset(data), data.remaining());
}
return ByteBuffer.wrap(baos.toByteArray());
@@ -59,7 +75,7 @@ public class ZstandardCodec extends Codec {
ByteArrayOutputStream baos = getOutputBuffer(compressedData.remaining());
InputStream bytesIn = new ByteArrayInputStream(compressedData.array(),
computeOffset(compressedData),
compressedData.remaining());
- try (InputStream ios = new ZstdCompressorInputStream(bytesIn)) {
+ try (InputStream ios = ZstandardLoader.input(bytesIn)) {
IOUtils.copy(ios, baos);
}
return ByteBuffer.wrap(baos.toByteArray());
@@ -81,8 +97,11 @@ public class ZstandardCodec extends Codec {
@Override
public boolean equals(Object obj) {
- if (this == obj)
- return true;
- return obj != null && obj.getClass() == getClass();
+ return (this == obj) || (obj != null && obj.getClass() == this.getClass());
+ }
+
+ @Override
+ public String toString() {
+ return getName() + "[" + compressionLevel + "]";
}
}
diff --git
a/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardLoader.java
b/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardLoader.java
new file mode 100644
index 0000000..9a9599e
--- /dev/null
+++ b/lang/java/avro/src/main/java/org/apache/avro/file/ZstandardLoader.java
@@ -0,0 +1,43 @@
+/*
+ * 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.avro.file;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.github.luben.zstd.Zstd;
+import com.github.luben.zstd.ZstdInputStream;
+import com.github.luben.zstd.ZstdOutputStream;
+
+/* causes lazier classloader initialization of ZStandard libraries, so that
+ * we get NoClassDefFoundError when we try and use the Codec's compress
+ * or decompress methods rather than when we instantiate it */
+final class ZstandardLoader {
+
+ static InputStream input(InputStream compressed) throws IOException {
+ return new ZstdInputStream(compressed);
+ }
+
+ static OutputStream output(OutputStream compressed, int level, boolean
checksum) throws IOException {
+
+ int bounded = Math.max(Math.min(level, Zstd.maxCompressionLevel()),
Zstd.minCompressionLevel());
+
+ return new ZstdOutputStream(compressed, bounded, false, checksum);
+ }
+}
diff --git a/lang/java/avro/src/test/java/org/apache/avro/TestDataFile.java
b/lang/java/avro/src/test/java/org/apache/avro/TestDataFile.java
index 9803474..d3957e1 100644
--- a/lang/java/avro/src/test/java/org/apache/avro/TestDataFile.java
+++ b/lang/java/avro/src/test/java/org/apache/avro/TestDataFile.java
@@ -70,7 +70,10 @@ public class TestDataFile {
r.add(new Object[] { CodecFactory.xzCodec(0) });
r.add(new Object[] { CodecFactory.xzCodec(1) });
r.add(new Object[] { CodecFactory.xzCodec(6) });
- r.add(new Object[] { CodecFactory.zstandardCodec() });
+ r.add(new Object[] { CodecFactory.zstandardCodec(-5) });
+ r.add(new Object[] { CodecFactory.zstandardCodec(0, true) });
+ r.add(new Object[] { CodecFactory.zstandardCodec(5, false) });
+ r.add(new Object[] { CodecFactory.zstandardCodec(18, true) });
return r;
}
diff --git
a/lang/java/avro/src/test/java/org/apache/avro/file/TestZstandardCodec.java
b/lang/java/avro/src/test/java/org/apache/avro/file/TestZstandardCodec.java
new file mode 100644
index 0000000..2644ee0
--- /dev/null
+++ b/lang/java/avro/src/test/java/org/apache/avro/file/TestZstandardCodec.java
@@ -0,0 +1,35 @@
+/*
+ * 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.avro.file;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+public class TestZstandardCodec {
+
+ @Test
+ public void testZstandardToStringAndName() throws IOException {
+ Codec codec = CodecFactory.zstandardCodec(3).createInstance();
+ assertTrue(codec instanceof ZstandardCodec);
+ assertTrue(codec.getName().equals("zstandard"));
+ assertTrue(codec.toString().equals("zstandard[3]"));
+ }
+}
diff --git a/lang/java/mapred/pom.xml b/lang/java/mapred/pom.xml
index c826dec..03f3fc9 100644
--- a/lang/java/mapred/pom.xml
+++ b/lang/java/mapred/pom.xml
@@ -190,7 +190,6 @@
<artifactId>snappy-java</artifactId>
<scope>test</scope>
</dependency>
-
</dependencies>
</project>
diff --git
a/lang/java/mapred/src/test/java/org/apache/avro/mapreduce/TestAvroKeyOutputFormat.java
b/lang/java/mapred/src/test/java/org/apache/avro/mapreduce/TestAvroKeyOutputFormat.java
index e4f9cc9..4d7e2ab 100644
---
a/lang/java/mapred/src/test/java/org/apache/avro/mapreduce/TestAvroKeyOutputFormat.java
+++
b/lang/java/mapred/src/test/java/org/apache/avro/mapreduce/TestAvroKeyOutputFormat.java
@@ -84,7 +84,7 @@ public class TestAvroKeyOutputFormat {
Configuration conf = new Configuration();
conf.setBoolean("mapred.output.compress", true);
conf.set(AvroJob.CONF_OUTPUT_CODEC, DataFileConstants.ZSTANDARD_CODEC);
- testGetRecordWriter(conf, CodecFactory.zstandardCodec(),
DataFileConstants.DEFAULT_SYNC_INTERVAL);
+ testGetRecordWriter(conf, CodecFactory.zstandardCodec(3),
DataFileConstants.DEFAULT_SYNC_INTERVAL);
}
@Test
