<commit_before>// -*- mode:Java; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
package org.apache.hadoop.fs.ceph;

import java.io.IOException;
import java.io.FileNotFoundException;
import java.io.OutputStream;
import java.net.URI;
import java.util.Set;
import java.util.EnumSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.CreateFlag;

/**
 * <p>
 * A {@link FileSystem} backed by <a href="http://ceph.sourceforge.net">Ceph.</a>.
 * This will not start a Ceph instance; one must already be running.
 * </p>
  */
public class CephFileSystem extends FileSystem {

  private static final long DEFAULT_BLOCK_SIZE = 8 * 1024 * 1024;
  
  static {
    System.load("/usr/local/lib/libhadoopcephfs.so");
  }
  
  private URI uri;

  private FileSystem localFs;
  
  private Path root;

  private Path parent;

  private static boolean debug = true;
  
  private native boolean ceph_initializeClient();
  private native boolean ceph_copyFromLocalFile(String localPath, String cephPath);
  private native boolean ceph_copyToLocalFile(String cephPath, String localPath);
  private native String  ceph_getcwd();
  private native boolean ceph_setcwd(String path);
  private native boolean ceph_rmdir(String path);
  private native boolean ceph_mkdir(String path);
  private native boolean ceph_unlink(String path);
  private native boolean ceph_rename(String old_path, String new_path);
  private native boolean ceph_exists(String path);
  private native long    ceph_getblocksize(String path);
  private native long    ceph_getfilesize(String path);
  private native boolean ceph_isdirectory(String path);
  private native boolean ceph_isfile(String path);
  private native String[] ceph_getdir(String path);
  private native int ceph_mkdirs(String path, int mode);
  private native int ceph_open_for_append(String path);
  private native int ceph_open_for_read(String path);
  private native int ceph_open_for_overwrite(String path, int mode);
  private native boolean ceph_kill_client();

  public CephFileSystem() {
    debug("CephFileSystem:enter");
    root = new Path("/");
    parent = new Path("..");
    debug("CephFileSystem:exit");
  }

  /*
    public S3FileSystem(FileSystemStore store) {
    this.store = store;
    } */

  public URI getUri() {
    debug("getUri:enter");
    debug("getUri:exit with return " + uri);
    return uri;
  }

  @Override
    public void initialize(URI uri, Configuration conf) throws IOException {
    debug("initialize:enter");
    //store.initialize(uri, conf);
    setConf(conf);
    this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());    

    // TODO: local filesystem? we really need to figure out this conf thingy
    this.localFs = get(URI.create("file:///"), conf);

