package com.twitter.elephantbird.mapreduce.input;

import com.google.common.base.Preconditions;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.IOContext;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.IndexOutput;

import java.io.IOException;
import java.util.Collection;

/**
 * Implementation of a Lucene {@link Directory} for reading indexes directly off HDFS.
 * Note: this implementation is READ ONLY, it cannot be used to write to HDFS
 *
 * @author Jimmy Lin
 */
public class LuceneHdfsDirectory extends Directory {
  private final FileSystem fs;
  private final Path dir;

  public LuceneHdfsDirectory(String name, FileSystem fs) {
    this.fs = Preconditions.checkNotNull(fs,
        "FileSystem provided to LuceneHdfsDirectory cannot be null");
    Preconditions.checkNotNull(name, "File name provided to LuceneHdfsDirectory cannot be null");
    dir = new Path(name);
  }

  public LuceneHdfsDirectory(Path path, FileSystem fs) {
    this.fs = Preconditions.checkNotNull(fs);
    dir = path;
  }

  @Override
  public void close() throws IOException {
    fs.close();
  }

  @Override
  public void deleteFile(String name) throws IOException {
    Path path = new Path(dir, name);
    fs.delete(path, false);
  }

  @Override
  public boolean fileExists(String name) throws IOException {
    return fs.exists(new Path(dir, name));
  }

  @Override
  public long fileLength(String name) throws IOException {
    return fs.getFileStatus(new Path(dir, name)).getLen();
  }

  @Override
  public String[] listAll() throws IOException {
    FileStatus[] statuses = fs.listStatus(dir);
    String[] files = new String[statuses.length];

    for (int i = 0; i < statuses.length; i++) {
      files[i] = statuses[i].getPath().getName();
    }
    return files;
  }

  @Override
  public IndexInput openInput(String name, IOContext context) throws IOException {
    return new HDFSIndexInput(new Path(dir, name).toString());
  }

  @Override
  public IndexOutput createOutput(String name, IOContext context) throws IOException {
    throw new UnsupportedOperationException();
  }

  // TODO do we need to do anything here?
  @Override
  public void sync(Collection<String> strings) throws IOException { }

  private class HDFSIndexInput extends IndexInput {
    private Path path;
    private final FSDataInputStream in;
    private String resourceDescription;

    protected HDFSIndexInput(String resourceDescription) throws IOException {
      super(resourceDescription);
      this.resourceDescription = resourceDescription;
      path = new Path(resourceDescription);
      this.in = fs.open(path);
    }

    @Override
    public void close() throws IOException {
      in.close();
    }

    @Override
    public long getFilePointer() {
      try {
        return in.getPos();
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }

    @Override
    public long length() {
      try {
        return fs.getFileStatus(path).getLen();
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }

    @Override
    public byte readByte() throws IOException {
      return in.readByte();
    }

    @Override
    public void readBytes(byte[] b, int offset, int len) throws IOException {
      // Important: use readFully instead of read.
      in.readFully(b, offset, len);
    }

    @Override
    public void seek(long pos) throws IOException {
      in.seek(pos);
    }

    // TODO find out if this is necessary anymore
    @Override
    public IndexInput clone() {
      try {
        HDFSIndexInput copy = new HDFSIndexInput(resourceDescription);
        copy.seek(getFilePointer());
        return copy;
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  }
}

@@ -467,26 +467,81 @@
 va.util.
-Collection
+Arrays;%0Aimport java.util.Collection;%0Aimport java.util.Collections
 ;%0A%0A/**%0A 
@@ -785,16 +785,83 @@
 ctory %7B%0A
+  private static final String%5B%5D EMPTY_STRING_LIST = new String%5B0%5D;%0A
   privat
@@ -1204,24 +1204,200 @@
 Path(name);%0A
+    try %7B%0A      Preconditions.checkArgument(fs.exists(dir), %22Directory: %22 + dir + %22 does not exist!%22);%0A    %7D catch (IOException e) %7B%0A      throw new RuntimeException(e);%0A    %7D%0A
   %7D%0A%0A  publi
@@ -2044,24 +2044,24 @@
 Exception %7B%0A
-
     FileStat
@@ -2096,16 +2096,352 @@
 s(dir);%0A
+%0A    // some versions of hadoop return null instead of an empty list%0A    // or throwing an exception for non-existent directories%0A    if (statuses == null) %7B%0A      if (fs.exists(dir)) %7B%0A        return EMPTY_STRING_LIST;%0A      %7D else %7B%0A        throw new IllegalArgumentException(%22Directory: %22 + dir + %22 does not exist!%22);%0A      %7D%0A    %7D%0A%0A
     Stri

Add better error handling to HDFSDirectory