/*! ******************************************************************************
   *
   * Pentaho Data Integration
   *
   * Copyright (C) 2020 by Hitachi Vantara : http://www.pentaho.com
   *
   *******************************************************************************
   *
   * 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 org.pentaho.big.data.kettle.plugins.formats.impl.output;

import org.apache.commons.io.IOUtils;
import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileSystemException;
import org.pentaho.di.core.exception.KettleFileException;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.core.variables.VariableSpace;
import org.pentaho.di.core.vfs.AliasedFileObject;
import org.pentaho.di.core.vfs.KettleVFS;
import org.pentaho.hadoop.shim.api.format.IPvfsAliasGenerator;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.FileAlreadyExistsException;

/**
 * Logic to use a temporary file for output and then copy that file to some VFS/PVFS scheme that wasn't original
 * supoorted for the output content.
 */
public class PvfsFileAliaser {
  private String finalFilePath;

  private String temporaryFilePath;

  private VariableSpace variableSpace;

  private IPvfsAliasGenerator aliasGenerator;

  private boolean isOverwriteOutput;

  private LogChannelInterface log;

  public PvfsFileAliaser( String finalFilePath, VariableSpace variableSpace, IPvfsAliasGenerator aliasGenerator,
                          boolean isOverwriteOutput, LogChannelInterface log ) {
    this.finalFilePath = finalFilePath;
    this.variableSpace = variableSpace;
    this.aliasGenerator = aliasGenerator;
    this.isOverwriteOutput = isOverwriteOutput;
    this.log = log;
  }

  public String generateAlias() throws KettleFileException, FileSystemException, FileAlreadyExistsException {

    FileObject pvfsFileObject = KettleVFS.getFileObject( finalFilePath, variableSpace );
    if ( AliasedFileObject.isAliasedFile( pvfsFileObject ) ) {
      finalFilePath = ( (AliasedFileObject) pvfsFileObject ).getOriginalURIString();
    }
    //See if we need to use a another URI because the HadoopFileSystem is not supported for this URL.
    String aliasedFile = aliasGenerator.generateAlias( finalFilePath );
    temporaryFilePath = finalFilePath;
    if ( aliasedFile != null ) {
      if ( pvfsFileObject.exists() ) {
        if ( isOverwriteOutput ) {
          pvfsFileObject.delete();
        } else {
          throw new FileAlreadyExistsException( temporaryFilePath );
        }
      }
      temporaryFilePath = aliasedFile;  //set the outputFile to the temporary alias file
    }
    return temporaryFilePath;
  }

  public void copyFileToFinalDestination() throws KettleFileException, IOException {
    if ( aliasingIsActive() ) {
      FileObject srcFile = KettleVFS.getFileObject( temporaryFilePath, variableSpace );
      FileObject destFile = KettleVFS.getFileObject( finalFilePath, variableSpace );
      try ( InputStream in = KettleVFS.getInputStream( srcFile );
            OutputStream out = KettleVFS.getOutputStream( destFile, false ) ) {
        IOUtils.copy( in, out );
      }
    }
  }

  public void deleteTempFileAndFolder() {
    try {
      if ( aliasingIsActive() ) {
        FileObject srcFile = KettleVFS.getFileObject( temporaryFilePath, variableSpace );
        srcFile.getParent().deleteAll();
      }
    } catch ( FileSystemException | KettleFileException e ) {
      log.logError( e.getMessage(), e );
    }
  }

  private boolean aliasingIsActive() {
    return !finalFilePath.equals( temporaryFilePath ) && temporaryFilePath != null;
  }
}

@@ -77,22 +77,18 @@
 ******%0A 
-  *%0A  
+*%0A
  * Penta
@@ -108,22 +108,18 @@
 ration%0A 
-  *%0A  
+*%0A
  * Copyr
@@ -176,23 +176,19 @@
 aho.com%0A
-  
  *%0A
-  
  *******
@@ -260,23 +260,19 @@
 *******%0A
-  
  *%0A
-  
  * Licen
@@ -330,18 +330,16 @@
 ense%22);%0A
-  
  * you m
@@ -385,18 +385,16 @@
 ce with%0A
-  
  * the L
@@ -442,22 +442,18 @@
 nse at%0A 
-  *%0A  
+*%0A
  *    ht
@@ -494,22 +494,18 @@
 SE-2.0%0A 
-  *%0A  
+*%0A
  * Unles
@@ -567,18 +567,16 @@
 oftware%0A
-  
  * distr
@@ -636,18 +636,16 @@
  BASIS,%0A
-  
  * WITHO
@@ -712,18 +712,16 @@
 mplied.%0A
-  
  * See t
@@ -783,18 +783,16 @@
 ons and%0A
-  
  * limit
@@ -822,14 +822,10 @@
 e.%0A 
-  *%0A  
+*%0A
  ***
@@ -4257,16 +4257,231 @@
  != null
+ && !s3nSwitchedTos3a();%0A  %7D%0A%0A  private boolean s3nSwitchedTos3a() %7B%0A    return finalFilePath != null && temporaryFilePath != null && finalFilePath.startsWith( %22s3n%22 ) && temporaryFilePath%0A      .startsWith( %22s3a%22 )
 ;%0A  %7D%0A%7D%0A

Add default Pentaho VFS S3 support to Parquet and Orc steps.