/*
 * 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.pig.impl.io;

import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Shell;
import org.apache.pig.ExecType;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.backend.datastorage.ContainerDescriptor;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.datastorage.DataStorageException;
import org.apache.pig.backend.datastorage.ElementDescriptor;
import org.apache.pig.backend.datastorage.SeekableInputStream;
import org.apache.pig.backend.datastorage.SeekableInputStream.FLAGS;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.datastorage.HDataStorage;
import org.apache.pig.backend.hadoop.datastorage.HDirectory;
import org.apache.pig.backend.hadoop.datastorage.HPath;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.util.Utils;

import com.google.common.annotations.VisibleForTesting;

public class FileLocalizer {
    private static final Log log = LogFactory.getLog(FileLocalizer.class);

    static public final String LOCAL_PREFIX  = "file:";
    static public final int STYLE_UNIX = 0;
    static public final int STYLE_WINDOWS = 1;

    public static FsPermission OWNER_ONLY_PERMS = new FsPermission(FsAction.ALL, FsAction.NONE,
            FsAction.NONE); // rwx------

    public static class DataStorageInputStreamIterator extends InputStream {
        InputStream current;
        ElementDescriptor[] elements;
        int currentElement;

        public DataStorageInputStreamIterator(ElementDescriptor[] elements) {
            this.elements = elements;
        }

        private boolean isEOF() throws IOException {
            if (current == null) {
                if (currentElement == elements.length) {
                    return true;
                }
                current = elements[ currentElement++ ].open();
            }
            return false;
        }

        private void doNext() throws IOException {
            current.close();
            current = null;
        }

        @Override
        public int read() throws IOException {
            while (!isEOF()) {
                int rc = current.read();
                if (rc != -1)
                    return rc;
                doNext();
            }
            return -1;
        }

        @Override
        public int available() throws IOException {
            if (isEOF())
                return 0;
            return current.available();
        }

        @Override
        public void close() throws IOException {
            if (current != null) {
                current.close();
                current = null;
            }
            currentElement = elements.length;
        }

        @Override
        public int read(byte[] b, int off, int len) throws IOException {
            int count = 0;
            while (!isEOF() && len > 0) {
                int rc = current.read(b, off, len);
                if (rc <= 0) {
                    doNext();
                    continue;
                }
                off += rc;
                len -= rc;
                count += rc;
            }
            return count == 0 ? (isEOF() ? -1 : 0) : count;
        }

        @Override
        public int read(byte[] b) throws IOException {
            return read(b, 0, b.length);
        }

        @Override
        public long skip(long n) throws IOException {
            while (!isEOF() && n > 0) {
                n -= current.skip(n);
            }
            return n;
        }

    }

    static String checkDefaultPrefix(ExecType execType, String fileSpec) {
        if (fileSpec.startsWith(LOCAL_PREFIX))
            return fileSpec;
        return (execType == ExecType.LOCAL ? LOCAL_PREFIX : "") + fileSpec;
    }

    /**
     * This function is meant to be used if the mappers/reducers want to access any HDFS file
     * @param fileName
     * @return InputStream of the open file.
     * @throws IOException
     */
    public static InputStream openDFSFile(String fileName) throws IOException {
        Configuration conf = PigMapReduce.sJobConfInternal.get();
        if (conf == null) {
            throw new RuntimeException(
                    "can't open DFS file while executing locally");
        }

        return openDFSFile(fileName, ConfigurationUtil.toProperties(conf));

    }

    public static InputStream openDFSFile(String fileName, Properties properties) throws IOException{
        DataStorage dds = new HDataStorage(properties);
        ElementDescriptor elem = dds.asElement(fileName);
        return openDFSFile(elem);
    }

    public static long getSize(String fileName) throws IOException {
    	Configuration conf = PigMapReduce.sJobConfInternal.get();

    	if (conf == null) {
    		throw new RuntimeException(
                "can't open DFS file while executing locally");
    	}

        return getSize(fileName, ConfigurationUtil.toProperties(conf));
    }

    public static long getSize(String fileName, Properties properties) throws IOException {
    	DataStorage dds = new HDataStorage(properties);
        ElementDescriptor elem = dds.asElement(fileName);

        // recursively get all the files under this path
        ElementDescriptor[] allElems = getFileElementDescriptors(elem);

        long size = 0;

        // add up the sizes of all files found
        for (int i=0; i<allElems.length; i++) {
        	Map<String, Object> stats = allElems[i].getStatistics();
        	size += (Long) (stats.get(ElementDescriptor.LENGTH_KEY));
        }

        return size;
    }

    private static InputStream openDFSFile(ElementDescriptor elem) throws IOException{
        ElementDescriptor[] elements = null;

        if (elem.exists()) {
            try {
                if(! elem.getDataStorage().isContainer(elem.toString())) {
                    if (elem.systemElement())
                        throw new IOException ("Attempt is made to open system file " + elem.toString());
                    return elem.open();
                }
            }
            catch (DataStorageException e) {
                throw new IOException("Failed to determine if elem=" + elem + " is container", e);
            }

            // elem is a directory - recursively get all files in it
            elements = getFileElementDescriptors(elem);
        } else {
            // It might be a glob
            if (!globMatchesFiles(elem, elem.getDataStorage())) {
                throw new IOException(elem.toString() + " does not exist");
            } else {
                elements = getFileElementDescriptors(elem);
                return new DataStorageInputStreamIterator(elements);

            }
        }

        return new DataStorageInputStreamIterator(elements);
    }

    /**
     * recursively get all "File" element descriptors present in the input element descriptor
     * @param elem input element descriptor
     * @return an array of Element descriptors for files present (found by traversing all levels of dirs)
     *  in the input element descriptor
     * @throws DataStorageException
     */
    private static ElementDescriptor[] getFileElementDescriptors(ElementDescriptor elem) throws DataStorageException {
        DataStorage store = elem.getDataStorage();
        ElementDescriptor[] elems = store.asCollection(elem.toString());
        // elems could have directories in it, if so
        // get the files out so that it contains only files
        List<ElementDescriptor> paths = new ArrayList<ElementDescriptor>();
        List<ElementDescriptor> filePaths = new ArrayList<ElementDescriptor>();
        for (int m = 0; m < elems.length; m++) {
            paths.add(elems[m]);
        }
        for (int j = 0; j < paths.size(); j++) {
            ElementDescriptor fullPath = store.asElement(store
                    .getActiveContainer(), paths.get(j));
            // Skip hadoop's private/meta files ...
            if (fullPath.systemElement()) {
                continue;
            }

            if (fullPath instanceof ContainerDescriptor) {
                for (ElementDescriptor child : ((ContainerDescriptor) fullPath)) {
                    paths.add(child);
                }
                continue;
            } else {
                // this is a file, add it to filePaths
                filePaths.add(fullPath);
            }
        }
        elems = new ElementDescriptor[filePaths.size()];
        filePaths.toArray(elems);
        return elems;
    }

    private static InputStream openLFSFile(ElementDescriptor elem) throws IOException{
        // IMPORTANT NOTE: Currently we use HXXX classes to represent
        // files and dirs in local mode - so we can just delegate this
        // call to openDFSFile(elem). When we have true local mode files
        // and dirs THIS WILL NEED TO CHANGE
        return openDFSFile(elem);
    }

    /**
     * This function returns an input stream to a local file system file or
     * a file residing on Hadoop's DFS
     * @param fileName The filename to open
     * @param execType execType indicating whether executing in local mode or MapReduce mode (Hadoop)
     * @param storage The DataStorage object used to open the fileSpec
     * @return InputStream to the fileSpec
     * @throws IOException
     * @deprecated Use {@link #open(String, PigContext)} instead
     */
    @Deprecated
    static public InputStream open(String fileName, ExecType execType, DataStorage storage) throws IOException {
        fileName = checkDefaultPrefix(execType, fileName);
        if (!fileName.startsWith(LOCAL_PREFIX)) {
            ElementDescriptor elem = storage.asElement(fullPath(fileName, storage));
            return openDFSFile(elem);
        }
        else {
            fileName = fileName.substring(LOCAL_PREFIX.length());
            ElementDescriptor elem = storage.asElement(fullPath(fileName, storage));
            return openLFSFile(elem);
        }
    }

    /**
     * @deprecated Use {@link #fullPath(String, PigContext)} instead
     */
    @Deprecated
    public static String fullPath(String fileName, DataStorage storage) {
        String fullPath;
        try {
            if (fileName.charAt(0) != '/') {
                ElementDescriptor currentDir = storage.getActiveContainer();
                ElementDescriptor elem = storage.asElement(currentDir.toString(), fileName);

                fullPath = elem.toString();
            } else {
                fullPath = fileName;
            }
        } catch (DataStorageException e) {
            fullPath = fileName;
        }
        return fullPath;
    }

    static public InputStream open(String fileSpec, PigContext pigContext) throws IOException {
        fileSpec = checkDefaultPrefix(pigContext.getExecType(), fileSpec);
        if (!fileSpec.startsWith(LOCAL_PREFIX)) {
            ElementDescriptor elem = pigContext.getDfs().asElement(fullPath(fileSpec, pigContext));
            return openDFSFile(elem);
        }
        else {
            fileSpec = fileSpec.substring(LOCAL_PREFIX.length());
            //buffering because we only want buffered streams to be passed to load functions.
            /*return new BufferedInputStream(new FileInputStream(fileSpec));*/
            ElementDescriptor elem = pigContext.getLfs().asElement(fullPath(fileSpec, pigContext));
            return openLFSFile(elem);
        }
    }

    /**
     * @param fileSpec
     * @param offset
     * @param pigContext
     * @return SeekableInputStream
     * @throws IOException
     *
     * This is an overloaded version of open where there is a need to seek in stream. Currently seek is supported
     * only in file, not in directory or glob.
     */
    static public SeekableInputStream open(String fileSpec, long offset, PigContext pigContext) throws IOException {

        fileSpec = checkDefaultPrefix(pigContext.getExecType(), fileSpec);

        ElementDescriptor elem;
        if (!fileSpec.startsWith(LOCAL_PREFIX))
            elem = pigContext.getDfs().asElement(fullPath(fileSpec, pigContext));

        else{
            fileSpec = fileSpec.substring(LOCAL_PREFIX.length());
            elem = pigContext.getLfs().asElement(fullPath(fileSpec, pigContext));
        }

        if (elem.exists() && (!elem.getDataStorage().isContainer(elem.toString()))) {
            try {
                if (elem.systemElement())
                    throw new IOException ("Attempt is made to open system file " + elem.toString());

                SeekableInputStream sis = elem.sopen();
                sis.seek(offset, FLAGS.SEEK_SET);
                return sis;
            }
            catch (DataStorageException e) {
                throw new IOException("Failed to determine if elem=" + elem + " is container", e);
            }
        }
        // Either a directory or a glob.
        else
            throw new IOException("Currently seek is supported only in a file, not in glob or directory.");
    }

    static public OutputStream create(String fileSpec, PigContext pigContext) throws IOException{
        return create(fileSpec,false,pigContext);
    }

    static public OutputStream create(String fileSpec, boolean append, PigContext pigContext) throws IOException {
        fileSpec = checkDefaultPrefix(pigContext.getExecType(), fileSpec);
        if (!fileSpec.startsWith(LOCAL_PREFIX)) {
            ElementDescriptor elem = pigContext.getDfs().asElement(fileSpec);
            return elem.create();
        }
        else {
            fileSpec = fileSpec.substring(LOCAL_PREFIX.length());

            // TODO probably this should be replaced with the local file system
            File f = (new File(fileSpec)).getParentFile();
            if (f!=null){
                boolean res = f.mkdirs();
                if (!res)
                    log.warn("FileLocalizer.create: failed to create " + f);
            }

            return new FileOutputStream(fileSpec,append);
        }
    }

    static public boolean delete(String fileSpec, PigContext pigContext) throws IOException{
        fileSpec = checkDefaultPrefix(pigContext.getExecType(), fileSpec);
        ElementDescriptor elem = null;
        if (!fileSpec.startsWith(LOCAL_PREFIX)) {
            elem = pigContext.getDfs().asElement(fileSpec);
        } else {
            elem = pigContext.getLfs().asElement(fileSpec);
        }
        elem.delete();
        return true;
    }

    static Random r = new Random();

    /**
     * Thread local relativeRoot ContainerDescriptor. Do not access this object
     * directly, since it's lazy initialized in the relativeRoot(PigContext)
     * method, which should be used instead.
     */
    private static ThreadLocal<ContainerDescriptor> relativeRoot =
        new ThreadLocal<ContainerDescriptor>() {
    };

    private static ContainerDescriptor resourcePath;

    /**
     * This method is only used by test code to reset state.
     * @param initialized
     */
    public static void setInitialized(boolean initialized) {
        if (!initialized) {
            relativeRoot.set(null);
        }
    }

    /**
     * Accessor method to get the root ContainerDescriptor used for temporary
     * files bound to this thread. Calling this method lazy-initialized the
     * relativeRoot object.
     *
     * @param pigContext
     * @return
     * @throws DataStorageException
     */
    private static synchronized ContainerDescriptor relativeRoot(final PigContext pigContext)
            throws DataStorageException {
        if (relativeRoot.get() == null) {
            ContainerDescriptor relative = getTempContainer(pigContext);
            relativeRoot.set(relative);
        }
        return relativeRoot.get();
    }

    /**
     * Accessor method to get the resource ContainerDescriptor used for tez resource
     * path bound to this thread. Calling this method lazy-initialized the
     * resourcePath object. This path is different than relativeRoot in that
     * calling PigServer.shutdown will only remove relativeRoot but not resourthPath
     * since resourthPath should be available in the entire session
     *
     * @param pigContext
     * @return temporary resource path
     * @throws DataStorageException
     */
    public static synchronized Path getTemporaryResourcePath(final PigContext pigContext)
            throws DataStorageException {
        if (resourcePath == null) {
            resourcePath = getTempContainer(pigContext);
        }
        return ((HPath)resourcePath).getPath();
    }

    private static synchronized ContainerDescriptor getTempContainer(final PigContext pigContext)
            throws DataStorageException {
        ContainerDescriptor tempContainer = null;
        String tdir= Utils.substituteVars(pigContext.getProperties().getProperty(PigConfiguration.PIG_TEMP_DIR, "/tmp"));
        try {
            do {
                tempContainer = pigContext.getDfs().asContainer(tdir + "/temp" + r.nextInt());
            } while (tempContainer.exists());
            createContainer(tempContainer);
        }
        catch (IOException e) {
            // try one last time in case this was due IO Exception caused by dir
            // operations on directory created by another JVM at the same instant
            tempContainer = pigContext.getDfs().asContainer(tdir + "/temp" + r.nextInt());
            try {
                createContainer(tempContainer);
            }
            catch (IOException e1) {
                throw new DataStorageException(e1);
            }
        }
        return tempContainer;
    }

    private static void createContainer(ContainerDescriptor container) throws IOException {
        container.create();
        if (container instanceof HDirectory) {
            ((HDirectory) container).setPermission(OWNER_ONLY_PERMS);
        }
    }

    public static void deleteTempFiles() {
        if (relativeRoot.get() != null) {
            try {
                relativeRoot.get().delete();
            } catch (IOException e) {
                log.error(e);
            }
            setInitialized(false);
        }
    }

    public static void deleteTempResourceFiles() {
        if (resourcePath != null) {
            try {
                resourcePath.delete();
            } catch (IOException e) {
                log.error(e);
            }
        }
    }

    public static Path getTemporaryPath(PigContext pigContext) throws IOException {
        return getTemporaryPath(pigContext, "");
    }

    public static Path getTemporaryPath(PigContext pigContext, String suffix) throws IOException {
      ElementDescriptor relative = relativeRoot(pigContext);

      ElementDescriptor elem=
          pigContext.getDfs().asElement(relative.toString(), "tmp" + r.nextInt() + suffix);
      return ((HPath)elem).getPath();
  }

    public static String hadoopify(String filename, PigContext pigContext) throws IOException {
        if (Shell.WINDOWS){
            filename = filename.replace('\\','/');
        }
        if (filename.startsWith(LOCAL_PREFIX)) {
            filename = filename.substring(LOCAL_PREFIX.length());
        }

        ElementDescriptor localElem =
            pigContext.getLfs().asElement(filename);

        if (!localElem.exists()) {
            throw new FileNotFoundException(filename);
        }

        ElementDescriptor distribElem = pigContext.getDfs().asElement(
                getTemporaryPath(pigContext).toString());

        int suffixStart = filename.lastIndexOf('.');
        if (suffixStart != -1) {
            distribElem = pigContext.getDfs().asElement(distribElem.toString() +
                    filename.substring(suffixStart));
        }

        // TODO: currently the copy method in Data Storage does not allow to specify overwrite
        //       so the work around is to delete the dst file first, if it exists
        if (distribElem.exists()) {
            distribElem.delete();
        }
        localElem.copy(distribElem, null, false);

        return distribElem.toString();
    }

    public static String fullPath(String filename, PigContext pigContext) throws IOException {
        try {
            if (filename.charAt(0) != '/') {
                ElementDescriptor currentDir = pigContext.getDfs().getActiveContainer();
                ElementDescriptor elem = pigContext.getDfs().asElement(currentDir.toString(),
                                                                                  filename);

                return elem.toString();
            }
            return filename;
        }
        catch (DataStorageException e) {
            return filename;
        }
    }

    public static boolean fileExists(String filename, PigContext context)
            throws IOException {
        return fileExists(filename, context.getFs());
    }

    /**
     * @deprecated Use {@link #fileExists(String, PigContext)} instead
     */
    @Deprecated
    public static boolean fileExists(String filename, DataStorage store)
            throws IOException {
        ElementDescriptor elem = store.asElement(filename);
        return elem.exists() || globMatchesFiles(elem, store);
    }

    public static boolean isFile(String filename, PigContext context)
    throws IOException {
        return !isDirectory(filename, context.getDfs());
    }

    /**
     * @deprecated Use {@link #isFile(String, PigContext)} instead
     */
    @Deprecated
    public static boolean isFile(String filename, DataStorage store)
    throws IOException {
        return !isDirectory(filename, store);
    }

    public static boolean isDirectory(String filename, PigContext context)
    throws IOException {
        return isDirectory(filename, context.getDfs());
    }

    /**
    * @deprecated Use {@link #isDirectory(String, PigContext)} instead.
    */
    @Deprecated
    public static boolean isDirectory(String filename, DataStorage store)
    throws IOException {
        ElementDescriptor elem = store.asElement(filename);
        return (elem instanceof ContainerDescriptor);
    }

    private static boolean globMatchesFiles(ElementDescriptor elem,
                                            DataStorage fs)
            throws IOException
    {
        try {
            // Currently, if you give a glob with non-special glob characters, hadoop
            // returns an array with your file name in it.  So check for that.
            ElementDescriptor[] elems = fs.asCollection(elem.toString());
            switch (elems.length) {
            case 0:
                return false;

            case 1:
                return !elems[0].equals(elem);

            default:
                return true;
            }
        }
        catch (DataStorageException e) {
            throw e;
        }
    }

    @VisibleForTesting
    public static void setR(Random r) {
        FileLocalizer.r = r;
    }

    /**
     * Convert path from Windows convention to Unix convention. Invoked under
     * cygwin.
     *
     * @param path
     *            path in Windows convention
     * @return path in Unix convention, null if fail
     */
    static public String parseCygPath(String path, int style) {
        String[] command;
        if (style==STYLE_WINDOWS)
            command = new String[] { "cygpath", "-w", path };
        else
            command = new String[] { "cygpath", "-u", path };
        Process p = null;
        try {
            p = Runtime.getRuntime().exec(command);
        } catch (IOException e) {
            return null;
        }
        int exitVal = 0;
        try {
            exitVal = p.waitFor();
        } catch (InterruptedException e) {
            return null;
        }
        if (exitVal != 0)
            return null;
        String line = null;
        BufferedReader br = null;
        try {
            InputStreamReader isr = new InputStreamReader(p.getInputStream());
            br = new BufferedReader(isr);
            line = br.readLine();
            isr.close();
        } catch (IOException e) {
            return null;
        } finally {
            if (br != null) try {br.close();} catch (Exception e) {}
        }
        return line;
    }

    static File localTempDir = null;
    static {
        File f;
        boolean success = true;
        try {
            f = File.createTempFile("pig", "tmp");
            success &= f.delete();
            success &= f.mkdir();
            localTempDir = f;
            localTempDir.deleteOnExit();
        } catch (IOException e) {
        }
        if (!success) {
          throw new RuntimeException("Error creating FileLocalizer temp directory.");
        }
    }

    public static class FetchFileRet {
        public FetchFileRet(File file, boolean didFetch) {
            this.file = file;
            this.didFetch = didFetch;
        }
        public File file;
        public boolean didFetch;
    }

    /**
     * Ensures that the passed path is on the local file system, fetching it
     * to the java.io.tmpdir if necessary. If pig.jars.relative.to.dfs is true
     * and dfs is not null, then a relative path is assumed to be relative to the passed
     * dfs active directory. Else they are assumed to be relative to the local working
     * directory.
     */
    public static FetchFileRet fetchFile(Properties properties, String filePath) throws IOException {
        return fetchFilesInternal(properties, filePath, false)[0];
    }

    /**
     * Ensures that the passed files pointed to by path are on the local file system,
     * fetching them to the java.io.tmpdir if necessary. If pig.jars.relative.to.dfs is true
     * and dfs is not null, then a relative path is assumed to be relative to the passed
     * dfs active directory. Else they are assumed to be relative to the local working
     * directory.
     */
    public static FetchFileRet[] fetchFiles(Properties properties, String filePath) throws IOException {
        return fetchFilesInternal(properties, filePath, true);
    }

    /**
     * Copies the files from remote to local filesystem.
     * When 'multipleFiles' is set the path could point to multiple files
     * through globs or a directory. In this case, return array contains multiple
     * files, otherwise a single file is returned.
     *
     * If pig.jars.relative.to.dfs is true then a relative path is assumed to be
     * relative to the default filesystem's active directory.
     * Else they are assumed to be relative to the local working directory.
     *
     * @param properties
     * @param filePath
     * @param multipleFiles
     * @return
     */
    private static FetchFileRet[] fetchFilesInternal(Properties properties,
                                            String filePath,
                                            boolean multipleFiles) throws IOException {

        Path path = new Path(filePath);
        if (path.getName().isEmpty()) {
            return new FetchFileRet[0];
        }
        URI uri = path.toUri();
        Configuration conf = new Configuration();
        ConfigurationUtil.mergeConf(conf, ConfigurationUtil.toConfiguration(properties));

        // if there is no schema or if the schema is "local", then it is
        // expected to be a local path.

        FileSystem localFs = FileSystem.getLocal(conf);
        FileSystem srcFs;
        if ( (!"true".equals(properties.getProperty("pig.jars.relative.to.dfs"))
                && uri.getScheme() == null )||
                // For Windows local files
                (uri.getScheme() == null && uri.getPath().matches("^/[A-Za-z]:.*")) ||
                (uri.getScheme() != null && uri.getScheme().equals("local"))
            ) {
            srcFs = localFs;
        } else {
            srcFs = path.getFileSystem(conf);
        }

        FileStatus[] files;

        if (multipleFiles) {
            files = srcFs.globStatus(path);
        } else {
            files = new FileStatus[]{ srcFs.getFileStatus(path) };
        }
        if (files == null || files.length == 0) {
            throw new ExecException("file '" + filePath + "' does not exist.", 101, PigException.INPUT);
        }

        FetchFileRet[] fetchFiles = new FetchFileRet[files.length];
        int idx = 0;

        for(FileStatus file : files) {
            // should throw an exception if this is not a file?

            String pathname = file.getPath().toUri().getPath();
            String filename = file.getPath().getName();

            if (srcFs == localFs) {
                fetchFiles[idx++] = new FetchFileRet(new File(pathname), false);
            } else {
                // fetch from remote:
                File dest = new File(localTempDir, filename);
                dest.deleteOnExit();
                try {
                    srcFs.copyToLocalFile(file.getPath(), new Path(dest.getAbsolutePath()));
                } catch (IOException e) {
                    throw new ExecException("Could not copy " + filePath + " to local destination " + dest, 101, PigException.INPUT, e);
                }
                fetchFiles[idx++] = new FetchFileRet(dest, true);
            }
        }

        return fetchFiles;
    }

    /**
     * Ensures that the passed resource is available from the local file system, fetching
     * it to a temporary directory.
     *
     * @throws ResourceNotFoundException
     */
    public static FetchFileRet fetchResource(String name) throws IOException, ResourceNotFoundException {
      FetchFileRet localFileRet = null;
      InputStream resourceStream = PigContext.getClassLoader().getResourceAsStream(name);
      if (resourceStream != null) {
        File dest = new File(localTempDir, name);
        dest.getParentFile().mkdirs();
        dest.deleteOnExit();

        OutputStream outputStream = null;
        try {
            outputStream = new BufferedOutputStream(new FileOutputStream(dest));
            byte[] buffer = new byte[1024];
            int len;
            while ((len=resourceStream.read(buffer)) > 0) {
              outputStream.write(buffer,0,len);
            }
        } finally {
            resourceStream.close();
            if (outputStream != null) {
                outputStream.close();
            }
        }
        localFileRet = new FetchFileRet(dest,false);
      }
      else
      {
        throw new ResourceNotFoundException(name);
      }

      return localFileRet;
    }
}