/*! ******************************************************************************
 *
 * Pentaho Data Integration
 *
 * Copyright (C) 2017 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.hadoop.shim.common.format;

import java.nio.file.FileAlreadyExistsException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.orc.TypeDescription;
import org.pentaho.hadoop.shim.api.format.IPentahoOrcOutputFormat;
import org.pentaho.hadoop.shim.api.format.SchemaDescription;

import org.apache.hadoop.mapreduce.Job;
import org.pentaho.hadoop.shim.common.ConfigurationProxy;
import org.pentaho.hadoop.shim.common.format.orc.OrcSchemaConverter;
import org.pentaho.hadoop.shim.common.format.orc.PentahoOrcRecordWriter;

/**
 * Created by tkafalas on 11/3/2017.
 */
public class PentahoOrcOutputFormat extends HadoopFormatBase implements IPentahoOrcOutputFormat {

  private Job job;
  private TypeDescription schema;
  private String outputFilename;
  private SchemaDescription schemaDescription;
  private Configuration conf;
  private COMPRESSION compression = COMPRESSION.NONE;
  private int compressSize = 0;
  private int stripeSize = DEFAULT_STRIPE_SIZE;
  private int rowIndexStride = 0;

  private static final Logger logger = Logger.getLogger( PentahoOrcOutputFormat.class );

  public PentahoOrcOutputFormat() throws Exception {
    conf = inClassloader( () -> {
      Configuration conf = new ConfigurationProxy();
      job = Job.getInstance( conf );
      conf.addResource( "hive-site.xml" );
      return conf;
    } );
  }

  @Override public IPentahoRecordWriter createRecordWriter() throws Exception {
    logger.info( "Initializing Orc Writer" );
    if ( schemaDescription == null ) {
      throw new Exception( "Invalid state.  The schemaDescription is null" );
    }
    if ( outputFilename == null ) {
      throw new Exception( "Invalid state.  The outputFileName is null" );
    }
    OrcSchemaConverter converter = new OrcSchemaConverter( );
    schema = converter.buildTypeDescription( schemaDescription );

    return new PentahoOrcRecordWriter( schemaDescription, schema, outputFilename, conf );
  }

  @Override public void setSchemaDescription( SchemaDescription schema ) throws Exception {
    schemaDescription = schema;
  }

  @Override public void setOutputFile( String file, boolean override ) throws Exception {
    inClassloader( () -> {
      Path outputFile = new Path( file );
      FileSystem fs = FileSystem.get( outputFile.toUri(), job.getConfiguration() );
      if ( fs.exists( outputFile ) ) {
        if ( override ) {
          fs.delete( outputFile, true );
        } else {
          throw new FileAlreadyExistsException( file );
        }
      }

      this.outputFilename = file;
    } );
  }

  @Override public void setCompression( COMPRESSION compression ) {
    this.compression = compression;
    conf.set( COMPRESSION_KEY, compression.toString() );
    if ( compression == COMPRESSION.NONE ) {
      compressSize = 0;
      conf.unset( COMPRESS_SIZE_KEY );
    } else if ( compressSize == 0 ) {
      compressSize = DEFAULT_COMPRESS_SIZE;
      conf.set( COMPRESS_SIZE_KEY, Integer.toString( DEFAULT_COMPRESS_SIZE ) );
    }
  }

  @Override public void setStripeSize( int megabytes ) {
    if ( stripeSize > 0 ) {
      stripeSize = megabytes;
      conf.set( STRIPE_SIZE_KEY, Integer.toString( 1024 * 1024 * stripeSize ) );
    }
  }

  @Override
  public void setRowIndexStride( int numRows ) {
    if ( numRows > 0 ) {
      rowIndexStride = numRows;
      conf.set( CREATE_INDEX_KEY, "true" );
      conf.set( ROW_INDEX_STRIDE_KEY, Integer.toString( 1024 * 1024 * rowIndexStride ) );
    } else if ( numRows == 0 ) {
      rowIndexStride = numRows;
      conf.set( CREATE_INDEX_KEY, "false" );
      conf.unset( ROW_INDEX_STRIDE_KEY );
    }
  }

  @Override
  public void setCompressSize( int kilobytes ) {
    if ( kilobytes > 0 ) {
      compressSize = kilobytes;
      conf.set( COMPRESS_SIZE_KEY, Integer.toString( 1024 * compressSize ) );
    } else if ( kilobytes == 0 ) {
      compressSize = kilobytes;
      compression = COMPRESSION.NONE;
      conf.unset( COMPRESS_SIZE_KEY );
      conf.set( COMPRESSION_KEY, compression.toString() );
    }
  }
}

@@ -2185,57 +2185,88 @@
     
-conf = inClassloader( () -%3E %7B%0A      Configuration
+Thread.currentThread().setContextClassLoader( getClass().getClassLoader() );%0A   
  con
@@ -2299,18 +2299,16 @@
 ();%0A    
-  
 job = Jo
@@ -2330,26 +2330,24 @@
 conf );%0A    
- 
- 
 conf.addReso
@@ -2375,36 +2375,8 @@
  );%0A
-      return conf;%0A    %7D );%0A
   %7D%0A
@@ -3189,37 +3189,8 @@
 n %7B%0A
-    inClassloader( () -%3E %7B%0A  
     
@@ -3221,26 +3221,24 @@
 th( file );%0A
-  
     FileSyst
@@ -3307,26 +3307,24 @@
 on() );%0A    
-  
 if ( fs.exis
@@ -3340,26 +3340,24 @@
 tFile ) ) %7B%0A
-  
       if ( o
@@ -3376,18 +3376,16 @@
         
-  
 fs.delet
@@ -3413,18 +3413,16 @@
 ;%0A      
-  
 %7D else %7B
@@ -3422,18 +3422,16 @@
  else %7B%0A
-  
         
@@ -3486,23 +3486,17 @@
     
- 
- 
 %7D%0A
-  
     %7D%0A%0A
-  
     
@@ -3527,17 +3527,8 @@
 le;%0A
-    %7D );%0A
   %7D%0A

Update the UI and shims to write orc output file