content: /*
   * 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.carbondata.processing.loading.model;
  
  import java.io.IOException;
  import java.text.SimpleDateFormat;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.carbondata.common.Maps;
  import org.apache.carbondata.common.Strings;
  import org.apache.carbondata.common.annotations.InterfaceAudience;
  import org.apache.carbondata.common.constants.LoggerAction;
  import org.apache.carbondata.common.exceptions.sql.InvalidLoadOptionException;
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.constants.CarbonLoadOptionConstants;
  import org.apache.carbondata.core.datastore.compression.CompressorFactory;
  import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
  import org.apache.carbondata.core.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.DataLoadMetrics;
  import org.apache.carbondata.processing.loading.constants.DataLoadProcessorConstants;
  import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat;
  import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
  import org.apache.carbondata.processing.util.CarbonBadRecordUtil;
  import org.apache.carbondata.processing.util.CarbonLoaderUtil;
  import org.apache.carbondata.processing.util.TableOptionConstant;
  
  import org.apache.commons.lang.StringUtils;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.log4j.Logger;
  
  /**
   * Builder for {@link CarbonLoadModel}
   */
  @InterfaceAudience.Internal
  public class CarbonLoadModelBuilder {
    private static final Logger LOGGER = LogServiceFactory.getLogService(
        CarbonLoadModelBuilder.class.getName());
    private CarbonTable table;
  
    public CarbonLoadModelBuilder(CarbonTable table) {
      this.table = table;
    }
  
    /**
     * build CarbonLoadModel for data loading
     * @param options Load options from user input
     * @param taskNo
     * @return a new CarbonLoadModel instance
     */
    public CarbonLoadModel build(Map<String, String>  options, long timestamp, String taskNo)
        throws InvalidLoadOptionException, IOException {
      Map<String, String> optionsFinal = LoadOption.fillOptionWithDefaultValue(options);
      Map<String, String> tableProperties = table.getTableInfo().getFactTable().getTableProperties();
  
      if (!options.containsKey("fileheader")) {
        List<CarbonColumn> csvHeader = table.getCreateOrderColumn();
        String[] columns = new String[csvHeader.size()];
        for (int i = 0; i < columns.length; i++) {
          columns[i] = csvHeader.get(i).getColName();
        }
        optionsFinal.put("fileheader", Strings.mkString(columns, ","));
      } else {
        optionsFinal.put("fileheader", options.get("fileheader"));
      }
      optionsFinal.put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(options, table));
      optionsFinal.put("sort_scope",
          Maps.getOrDefault(options, "sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT));
      CarbonLoadModel model = new CarbonLoadModel();
      model.setCarbonTransactionalTable(table.isTransactionalTable());
      model.setFactTimeStamp(timestamp);
      model.setTaskNo(taskNo);
  
      // we have provided 'fileheader', so it hadoopConf can be null
      build(options, optionsFinal, model, null);
      String timestampFormat = options.get("timestampformat");
      // If TIMESTAMPFORMAT is not present in load options, check from table properties.
      if (timestampFormat == null) {
        timestampFormat = CarbonProperties.getInstance()
            .getProperty(CarbonLoadOptionConstants.CARBON_OPTIONS_TIMESTAMPFORMAT,
                Maps.getOrDefault(tableProperties, "timestampformat", CarbonProperties.getInstance()
                    .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
                        CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)));
      }
      String dateFormat = options.get("dateFormat");
      // If DATEFORMAT is not present in load options, check from table properties.
      if (dateFormat == null) {
        dateFormat = CarbonProperties.getInstance()
            .getProperty(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT,
                Maps.getOrDefault(tableProperties, "dateformat", CarbonProperties.getInstance()
                    .getProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
                        CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)));
      }
      model.setDateFormat(dateFormat);
      model.setTimestampFormat(timestampFormat);
      validateAndSetColumnCompressor(model);
      validateAndSetBinaryDecoder(model);
      return model;
    }
  
    /**
     * build CarbonLoadModel for data loading
     * @param options Load options from user input
     * @param optionsFinal Load options that populated with default values for optional options
     * @param carbonLoadModel The output load model
     * @param hadoopConf hadoopConf is needed to read CSV header if there 'fileheader' is not set in
     *                   user provided load options
     */
    public void build(
        Map<String, String> options,
        Map<String, String> optionsFinal,
        CarbonLoadModel carbonLoadModel,
        Configuration hadoopConf) throws InvalidLoadOptionException, IOException {
      build(options, optionsFinal, carbonLoadModel, hadoopConf, new HashMap<String, String>(), false);
    }
  
    /**
     * build CarbonLoadModel for data loading
     * @param options Load options from user input
     * @param optionsFinal Load options that populated with default values for optional options
     * @param carbonLoadModel The output load model
     * @param hadoopConf hadoopConf is needed to read CSV header if there 'fileheader' is not set in
     *                   user provided load options
     * @param partitions partition name map to path
     * @param isDataFrame true if build for load for dataframe
     */
    public void build(
        Map<String, String> options,
        Map<String, String> optionsFinal,
        CarbonLoadModel carbonLoadModel,
        Configuration hadoopConf,
        Map<String, String> partitions,
        boolean isDataFrame) throws InvalidLoadOptionException, IOException {
      carbonLoadModel.setTableName(table.getTableName());
      carbonLoadModel.setDatabaseName(table.getDatabaseName());
      carbonLoadModel.setTablePath(table.getTablePath());
      carbonLoadModel.setTableName(table.getTableName());
      carbonLoadModel.setCarbonTransactionalTable(table.isTransactionalTable());
      CarbonDataLoadSchema dataLoadSchema = new CarbonDataLoadSchema(table);
      // Need to fill dimension relation
      carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema);
      String sortScope = optionsFinal.get("sort_scope");
      String badRecordsLoggerEnable = optionsFinal.get("bad_records_logger_enable");
      String badRecordsAction = optionsFinal.get("bad_records_action");
      String badRecordPath = optionsFinal.get("bad_record_path");
      String globalSortPartitions = optionsFinal.get("global_sort_partitions");
      String timestampformat = optionsFinal.get("timestampformat");
      String dateFormat = optionsFinal.get("dateformat");
      String delimiter = optionsFinal.get("delimiter");
      String complexDelimiterLevel1 = optionsFinal.get("complex_delimiter_level_1");
      String complexDelimiterLevel2 = optionsFinal.get("complex_delimiter_level_2");
      String complexDelimiterLevel3 = optionsFinal.get("complex_delimiter_level_3");
      String complexDelimiterLevel4 = optionsFinal.get("complex_delimiter_level_4");
      validateDateTimeFormat(timestampformat, "TimestampFormat");
      validateDateTimeFormat(dateFormat, "DateFormat");
  
      if (Boolean.parseBoolean(badRecordsLoggerEnable) ||
          LoggerAction.REDIRECT.name().equalsIgnoreCase(badRecordsAction)) {
        if (!StringUtils.isEmpty(badRecordPath)) {
          badRecordPath = CarbonUtil.checkAndAppendHDFSUrl(badRecordPath);
        } else {
          throw new InvalidLoadOptionException(
              "Cannot redirect bad records as bad record location is not provided.");
        }
      }
  
      carbonLoadModel.setBadRecordsLocation(badRecordPath);
  
      validateGlobalSortPartitions(globalSortPartitions);
      carbonLoadModel.setEscapeChar(checkDefaultValue(optionsFinal.get("escapechar"), "\\"));
      carbonLoadModel.setQuoteChar(
          CarbonUtil.unescapeChar(checkDefaultValue(optionsFinal.get("quotechar"), "\"")));
      carbonLoadModel.setCommentChar(checkDefaultValue(optionsFinal.get("commentchar"), "#"));
      String lineSeparator = CarbonUtil.unescapeChar(options.get("line_separator"));
      if (lineSeparator != null) {
        carbonLoadModel.setLineSeparator(lineSeparator);
      }
  
      // if there isn't file header in csv file and load sql doesn't provide FILEHEADER option,
      // we should use table schema to generate file header.
      String fileHeader = optionsFinal.get("fileheader");
      String headerOption = optionsFinal.get("header");
      if (StringUtils.isNotEmpty(headerOption)) {
        if (!headerOption.equalsIgnoreCase("true") &&
            !headerOption.equalsIgnoreCase("false")) {
          throw new InvalidLoadOptionException(
              "'header' option should be either 'true' or 'false'.");
        }
        // whether the csv file has file header, the default value is true
        if (Boolean.valueOf(headerOption)) {
          if (!StringUtils.isEmpty(fileHeader)) {
            throw new InvalidLoadOptionException(
                "When 'header' option is true, 'fileheader' option is not required.");
          }
        } else {
          if (StringUtils.isEmpty(fileHeader)) {
            List<CarbonColumn> columns = table.getCreateOrderColumn();
            List<String> columnNames = new ArrayList<>();
            List<String> partitionColumns = new ArrayList<>();
            for (int i = 0; i < columns.size(); i++) {
              columnNames.add(columns.get(i).getColName());
            }
            columnNames.addAll(partitionColumns);
            fileHeader = Strings.mkString(columnNames.toArray(new String[columnNames.size()]), ",");
          }
        }
      }
  
      String binaryDecoder = options.get("binary_decoder");
      carbonLoadModel.setBinaryDecoder(binaryDecoder);
      carbonLoadModel.setTimestampFormat(timestampformat);
      carbonLoadModel.setDateFormat(dateFormat);
      carbonLoadModel.setDefaultTimestampFormat(
          CarbonProperties.getInstance().getProperty(
              CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
              CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT));
  
      carbonLoadModel.setDefaultDateFormat(
          CarbonProperties.getInstance().getProperty(
              CarbonCommonConstants.CARBON_DATE_FORMAT,
              CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT));
  
      carbonLoadModel.setSerializationNullFormat(
          TableOptionConstant.SERIALIZATION_NULL_FORMAT.getName() + "," +
              optionsFinal.get("serialization_null_format"));
  
      carbonLoadModel.setBadRecordsLoggerEnable(
          TableOptionConstant.BAD_RECORDS_LOGGER_ENABLE.getName() + "," + badRecordsLoggerEnable);
  
      carbonLoadModel.setBadRecordsAction(
          TableOptionConstant.BAD_RECORDS_ACTION.getName() + "," + badRecordsAction.toUpperCase());
  
      carbonLoadModel.setIsEmptyDataBadRecord(
          DataLoadProcessorConstants.IS_EMPTY_DATA_BAD_RECORD + "," +
              optionsFinal.get("is_empty_data_bad_record"));
  
      carbonLoadModel.setSkipEmptyLine(optionsFinal.get("skip_empty_line"));
  
      carbonLoadModel.setSortScope(sortScope);
      if (globalSortPartitions == null) {
        globalSortPartitions = table.getGlobalSortPartitions();
      }
      carbonLoadModel.setGlobalSortPartitions(globalSortPartitions);
  
      if (delimiter.equalsIgnoreCase(complexDelimiterLevel1) ||
          complexDelimiterLevel1.equalsIgnoreCase(complexDelimiterLevel2) ||
          delimiter.equalsIgnoreCase(complexDelimiterLevel2) ||
          delimiter.equalsIgnoreCase(complexDelimiterLevel3)) {
        throw new InvalidLoadOptionException("Field Delimiter and Complex types delimiter are same");
      } else {
        carbonLoadModel.setComplexDelimiter(complexDelimiterLevel1);
        carbonLoadModel.setComplexDelimiter(complexDelimiterLevel2);
        carbonLoadModel.setComplexDelimiter(complexDelimiterLevel3);
        carbonLoadModel.setComplexDelimiter(complexDelimiterLevel4);
      }
      carbonLoadModel.setCsvDelimiter(CarbonUtil.unescapeChar(delimiter));
      carbonLoadModel.setCsvHeader(fileHeader);
  
      List<String> ignoreColumns = new ArrayList<>();
      if (!isDataFrame) {
        for (Map.Entry<String, String> partition : partitions.entrySet()) {
          if (partition.getValue() != null) {
            ignoreColumns.add(partition.getKey());
          }
        }
      }
  
      carbonLoadModel.setCsvHeaderColumns(
          LoadOption.getCsvHeaderColumns(carbonLoadModel, hadoopConf, ignoreColumns));
  
      int validatedMaxColumns = validateMaxColumns(
          carbonLoadModel.getCsvHeaderColumns(),
          optionsFinal.get("maxcolumns"));
  
      carbonLoadModel.setMaxColumns(String.valueOf(validatedMaxColumns));
      if (carbonLoadModel.isCarbonTransactionalTable()) {
        carbonLoadModel.readAndSetLoadMetadataDetails();
      }
      carbonLoadModel.setSortColumnsBoundsStr(optionsFinal.get("sort_column_bounds"));
      carbonLoadModel.setLoadMinSize(
          optionsFinal.get(CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB));
      validateAndSetLoadMinSize(carbonLoadModel);
  
      validateAndSetColumnCompressor(carbonLoadModel);
      validateAndSetBinaryDecoder(carbonLoadModel);
  
      validateRangeColumn(optionsFinal, carbonLoadModel);
  
      carbonLoadModel.setMetrics(new DataLoadMetrics());
    }
  
    private void validateRangeColumn(Map<String, String> optionsFinal,
        CarbonLoadModel carbonLoadModel) throws InvalidLoadOptionException {
      String scaleFactor = optionsFinal.get("scale_factor");
      if (scaleFactor != null) {
        try {
          int scale = Integer.parseInt(scaleFactor);
          if (scale < 1 || scale > 300) {
            throw new InvalidLoadOptionException(
                "Invalid scale_factor option, the range of scale_factor should be [1, 300]");
          }
          carbonLoadModel.setScaleFactor(scale);
        } catch (NumberFormatException ex) {
          throw new InvalidLoadOptionException(
              "Invalid scale_factor option, scale_factor should be a integer");
        }
      }
    }
  
    private int validateMaxColumns(String[] csvHeaders, String maxColumns)
        throws InvalidLoadOptionException {
      /*
      User configures both csvheadercolumns, maxcolumns,
        if csvheadercolumns >= maxcolumns, give error
        if maxcolumns > threashold, give error
      User configures csvheadercolumns
        if csvheadercolumns >= maxcolumns(default) then maxcolumns = csvheadercolumns+1
        if csvheadercolumns >= threashold, give error
      User configures nothing
        if csvheadercolumns >= maxcolumns(default) then maxcolumns = csvheadercolumns+1
        if csvheadercolumns >= threashold, give error
       */
      int columnCountInSchema = csvHeaders.length;
      int maxNumberOfColumnsForParsing = 0;
      Integer maxColumnsInt = getMaxColumnValue(maxColumns);
      if (maxColumnsInt != null) {
        if (columnCountInSchema >= maxColumnsInt) {
          throw new InvalidLoadOptionException(
              "csv headers should be less than the max columns " + maxColumnsInt);
        } else if (maxColumnsInt > CSVInputFormat.THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING) {
          throw new InvalidLoadOptionException(
              "max columns cannot be greater than the threshold value: " +
                  CSVInputFormat.THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING);
        } else {
          maxNumberOfColumnsForParsing = maxColumnsInt;
        }
      } else if (columnCountInSchema >= CSVInputFormat.THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING) {
        throw new InvalidLoadOptionException(
            "csv header columns should be less than max threashold: " +
                CSVInputFormat.THRESHOLD_MAX_NUMBER_OF_COLUMNS_FOR_PARSING);
      } else if (columnCountInSchema >= CSVInputFormat.DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING) {
        maxNumberOfColumnsForParsing = columnCountInSchema + 1;
      } else {
        maxNumberOfColumnsForParsing = CSVInputFormat.DEFAULT_MAX_NUMBER_OF_COLUMNS_FOR_PARSING;
      }
      return maxNumberOfColumnsForParsing;
    }
  
    private Integer getMaxColumnValue(String maxColumn) {
      return (maxColumn == null) ? null : Integer.parseInt(maxColumn);
    }
  
    /**
     * validates both timestamp and date for illegal values
     */
    private void validateDateTimeFormat(String dateTimeLoadFormat, String dateTimeLoadOption)
        throws InvalidLoadOptionException {
      // allowing empty value to be configured for dateformat option.
      if (dateTimeLoadFormat != null && !dateTimeLoadFormat.trim().equalsIgnoreCase("")) {
        try {
          new SimpleDateFormat(dateTimeLoadFormat);
        } catch (IllegalArgumentException e) {
          throw new InvalidLoadOptionException(
              "Error: Wrong option: " + dateTimeLoadFormat + " is provided for option "
                  + dateTimeLoadOption);
        }
      }
    }
  
    private void validateGlobalSortPartitions(String globalSortPartitions)
        throws InvalidLoadOptionException {
      if (globalSortPartitions != null) {
        try {
          int num = Integer.parseInt(globalSortPartitions);
          if (num <= 0) {
            throw new InvalidLoadOptionException("'GLOBAL_SORT_PARTITIONS' should be greater than 0");
          }
        } catch (NumberFormatException e) {
          throw new InvalidLoadOptionException(e.getMessage());
        }
      }
    }
  
    private void validateAndSetColumnCompressor(CarbonLoadModel carbonLoadModel)
        throws InvalidLoadOptionException {
      try {
        String columnCompressor = carbonLoadModel.getColumnCompressor();
        if (StringUtils.isBlank(columnCompressor)) {
          columnCompressor = CarbonProperties.getInstance().getProperty(
              CarbonCommonConstants.COMPRESSOR, CarbonCommonConstants.DEFAULT_COMPRESSOR);
        }
        // check and load compressor
        CompressorFactory.getInstance().getCompressor(columnCompressor);
        carbonLoadModel.setColumnCompressor(columnCompressor);
      } catch (Exception e) {
        LOGGER.error(e.getMessage(), e);
        throw new InvalidLoadOptionException("Failed to load the compressor");
      }
    }
  
    private void validateAndSetBinaryDecoder(CarbonLoadModel carbonLoadModel) {
      String binaryDecoder = carbonLoadModel.getBinaryDecoder();
      if (!CarbonLoaderUtil.isValidBinaryDecoder(binaryDecoder)) {
        throw new CarbonDataLoadingException("Binary decoder only support Base64, " +
            "Hex or no decode for string, don't support " + binaryDecoder);
      }
      if (StringUtils.isBlank(binaryDecoder)) {
        binaryDecoder = CarbonProperties.getInstance().getProperty(
            CarbonLoadOptionConstants.CARBON_OPTIONS_BINARY_DECODER,
            CarbonLoadOptionConstants.CARBON_OPTIONS_BINARY_DECODER_DEFAULT);
      }
      // check and load binary decoder
      carbonLoadModel.setBinaryDecoder(binaryDecoder);
    }
  
    /**
     * check whether using default value or not
     */
    private String checkDefaultValue(String value, String defaultValue) {
      if (StringUtils.isEmpty(value)) {
        return defaultValue;
      } else {
        return value;
      }
    }
  
    private void validateAndSetLoadMinSize(CarbonLoadModel carbonLoadModel) {
      int size = 0;
      String loadMinSize = carbonLoadModel.getLoadMinSize();
      try {
        size = Integer.parseInt(loadMinSize);
      } catch (Exception e) {
        size = 0;
      }
      // if the value is negative, set the value is 0
      if (size > 0) {
        carbonLoadModel.setLoadMinSize(loadMinSize);
      } else {
        carbonLoadModel.setLoadMinSize(CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB_DEFAULT);
      }
    }
  }
  