add compression awareness to HdfsResources

SHDP-86
as long as the compression contains codecs, the inputstream of
compressed files is automatically decompressed. further more, this
behaviour can be disabled if needed.

Author: Costin Leau

src/main/java/org/springframework/data/hadoop/fs/HdfsResource.java

@@ -27,6 +27,10 @@
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.compress.CodecPool;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.Decompressor;
 import org.springframework.core.io.ContextResource;
 import org.springframework.core.io.Resource;
 import org.springframework.util.Assert;
@@ -45,16 +49,17 @@ class HdfsResource implements ContextResource {
 	private final FileSystem fs;
 	private boolean exists;
 	private final FileStatus status;
+	private final CompressionCodecFactory codecsFactory;
 
-	HdfsResource(String location, FileSystem fs) {
-		this(location, null, fs);
+	HdfsResource(String location, FileSystem fs, CompressionCodecFactory codecsFactory) {
+		this(location, null, fs, codecsFactory);
 	}
 
-	HdfsResource(String parent, String child, FileSystem fs) {
-		this(StringUtils.hasText(child) ? new Path(new Path(URI.create(parent)), new Path(URI.create(child))) : new Path(URI.create(parent)), fs);
+	HdfsResource(String parent, String child, FileSystem fs, CompressionCodecFactory codecsFactory) {
+		this(StringUtils.hasText(child) ? new Path(new Path(URI.create(parent)), new Path(URI.create(child))) : new Path(URI.create(parent)), fs, codecsFactory);
 	}
 
-	HdfsResource(Path path, FileSystem fs) {
+	HdfsResource(Path path, FileSystem fs, CompressionCodecFactory codecsFactory) {
 		Assert.notNull(path, "a valid path is required");
 		Assert.notNull(fs, "non null file system required");
 
@@ -76,6 +81,7 @@ class HdfsResource implements ContextResource {
 		} catch (Exception ex) {
 		}
 		this.status = status;
+		this.codecsFactory = codecsFactory;
 	}
 
 
@@ -89,7 +95,7 @@ public long contentLength() throws IOException {
 	}
 
 	public Resource createRelative(String relativePath) throws IOException {
-		return new HdfsResource(location, relativePath, fs);
+		return new HdfsResource(location, relativePath, fs, codecsFactory);
 	}
 
 	public boolean exists() {
@@ -142,7 +148,22 @@ public long lastModified() throws IOException {
 
 	public InputStream getInputStream() throws IOException {
 		if (exists) {
-			return fs.open(path);
+			InputStream stream = fs.open(path);
+
+			if (codecsFactory != null) {
+				CompressionCodec codec = codecsFactory.getCodec(path);
+				if (codec != null) {
+					// the pool is not used since the returned inputstream needs to be decorated to return the decompressor on close
+					// which can mask the actual stream
+					// it's also unclear whether the pool is actually useful or not
+					// Decompressor decompressor = CodecPool.getDecompressor(codec);
+					// stream = (decompressor != null ? codec.createInputStream(stream, decompressor) : codec.createInputStream(stream));
+					
+					stream = codec.createInputStream(stream);
+				}
+			}
+
+			return stream;
 		}
 		throw new IOException("Cannot open stream for " + getDescription());
 	}

src/main/java/org/springframework/data/hadoop/fs/HdfsResourceLoader.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.core.PriorityOrdered;
 import org.springframework.core.io.Resource;
@@ -48,6 +49,8 @@ public class HdfsResourceLoader implements ResourcePatternResolver, PriorityOrde
 	private final FileSystem fs;
 	private final PathMatcher pathMatcher = new AntPathMatcher();
 	private final boolean internalFS;
+	private volatile boolean useCodecs = true;
+	private volatile CompressionCodecFactory codecsFactory;
 
 	/**
 	 * Constructs a new <code>HdfsResourceLoader</code> instance.
@@ -70,6 +73,7 @@ public HdfsResourceLoader(Configuration config, URI uri, String user) {
 
 		internalFS = true;
 		FileSystem tempFS = null;
+		codecsFactory = new CompressionCodecFactory(config);
 
 		try {
 			if (uri == null) {
@@ -103,6 +107,7 @@ public HdfsResourceLoader(FileSystem fs) {
 		Assert.notNull(fs, "a non-null file-system required");
 		this.fs = fs;
 		internalFS = false;
+		codecsFactory = new CompressionCodecFactory(fs.getConf());
 	}
 
 	/**
@@ -119,7 +124,11 @@ public ClassLoader getClassLoader() {
 	}
 
 	public Resource getResource(String location) {
-		return new HdfsResource(location, fs);
+		return new HdfsResource(location, fs, codecs());
+	}
+
+	private CompressionCodecFactory codecs() {
+		return (useCodecs ? codecsFactory : null);
 	}
 
 	public Resource[] getResources(String locationPattern) throws IOException {
@@ -209,15 +218,15 @@ private void doRetrieveMatchingResources(Path rootDir, String subPattern, Set<Re
 					}
 
 					else if (pathMatcher.match(subPattern, location)) {
-						results.add(new HdfsResource(p, fs));
+						results.add(new HdfsResource(p, fs, codecs()));
 					}
 				}
 			}
 		}
 
 		// Remove "if" to allow folders to be added as well
 		else if (pathMatcher.match(subPattern, stripPrefix(rootDir.toUri().getPath()))) {
-			results.add(new HdfsResource(rootDir, fs));
+			results.add(new HdfsResource(rootDir, fs, codecs()));
 		}
 	}
 
@@ -241,4 +250,15 @@ public void close() throws IOException {
 			fs.close();
 		}
 	}
+
+	/**
+	 * Indicates whether to use (or not) the codecs found inside the Hadoop configuration.
+	 * This affects the content of the streams backing this resource - whether the raw content is delivered as is
+	 * or decompressed on the fly (if the configuration allows it so). The latter is the default.
+	 * 
+	 * @param useCodecs whether to use any codecs defined in the Hadoop configuration
+	 */
+	public void setUseCodecs(boolean useCodecs) {
+		this.useCodecs = useCodecs;
+	}
 }

src/test/java/org/springframework/data/hadoop/fs/HdfsResouceLoaderTest.java

@@ -15,7 +15,9 @@
  */
 package org.springframework.data.hadoop.fs;
 
+import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.io.OutputStream;
 import java.net.URI;
 import java.net.URL;
 import java.util.Arrays;
@@ -26,6 +28,8 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FsUrlStreamHandlerFactory;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.compress.DecompressorStream;
+import org.apache.hadoop.io.compress.DefaultCodec;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.After;
 import org.junit.Before;
@@ -230,4 +234,46 @@ public void testFilesMatchWithPrefix() throws Exception {
 			fs.delete(new Path("local/"), true);
 		}
 	}
+
+	@Test
+	public void testDecompressedStream() throws Exception {
+		DefaultCodec codec = new DefaultCodec();
+		codec.setConf(fs.getConf());
+		String name = "local/" + UUID.randomUUID() + codec.getDefaultExtension();
+		OutputStream outputStream = codec.createOutputStream(fs.create(new Path(name)));
+		byte[] content = name.getBytes();
+		outputStream.write(content);
+		outputStream.close();
+
+		Resource resource = loader.getResource(name);
+		assertNotNull(resource);
+		InputStream inputStream = resource.getInputStream();
+		assertEquals(DecompressorStream.class, inputStream.getClass());
+		assertTrue(TestUtils.compareStreams(new ByteArrayInputStream(content), inputStream));
+	}
+
+	@Test
+	public void testCompressedStream() throws Exception {
+
+		DefaultCodec codec = new DefaultCodec();
+		codec.setConf(fs.getConf());
+		String name = "local/" + UUID.randomUUID() + codec.getDefaultExtension();
+		OutputStream outputStream = codec.createOutputStream(fs.create(new Path(name)));
+		byte[] content = name.getBytes();
+		outputStream.write(content);
+		outputStream.close();
+
+		loader.setUseCodecs(false);
+
+		try {
+			Resource resource = loader.getResource(name);
+			assertNotNull(resource);
+			InputStream inputStream = resource.getInputStream();
+			System.out.println(inputStream.getClass());
+			assertFalse(DecompressorStream.class.equals(inputStream.getClass()));
+			assertFalse(TestUtils.compareStreams(new ByteArrayInputStream(content), inputStream));
+		} finally {
+			loader.setUseCodecs(true);
+		}
+	}
 }