/*
 * Copyright © 2016 Cask Data, Inc.
 *
 * 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 co.cask.hydrator.common;

import co.cask.cdap.api.data.schema.Schema;
import co.cask.cdap.etl.api.PipelineConfigurer;
import com.google.common.base.Preconditions;

import java.io.IOException;

/**
 * Utility class for with methods for validating schema fields and its types,
 * validating fields are simple, checking if schema is subset of another, checking if fields are present in schema, etc
 */
public final class SchemaValidator {

  /**
   * Validate output schema fields and if input schema is present
   * check if output schema is a subset of the input schema and return output schema.
   * @param outputSchemaString output schema from config
   * @param rowKeyField row key field from config
   * @param pipelineConfigurer Pipelineconfigurer that's used to get input schema and set output schema.
   * @return Schema - output schema
   */
  public static Schema validateOutputSchemaAndInputSchemaIfPresent(String outputSchemaString, String rowKeyField,
                                                                 PipelineConfigurer pipelineConfigurer) {
    Schema outputSchema;
    // initialize output schema if presnet
    Preconditions.checkNotNull(outputSchemaString, "Output schema string is null");

    try {
      outputSchema = Schema.parseJson(outputSchemaString);
    } catch (IOException e) {
      throw new IllegalArgumentException("Unable to parse output schema : " + e.getMessage(), e);
    }

    Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema();

    // check that all fields in output schema are simple
    validateSchemaFieldsAreSimple(outputSchema);

    if (inputSchema != null) {
      // check if output schema is a subset of input schema and check if rowkey field is present in input schema
      validateFieldsArePresentInSchema(inputSchema, rowKeyField);
      validateOutputSchemaIsSubsetOfInputSchema(inputSchema, outputSchema);
    }
    return outputSchema;
  }

  /**
   * Checks that all the fields in output schema is part of input schema and the fields schema type matches.
   * @param inputSchema
   * @param outputSchema
   */
  public static void validateOutputSchemaIsSubsetOfInputSchema(Schema inputSchema, Schema outputSchema) {
    // check if input schema contains all the fields expected in the output schema
    for (Schema.Field field : outputSchema.getFields()) {
      if (inputSchema.getField(field.getName()) == null) {
        throw new IllegalArgumentException(
          String.format("Field '%s' is present in output schema but not present in input schema",
                        field.getName()));
      } else if (!inputSchema.getField(field.getName()).getSchema().equals(field.getSchema())) {
        throw new IllegalArgumentException(
          String.format("Field type mismatch, field '%s' type in input schema is %s, " +
                          "while in output schema its of type %s", field.getName(),
                        inputSchema.getField(field.getName()).getSchema(), field.getSchema()));
      }
    }
  }

  /**
   * Iterates through the schema fields and checks their type is simple
   * @param schema
   */
  public static void validateSchemaFieldsAreSimple(Schema schema) {
    for (Schema.Field field : schema.getFields()) {
      // simple type check for fields
      if (!field.getSchema().isSimpleOrNullableSimple()) {
        throw new IllegalArgumentException(
          String.format("Field '%s' is not of simple type, All fields for table sink should of simple type",
                        field.getName()));
      }
    }
  }

  /**
   * Iterates through the required fields and checks if they are present in the schema
   * @param schema
   * @param requiredFields
   */
  public static void validateFieldsArePresentInSchema(Schema schema, String... requiredFields) {
    for (String field : requiredFields) {
      if (schema.getField(field) == null) {
        throw new IllegalArgumentException(
          String.format("Field : '%s' is not present in the input schema",  field));
      }
    }
  }

  private SchemaValidator() {
    throw new AssertionError("Suppress default constructor for non-instantiability");
  }
}

@@ -790,16 +790,50 @@
 ception;
+%0Aimport javax.annotation.Nullable;
 %0A%0A/**%0A *
@@ -1141,16 +1141,17 @@
  present
+,
 %0A   * ch
@@ -1475,32 +1475,44 @@
 ut schema%0A   */%0A
+  @Nullable%0A
   public static 
@@ -1674,16 +1674,18 @@
         
+  
 Pipeline
@@ -1720,35 +1720,91 @@
 r) %7B%0A    Schema 
-out
+inputSchema = pipelineConfigurer.getStageConfigurer().getIn
 putSchema;%0A    /
@@ -1800,138 +1800,290 @@
 hema
+()
 ;%0A    
-// initialize output schema if presnet%0A    Preconditions.checkNotNull(outputSchemaString, %22O
+if (inputSchema == null && outputSchemaString == null) %7B%0A       return null;%0A    %7D%0A%0A    // initialize output schema if present; otherwise, set it to input schema%0A    Schema outputSchema;%0A    if (outputSchemaString == null) %7B%0A      o
 utput
- s
+S
 chema 
-string is null%22);%0A%0A
+= inputSchema;%0A    %7D else %7B%0A  
     
@@ -2088,16 +2088,18 @@
   try %7B%0A
+  
       ou
@@ -2153,16 +2153,18 @@
 g);%0A    
+  
 %7D catch 
@@ -2177,24 +2177,26 @@
 eption e) %7B%0A
+  
       throw 
@@ -2289,93 +2289,17 @@
     
+  
 %7D
-%0A
 %0A    
-Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema();
+%7D
 %0A%0A  
@@ -2401,16 +2401,81 @@
 Schema);
+%0A    validateFieldsArePresentInSchema(outputSchema, rowKeyField);
 %0A%0A    if
@@ -2616,74 +2616,8 @@
 ema%0A
-      validateFieldsArePresentInSchema(inputSchema, rowKeyField);%0A
     

Allow for null output schema.