/*
 * Licensed 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 com.facebook.presto.raptor.filesystem;

import io.airlift.slice.XxHash64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.util.Progressable;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;

import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
import static java.util.Objects.requireNonNull;

/**
 * The class overrides some inefficient methods from local file system
 */
public final class RaptorLocalFileSystem
        extends RawLocalFileSystem
{
    public RaptorLocalFileSystem(Configuration configuration)
            throws IOException
    {
        initialize(getUri(), configuration);
    }

    @Override
    public FSDataOutputStream create(Path path, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress)
            throws IOException
    {
        if (exists(path) && !overwrite) {
            throw new IOException("file already exists: " + path);
        }
        Path parent = path.getParent();
        if ((parent != null) && !mkdirs(parent)) {
            throw new IOException("mkdirs failed to create " + parent.toString());
        }
        return new FSDataOutputStream(
                new BufferedOutputStream(new LocalFileOutputStream(pathToFile(path)), bufferSize),
                statistics);
    }

    @Override
    public boolean exists(Path file)
    {
        return pathToFile(file).exists();
    }

    @Override
    public boolean rename(Path from, Path to)
            throws IOException
    {
        Files.move(pathToFile(from).toPath(), pathToFile(to).toPath(), ATOMIC_MOVE);
        return true;
    }

    private static class LocalFileOutputStream
            extends OutputStream
    {
        private final byte[] oneByte = new byte[1];
        private final XxHash64 hash = new XxHash64();
        private final File file;
        private final FileOutputStream out;
        private boolean closed;

        private LocalFileOutputStream(File file)
                throws IOException
        {
            this.file = requireNonNull(file, "file is null");
            this.out = new FileOutputStream(file);
        }

        @Override
        public void close()
                throws IOException
        {
            if (closed) {
                return;
            }
            closed = true;

            flush();
            out.getFD().sync();
            out.close();

            // extremely paranoid code to detect a broken local file system
            try (InputStream in = new FileInputStream(file)) {
                if (hash.hash() != XxHash64.hash(in)) {
                    throw new IOException("File is corrupt after write");
                }
            }
        }

        @Override
        public void flush()
                throws IOException
        {
            out.flush();
        }

        @Override
        public void write(byte[] b, int off, int len)
                throws IOException
        {
            out.write(b, off, len);
            hash.update(b, off, len);
        }

        @SuppressWarnings("NumericCastThatLosesPrecision")
        @Override
        public void write(int b)
                throws IOException
        {
            oneByte[0] = (byte) (b & 0xFF);
            write(oneByte, 0, 1);
        }
    }
}

@@ -607,42 +607,8 @@
 m;%0A%0A
-import io.airlift.slice.XxHash64;%0A
 impo
@@ -652,56 +652,8 @@
 on;%0A
-import org.apache.hadoop.fs.FSDataOutputStream;%0A
 impo
@@ -734,259 +734,35 @@
 em;%0A
-import org.apache.hadoop.util.Progressable;%0A%0Aimport java.io.BufferedOutputStream;%0Aimport java.io.File;%0Aimport java.io.FileInputStream;%0Aimport java.io.FileOutputStream;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.OutputStream
+%0Aimport java.io.IOException
 ;%0Aim
@@ -851,56 +851,8 @@
 OVE;
-%0Aimport static java.util.Objects.requireNonNull;
 %0A%0A/*
@@ -1186,2633 +1186,291 @@
 lic 
-FSDataOutputStream create(Path path, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress)%0A            throws IOException%0A    %7B%0A        if (exists(path) && !overwrite) %7B%0A            throw new IOException(%22file already exists: %22 + path);%0A        %7D%0A        Path parent = path.getParent();%0A        if ((parent != null) && !mkdirs(parent)) %7B%0A            throw new IOException(%22mkdirs failed to create %22 + parent.toString());%0A        %7D%0A        return new FSDataOutputStream(%0A                new BufferedOutputStream(new LocalFileOutputStream(pathToFile(path)), bufferSize),%0A                statistics);%0A    %7D%0A%0A    @Override%0A    public boolean exists(Path file)%0A    %7B%0A        return pathToFile(file).exists();%0A    %7D%0A%0A    @Override%0A    public boolean rename(Path from, Path to)%0A            throws IOException%0A    %7B%0A        Files.move(pathToFile(from).toPath(), pathToFile(to).toPath(), ATOMIC_MOVE);%0A        return true;%0A    %7D%0A%0A    private static class LocalFileOutputStream%0A            extends OutputStream%0A    %7B%0A        private final byte%5B%5D oneByte = new byte%5B1%5D;%0A        private final XxHash64 hash = new XxHash64();%0A        private final File file;%0A        private final FileOutputStream out;%0A        private boolean closed;%0A%0A        private LocalFileOutputStream(File file)%0A                throws IOException%0A        %7B%0A            this.file = requireNonNull(file, %22file is null%22);%0A            this.out = new FileOutputStream(file);%0A        %7D%0A%0A        @Override%0A        public void close()%0A                throws IOException%0A        %7B%0A            if (closed) %7B%0A                return;%0A            %7D%0A            closed = true;%0A%0A            flush();%0A            out.getFD().sync();%0A            out.close();%0A%0A            // extremely paranoid code to detect a broken local file system%0A            try (InputStream in = new FileInputStream(file)) %7B%0A                if (hash.hash() != XxHash64.hash(in)) %7B%0A                    throw new IOException(%22File is corrupt after write%22);%0A                %7D%0A            %7D%0A        %7D%0A%0A        @Override%0A        public void flush()%0A                throws IOException%0A        %7B%0A            out.flush();%0A        %7D%0A%0A        @Override%0A        public void write(byte%5B%5D b, int off, int len)%0A                throws IOException%0A        %7B%0A            out.write(b, off, len);%0A            hash.update(b, off, len);%0A        %7D%0A%0A        @SuppressWarnings(%22NumericCastThatLosesPrecision%22)%0A        @Override%0A        public void write(int b)%0A                throws IOException%0A        %7B%0A            oneByte%5B0%5D = (byte) (b & 0xFF);%0A            write(oneByte, 0, 1);%0A        %7D
+boolean exists(Path file)%0A    %7B%0A        return pathToFile(file).exists();%0A    %7D%0A%0A    @Override%0A    public boolean rename(Path from, Path to)%0A            throws IOException%0A    %7B%0A        Files.move(pathToFile(from).toPath(), pathToFile(to).toPath(), ATOMIC_MOVE);%0A        return true;
 %0A   

Remove unused file input stream from Raptor local FS