    //  Initializes the client
    if (!ceph_initializeClient()) {
      throw new IOException("Ceph initialization failed!");
    }
    debug("Initialized client. Setting cwd to /");
    ceph_setcwd("/");
    debug("initialize:exit");
  }

  @Override
    public void close() throws IOException {
    debug("close:enter");
    //ceph_kill_client();
    //for some reason this just hangs, so not doing it for now
    debug("close:exit");
  }

  public FSDataOutputStream append (Path file, int bufferSize,
				    Progressable progress) throws IOException {
    debug("append:enter with path " + file + " bufferSize " + bufferSize);
    Path abs_path = makeAbsolute(file);
    int fd = ceph_open_for_append(abs_path.toString());
    if( fd < 0 ) { //error in open
      throw new IOException("append: Open for append failed on path \"" +
			    abs_path.toString() + "\"");
    }
    CephOutputStream cephOStream = new CephOutputStream(getConf(), fd);
    debug("append:exit");
    return new FSDataOutputStream(cephOStream);
  }

  public String getName() {
    debug("getName:enter");
    debug("getName:exit with value " + getUri().toString());
    return getUri().toString();
  }

  public Path getWorkingDirectory() {
    debug("getWorkingDirectory:enter");
    debug("Working directory is " + ceph_getcwd());
    debug("getWorkingDirectory:exit");
    return makeAbsolute(new Path(ceph_getcwd()));
  }

  @Override
    public void setWorkingDirectory(Path dir) {
    debug("setWorkingDirecty:enter with new working dir " + dir);
    Path abs_path = makeAbsolute(dir);

    // error conditions if path's not a directory
    boolean isDir = false;
    boolean path_exists = false;
    try {
      isDir = isDirectory(abs_path);
      path_exists = exists(abs_path);
    }

    catch (IOException e) {
      debug("Warning: isDirectory threw an exception");
    }

    if (!isDir) {
      if (path_exists)
	debug("Warning: SetWorkingDirectory(" + dir.toString() + 
			   "): path is not a directory");
      else
	debug("Warning: SetWorkingDirectory(" + dir.toString() + 
			   "): path does not exist");
    }
    else {
      debug("calling ceph_setcwd from Java");
      ceph_setcwd(dir.toString());
      debug("returned from ceph_setcwd to Java" );
    }
    debug("setWorkingDirectory:exit");
  }

  // Makes a Path absolute. In a cheap, dirty hack, we're
  // also going to strip off any "ceph://null" prefix we see. 
  private Path makeAbsolute(Path path) {
    debug("makeAbsolute:enter with path " + path);
    // first, check for the prefix
    if (path.toString().startsWith("ceph://null")) {
	  
      Path stripped_path = new Path(path.toString().substring("ceph://null".length()));
      debug("makeAbsolute:exit with path " + stripped_path);
      return stripped_path;
    }


    if (path.isAbsolute()) {
      debug("makeAbsolute:exit with path " + path);
      return path;
    }
    Path wd = getWorkingDirectory();
    if (wd.toString().equals("")){
      Path new_path = new Path(root, path);
      debug("makeAbsolute:exit with path " + new_path);
      return new_path;
    }
    else {
      Path new_path = new Path(root, path);
      debug("makeAbsolute:exit with path " + new_path);
      return new_path;
    }
  }

  @Override
    public boolean exists(Path path) throws IOException {
    debug("exists:enter with path " + path);
    boolean result;
    Path abs_path = makeAbsolute(path);
    if (abs_path.toString().equals("/")) {
      result = true;
    }
    else {
      debug("Calling ceph_exists from Java on path " + abs_path.toString() + ":");
      result =  ceph_exists(abs_path.toString());
      debug("Returned from ceph_exists to Java");
    }
    debug("exists:exit with value " + result);
    return result;
  }


  /* Creates the directory and all nonexistent parents.   */
  public boolean mkdirs(Path path, FsPermission perms) throws IOException {
    debug("mkdirs:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    debug("calling ceph_mkdirs from Java");
    int result = ceph_mkdirs(abs_path.toString(), (int)perms.toShort());
    debug("Returned from ceph_mkdirs to Java with result " + result);
    debug("mkdirs:exit with result " + result);
    if (result != 0)
      return true;
    else return false;
  }

  public boolean __isDirectory(Path path) throws IOException {
    debug("__isDirectory:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    boolean result;
    if (abs_path.toString().equals("/")) {
      result = true;
    }
    else {
      debug("calling ceph_isdirectory from Java");
      result = ceph_isdirectory(abs_path.toString());
      debug("Returned from ceph_isdirectory to Java");
    }
    debug("__isDirectory:exit with result " + result);
    return result;
  }

  @Override
    public boolean isFile(Path path) throws IOException {
    debug("isFile:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    boolean result;
    if (abs_path.toString().equals("/")) {
      result =  false;
    }
    else {
      result = ceph_isfile(abs_path.toString());
    }
    debug("isFile:exit with result " + result);
    return result;
  }

  public FileStatus getFileStatus(Path p) throws IOException {
    // For the moment, hardwired replication and modification time
    debug("getFileStatus:enter with path " + p);
    Path abs_p = makeAbsolute(p);
    //check it's a file or dir!
    boolean isFile, isDir;
    if (!(isFile=isFile(p)) && !(isDir=__isDirectory(p))) {
      throw new FileNotFoundException("org.apache.hadoop.fs.ceph.CephFileSystem.getFileStatus:not a file or dir!");
    }
    FileStatus status =  new FileStatus(__getLength(abs_p), __isDirectory(abs_p), 2,
			  getBlockSize(p), 0, abs_p);
    debug("getFileStatus:exit");
    return status;
  }

  // array of statuses for the directory's contents
  public FileStatus[] listStatus(Path p) throws IOException {
    debug("listStatus:enter with path " + p);
    Path abs_p = makeAbsolute(p);
    Path[] paths = listPaths(abs_p);
    FileStatus[] statuses = new FileStatus[paths.length];
    for (int i = 0; i < paths.length; ++i) {
      statuses[i] = getFileStatus(paths[i]);
    }
    debug("listStatus:exit");
    return statuses;
  }
  <commit_msg>Hadoop: getFileStatus behaves better with folders.<commit_after>// -*- mode:Java; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
package org.apache.hadoop.fs.ceph;

import java.io.IOException;
import java.io.FileNotFoundException;
import java.io.OutputStream;
import java.net.URI;
import java.util.Set;
import java.util.EnumSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.CreateFlag;

/**
 * <p>
 * A {@link FileSystem} backed by <a href="http://ceph.sourceforge.net">Ceph.</a>.
 * This will not start a Ceph instance; one must already be running.
 * </p>
  */
public class CephFileSystem extends FileSystem {

  private static final long DEFAULT_BLOCK_SIZE = 8 * 1024 * 1024;
  
  static {
    System.loadLibrary("hadoopcephfs");
  }
  
  private URI uri;

  private FileSystem localFs;
  
  private Path root;

  private Path parent;

  private static boolean debug = true;
  
  private native boolean ceph_initializeClient();
  private native boolean ceph_copyFromLocalFile(String localPath, String cephPath);
  private native boolean ceph_copyToLocalFile(String cephPath, String localPath);
  private native String  ceph_getcwd();
  private native boolean ceph_setcwd(String path);
  private native boolean ceph_rmdir(String path);
  private native boolean ceph_mkdir(String path);
  private native boolean ceph_unlink(String path);
  private native boolean ceph_rename(String old_path, String new_path);
  private native boolean ceph_exists(String path);
  private native long    ceph_getblocksize(String path);
  private native long    ceph_getfilesize(String path);
  private native boolean ceph_isdirectory(String path);
  private native boolean ceph_isfile(String path);
  private native String[] ceph_getdir(String path);
  private native int ceph_mkdirs(String path, int mode);
  private native int ceph_open_for_append(String path);
  private native int ceph_open_for_read(String path);
  private native int ceph_open_for_overwrite(String path, int mode);
  private native boolean ceph_kill_client();

  public CephFileSystem() {
    debug("CephFileSystem:enter");
    root = new Path("/");
    parent = new Path("..");
    debug("CephFileSystem:exit");
  }

  /*
    public S3FileSystem(FileSystemStore store) {
    this.store = store;
    } */

  public URI getUri() {
    debug("getUri:enter");
    debug("getUri:exit with return " + uri);
    return uri;
  }

  @Override
    public void initialize(URI uri, Configuration conf) throws IOException {
    debug("initialize:enter");
    //store.initialize(uri, conf);
    setConf(conf);
    this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());    

    // TODO: local filesystem? we really need to figure out this conf thingy
    this.localFs = get(URI.create("file:///"), conf);

    //  Initializes the client
    if (!ceph_initializeClient()) {
      throw new IOException("Ceph initialization failed!");
    }
    debug("Initialized client. Setting cwd to /");
    ceph_setcwd("/");
    debug("initialize:exit");
  }

  @Override
    public void close() throws IOException {
    debug("close:enter");
    ceph_kill_client();
    //for some reason this just hangs, so not doing it for now
    debug("close:exit");
  }

  public FSDataOutputStream append (Path file, int bufferSize,
				    Progressable progress) throws IOException {
    debug("append:enter with path " + file + " bufferSize " + bufferSize);
    Path abs_path = makeAbsolute(file);
    int fd = ceph_open_for_append(abs_path.toString());
    if( fd < 0 ) { //error in open
      throw new IOException("append: Open for append failed on path \"" +
			    abs_path.toString() + "\"");
    }
    CephOutputStream cephOStream = new CephOutputStream(getConf(), fd);
    debug("append:exit");
    return new FSDataOutputStream(cephOStream);
  }

  public String getName() {
    debug("getName:enter");
    debug("getName:exit with value " + getUri().toString());
    return getUri().toString();
  }

  public Path getWorkingDirectory() {
    debug("getWorkingDirectory:enter");
    debug("Working directory is " + ceph_getcwd());
    debug("getWorkingDirectory:exit");
    return makeAbsolute(new Path(ceph_getcwd()));
  }

  @Override
    public void setWorkingDirectory(Path dir) {
    debug("setWorkingDirecty:enter with new working dir " + dir);
    Path abs_path = makeAbsolute(dir);

    // error conditions if path's not a directory
    boolean isDir = false;
    boolean path_exists = false;
    try {
      isDir = isDirectory(abs_path);
      path_exists = exists(abs_path);
    }

    catch (IOException e) {
      debug("Warning: isDirectory threw an exception");
    }

    if (!isDir) {
      if (path_exists)
	debug("Warning: SetWorkingDirectory(" + dir.toString() + 
			   "): path is not a directory");
      else
	debug("Warning: SetWorkingDirectory(" + dir.toString() + 
			   "): path does not exist");
    }
    else {
      debug("calling ceph_setcwd from Java");
      ceph_setcwd(dir.toString());
      debug("returned from ceph_setcwd to Java" );
    }
    debug("setWorkingDirectory:exit");
  }

  // Makes a Path absolute. In a cheap, dirty hack, we're
  // also going to strip off any "ceph://null" prefix we see. 
  private Path makeAbsolute(Path path) {
    debug("makeAbsolute:enter with path " + path);
    // first, check for the prefix
    if (path.toString().startsWith("ceph://null")) {
	  
      Path stripped_path = new Path(path.toString().substring("ceph://null".length()));
      debug("makeAbsolute:exit with path " + stripped_path);
      return stripped_path;
    }


    if (path.isAbsolute()) {
      debug("makeAbsolute:exit with path " + path);
      return path;
    }
    Path wd = getWorkingDirectory();
    if (wd.toString().equals("")){
      Path new_path = new Path(root, path);
      debug("makeAbsolute:exit with path " + new_path);
      return new_path;
    }
    else {
      Path new_path = new Path(root, path);
      debug("makeAbsolute:exit with path " + new_path);
      return new_path;
    }
  }

  @Override
    public boolean exists(Path path) throws IOException {
    debug("exists:enter with path " + path);
    boolean result;
    Path abs_path = makeAbsolute(path);
    if (abs_path.toString().equals("/")) {
      result = true;
    }
    else {
      debug("Calling ceph_exists from Java on path " + abs_path.toString() + ":");
      result =  ceph_exists(abs_path.toString());
      debug("Returned from ceph_exists to Java");
    }
    debug("exists:exit with value " + result);
    return result;
  }


  /* Creates the directory and all nonexistent parents.   */
  public boolean mkdirs(Path path, FsPermission perms) throws IOException {
    debug("mkdirs:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    debug("calling ceph_mkdirs from Java");
    int result = ceph_mkdirs(abs_path.toString(), (int)perms.toShort());
    debug("Returned from ceph_mkdirs to Java with result " + result);
    debug("mkdirs:exit with result " + result);
    if (result != 0)
      return true;
    else return false;
  }

  public boolean __isDirectory(Path path) throws IOException {
    debug("__isDirectory:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    boolean result;
    if (abs_path.toString().equals("/")) {
      result = true;
    }
    else {
      debug("calling ceph_isdirectory from Java");
      result = ceph_isdirectory(abs_path.toString());
      debug("Returned from ceph_isdirectory to Java");
    }
    debug("__isDirectory:exit with result " + result);
    return result;
  }

  @Override
    public boolean isFile(Path path) throws IOException {
    debug("isFile:enter with path " + path);
    Path abs_path = makeAbsolute(path);
    boolean result;
    if (abs_path.toString().equals("/")) {
      result =  false;
    }
    else {
      result = ceph_isfile(abs_path.toString());
    }
    debug("isFile:exit with result " + result);
    return result;
  }

  public FileStatus getFileStatus(Path p) throws IOException {
    debug("getFileStatus:enter with path " + p);
    Path abs_p = makeAbsolute(p);
    // For the moment, hardwired replication and modification time
    int replication = 2;
    int mod_time = 0;
    FileStatus status;
    if (isFile(abs_p)) {
      debug("getFileStatus: is file");
      status = new FileStatus(__getLength(abs_p), false, replication, getBlockSize(abs_p),
			      mod_time, abs_p);
    }
    else if (__isDirectory(abs_p)) {
      debug("getFileStatus: is directory");
      status = new FileStatus( 0, true, replication, 0,
			      mod_time, abs_p);
    }
    else { //fail out
      throw new FileNotFoundException("org.apache.hadoop.fs.ceph.CephFileSystem: File "
				      + p + " does not exist.");
    }
    debug("getFileStatus:exit");
    return status;
  }

  // array of statuses for the directory's contents
  public FileStatus[] listStatus(Path p) throws IOException {
    debug("listStatus:enter with path " + p);
    Path abs_p = makeAbsolute(p);
    Path[] paths = listPaths(abs_p);
    FileStatus[] statuses = new FileStatus[paths.length];
    for (int i = 0; i < paths.length; ++i) {
      statuses[i] = getFileStatus(paths[i]);
    }
    debug("listStatus:exit");
    return statuses;
  }
  
{'fraction_non_alphanumeric': 0.11980068187778652, 'fraction_numerical': 0.0018882769472856019, 'mean_word_length': 3.4598830409356727, 'pattern_counts': {'":': 4, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '241134', 'n_tokens_mistral': 5868, 'n_tokens_neox': 5674, 'n_words': 1556}