Repository: incubator-blur Updated Branches: refs/heads/master 08565d13d -> ba12672c5
Cleaning up more code. Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/09e4068b Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/09e4068b Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/09e4068b Branch: refs/heads/master Commit: 09e4068bfc65eff9c059da6171d73c4098a6ec7a Parents: 08565d1 Author: Aaron McCurry <[email protected]> Authored: Mon Apr 6 06:12:41 2015 -0400 Committer: Aaron McCurry <[email protected]> Committed: Mon Apr 6 06:14:46 2015 -0400 ---------------------------------------------------------------------- .../apache/blur/store/hdfs/HdfsDirectory.java | 43 +++++++++++--------- .../apache/blur/store/hdfs/HdfsIndexInput.java | 19 ++++----- 2 files changed, 32 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/09e4068b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java index befcca9..7f695e3 100644 --- a/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java +++ b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java @@ -135,7 +135,6 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin protected final Map<Path, FSDataInputRandomAccess> _inputMap = new ConcurrentHashMap<Path, FSDataInputRandomAccess>(); protected final boolean _useCache = true; protected final boolean _asyncClosing; - protected final Path _localCachePath = new Path("/tmp/cache"); protected final SequentialReadControl _sequentialReadControl; public HdfsDirectory(Configuration configuration, Path path) throws IOException { @@ -310,14 +309,13 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin } FSDataInputRandomAccess inputRandomAccess = openForInput(name); long fileLength = fileLength(name); - Path path = getPath(name); - HdfsIndexInput input = new HdfsIndexInput(this, inputRandomAccess, fileLength, _metricsGroup, path, + HdfsIndexInput input = new HdfsIndexInput(this, inputRandomAccess, fileLength, _metricsGroup, name, _sequentialReadControl.clone()); return input; } protected synchronized FSDataInputRandomAccess openForInput(String name) throws IOException { - Path path = getPath(name); + final Path path = getPath(name); FSDataInputRandomAccess input = _inputMap.get(path); if (input != null) { return input; @@ -336,6 +334,12 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin public int read(long filePointer, byte[] b, int offset, int length) throws IOException { return inputStream.read(filePointer, b, offset, length); } + + @Override + public String toString() { + return path.toString(); + } + }; _inputMap.put(path, randomInputStream); return randomInputStream; @@ -591,14 +595,15 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin return this; } - protected FSDataInputSequentialAccess openForSequentialInput(Path p, Object key) throws IOException { - return openInputStream(_fileSystem, p, key); + protected FSDataInputSequentialAccess openForSequentialInput(String name, Object key) throws IOException { + return openInputStream(name, key, _fileSystem); } - protected FSDataInputSequentialAccess openInputStream(FileSystem fileSystem, Path p, Object key) throws IOException { - final FSDataInputStream input = fileSystem.open(p); - WEAK_CLOSING_QUEUE.add(new WeakRef(input, key)); - return new FSDataInputSequentialAccess() { + protected FSDataInputSequentialAccess openInputStream(String name, Object key, FileSystem fileSystem) + throws IOException { + final Path path = getPath(name); + final FSDataInputStream input = fileSystem.open(path); + FSDataInputSequentialAccess sequentialAccess = new FSDataInputSequentialAccess() { @Override public void close() throws IOException { @@ -624,7 +629,15 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin public long getPos() throws IOException { return input.getPos(); } + + @Override + public String toString() { + return path.toString(); + } + }; + WEAK_CLOSING_QUEUE.add(new WeakRef(sequentialAccess, key)); + return sequentialAccess; } static class WeakRef { @@ -643,14 +656,4 @@ public class HdfsDirectory extends Directory implements LastModified, HdfsSymlin } - protected boolean isAlreadyExistsLocally(FileSystem localFileSystem, Path localFile, long length) throws IOException { - if (localFileSystem.exists(localFile)) { - FileStatus fileStatus = localFileSystem.getFileStatus(localFile); - if (fileStatus.getLen() == length) { - return true; - } - } - return false; - } - } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/09e4068b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsIndexInput.java ---------------------------------------------------------------------- diff --git a/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsIndexInput.java b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsIndexInput.java index 3abff1e..665a025 100644 --- a/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsIndexInput.java +++ b/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsIndexInput.java @@ -23,7 +23,6 @@ import org.apache.blur.log.LogFactory; import org.apache.blur.store.buffer.ReusedBufferedIndexInput; import org.apache.blur.trace.Trace; import org.apache.blur.trace.Tracer; -import org.apache.hadoop.fs.Path; import org.apache.lucene.store.IndexInput; public class HdfsIndexInput extends ReusedBufferedIndexInput { @@ -33,7 +32,7 @@ public class HdfsIndexInput extends ReusedBufferedIndexInput { private final long _length; private final FSDataInputRandomAccess _input; private final MetricsGroup _metricsGroup; - private final Path _path; + private final String _name; private final HdfsDirectory _dir; private SequentialReadControl _sequentialReadControl; @@ -41,15 +40,15 @@ public class HdfsIndexInput extends ReusedBufferedIndexInput { private long _prevFilePointer; private FSDataInputSequentialAccess _sequentialInput; - public HdfsIndexInput(HdfsDirectory dir, FSDataInputRandomAccess inputStream, long length, MetricsGroup metricsGroup, - Path path, SequentialReadControl sequentialReadControl) throws IOException { - super("HdfsIndexInput(" + path.toString() + ")"); + public HdfsIndexInput(HdfsDirectory dir, FSDataInputRandomAccess input, long length, MetricsGroup metricsGroup, + String name, SequentialReadControl sequentialReadControl) throws IOException { + super("HdfsIndexInput(" + name + "@" + "" + input + ")"); _sequentialReadControl = sequentialReadControl; _dir = dir; - _input = inputStream; + _input = input; _length = length; _metricsGroup = metricsGroup; - _path = path; + _name = name; } @Override @@ -91,9 +90,9 @@ public class HdfsIndexInput extends ReusedBufferedIndexInput { _sequentialReadControl.setEnabled(true); if (_sequentialInput == null) { - Tracer trace = Trace.trace("filesystem - read - openForSequentialInput", Trace.param("file", _path), + Tracer trace = Trace.trace("filesystem - read - openForSequentialInput", Trace.param("file", toString()), Trace.param("location", getFilePointer())); - _sequentialInput = _dir.openForSequentialInput(_path, this); + _sequentialInput = _dir.openForSequentialInput(_name, this); trace.done(); } } @@ -112,7 +111,7 @@ public class HdfsIndexInput extends ReusedBufferedIndexInput { } private long randomAccessRead(byte[] b, int offset, int length, long start, long filePointer) throws IOException { - Tracer trace = Trace.trace("filesystem - read - randomAccessRead", Trace.param("file", _path), + Tracer trace = Trace.trace("filesystem - read - randomAccessRead", Trace.param("file", toString()), Trace.param("location", getFilePointer()), Trace.param("length", length)); try { int olen = length;
