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.util.ArrayList;
  import java.util.HashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.carbondata.common.Maps;
  import org.apache.carbondata.common.annotations.InterfaceAudience;
  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.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.processing.loading.ComplexDelimitersEnum;
  import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
  import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
  import org.apache.carbondata.processing.util.CarbonLoaderUtil;
  
  import org.apache.commons.lang.StringUtils;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.log4j.Logger;
  
  /**
   * Provide utilities to populate loading options
   */
  @InterfaceAudience.Internal
  public class LoadOption {
  
    private static final Logger LOG = LogServiceFactory.getLogService(LoadOption.class.getName());
  
    /**
     * Based on the input options, fill and return data loading options with default value
     */
    public static Map<String, String> fillOptionWithDefaultValue(Map<String, String> options)
        throws InvalidLoadOptionException {
      Map<String, String> optionsFinal = new HashMap<>();
      optionsFinal.put("delimiter", Maps.getOrDefault(options, "delimiter", ","));
      optionsFinal.put("quotechar", Maps.getOrDefault(options, "quotechar", "\""));
      optionsFinal.put("fileheader", Maps.getOrDefault(options, "fileheader", ""));
      optionsFinal.put("commentchar", Maps.getOrDefault(options, "commentchar", "#"));
      String headerOption = options.get("header");
      if (headerOption != null) {
        if (!headerOption.equalsIgnoreCase("true") &&
            !headerOption.equalsIgnoreCase("false")) {
          throw new InvalidLoadOptionException(
              "'header' option should be either 'true' or 'false'.");
        }
      }
      optionsFinal.put("header", Maps.getOrDefault(options, "header", ""));
      optionsFinal.put(
          "escapechar",
          CarbonLoaderUtil.getEscapeChar(Maps.getOrDefault(options, "escapechar", "\\")));
  
      optionsFinal.put(
          "serialization_null_format",
          Maps.getOrDefault(options, "serialization_null_format", "\\N"));
  
      optionsFinal.put(
          "bad_records_logger_enable",
          Maps.getOrDefault(
              options,
              "bad_records_logger_enable",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE,
                  CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE_DEFAULT)));
  
      String badRecordActionValue = CarbonProperties.getInstance().getProperty(
          CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION,
          CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION_DEFAULT);
  
      optionsFinal.put(
          "bad_records_action",
          Maps.getOrDefault(
              options,
              "bad_records_action",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION,
                  badRecordActionValue)));
  
      optionsFinal.put(
          "is_empty_data_bad_record",
          Maps.getOrDefault(
              options,
              "is_empty_data_bad_record",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD,
                  CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD_DEFAULT)));
  
      optionsFinal.put(
          "skip_empty_line",
          Maps.getOrDefault(
              options,
              "skip_empty_line",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_SKIP_EMPTY_LINE)));
  
      optionsFinal.put("complex_delimiter_level_1",
          Maps.getOrDefault(options, "complex_delimiter_level_1",
              CarbonProperties.getInstance().getProperty(
                  CarbonCommonConstants.COMPLEX_DELIMITERS_LEVEL_1,
                  ComplexDelimitersEnum.COMPLEX_DELIMITERS_LEVEL_1.value())));
  
      optionsFinal.put("complex_delimiter_level_2",
          Maps.getOrDefault(options, "complex_delimiter_level_2",
              CarbonProperties.getInstance().getProperty(
                  CarbonCommonConstants.COMPLEX_DELIMITERS_LEVEL_2,
                  ComplexDelimitersEnum.COMPLEX_DELIMITERS_LEVEL_2.value())));
  
      optionsFinal.put("complex_delimiter_level_3",
          Maps.getOrDefault(options, "complex_delimiter_level_3",
              CarbonProperties.getInstance().getProperty(
                  CarbonCommonConstants.COMPLEX_DELIMITERS_LEVEL_3,
                  ComplexDelimitersEnum.COMPLEX_DELIMITERS_LEVEL_3.value())));
  
      optionsFinal.put("complex_delimiter_level_4",
          Maps.getOrDefault(options, "complex_delimiter_level_4",
              CarbonProperties.getInstance().getProperty(
                  CarbonCommonConstants.COMPLEX_DELIMITERS_LEVEL_4,
                  ComplexDelimitersEnum.COMPLEX_DELIMITERS_LEVEL_4.value())));
  
      optionsFinal.put(
          "dateformat",
          Maps.getOrDefault(
              options,
              "dateformat",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT,
                  CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT_DEFAULT)));
  
      optionsFinal.put(
          "timestampformat",
          Maps.getOrDefault(
              options,
              "timestampformat",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_TIMESTAMPFORMAT,
                  CarbonLoadOptionConstants.CARBON_OPTIONS_TIMESTAMPFORMAT_DEFAULT)));
  
      optionsFinal.put(
          "global_sort_partitions",
          Maps.getOrDefault(
              options,
              "global_sort_partitions",
              CarbonProperties.getInstance().getProperty(
                  CarbonLoadOptionConstants.CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS,
                  null)));
  
      optionsFinal.put("maxcolumns", Maps.getOrDefault(options, "maxcolumns", null));
      optionsFinal.put("sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT);
      optionsFinal.put("sort_column_bounds", Maps.getOrDefault(options, "sort_column_bounds", ""));
      optionsFinal.put(CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB,
          Maps.getOrDefault(options, CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB,
              CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB_DEFAULT));
  
      optionsFinal.put("range_column", Maps.getOrDefault(options, "range_column", null));
      optionsFinal.put("scale_factor", Maps.getOrDefault(options, "scale_factor", null));
      return optionsFinal;
    }
  
    /**
     * Return CSV header field names
     */
    public static String[] getCsvHeaderColumns(
        CarbonLoadModel carbonLoadModel,
        Configuration hadoopConf) throws IOException {
      return getCsvHeaderColumns(carbonLoadModel, hadoopConf, new LinkedList<String>());
    }
  
    /**
     * Return CSV header field names, with partition column
     */
    public static String[] getCsvHeaderColumns(
        CarbonLoadModel carbonLoadModel,
        Configuration hadoopConf,
        List<String> staticPartitionCols) throws IOException {
      String delimiter;
      if (StringUtils.isEmpty(carbonLoadModel.getCsvDelimiter())) {
        delimiter = CarbonCommonConstants.COMMA;
      } else {
        delimiter = CarbonUtil.delimiterConverter(carbonLoadModel.getCsvDelimiter());
      }
      String csvFile = null;
      String csvHeader = carbonLoadModel.getCsvHeader();
      String[] csvColumns;
      if (StringUtils.isBlank(csvHeader)) {
        // read header from csv file
        csvFile = carbonLoadModel.getFactFilePath().split(",")[0];
        csvHeader = CarbonUtil.readHeader(csvFile, hadoopConf);
        if (StringUtils.isBlank(csvHeader)) {
          throw new CarbonDataLoadingException("First line of the csv is not valid.");
        }
        String[] headers = csvHeader.toLowerCase().split(delimiter);
        csvColumns = new String[headers.length];
        for (int i = 0; i < csvColumns.length; i++) {
          csvColumns[i] = headers[i].replaceAll("\"", "").trim();
        }
      } else {
        String[] headers = csvHeader.toLowerCase().split(CarbonCommonConstants.COMMA);
        csvColumns = new String[headers.length];
        for (int i = 0; i < csvColumns.length; i++) {
          csvColumns[i] = headers[i].trim();
        }
      }
  
      // In SDK flow, hadoopConf will always be null,
      // hence FileHeader check is not required for nontransactional table
      if (hadoopConf != null && !CarbonDataProcessorUtil
          .isHeaderValid(carbonLoadModel.getTableName(), csvColumns,
              carbonLoadModel.getCarbonDataLoadSchema(), staticPartitionCols)) {
        if (csvFile == null) {
          LOG.error("CSV header in DDL is not proper."
              + " Column names in schema and CSV header are not the same.");
          throw new CarbonDataLoadingException(
              "CSV header in DDL is not proper. Column names in schema and CSV header are "
                  + "not the same.");
        } else {
          LOG.error(
              "CSV header in input file is not proper. Column names in schema and csv header are not "
                  + "the same. Input file : " + CarbonUtil.removeAKSK(csvFile));
          throw new CarbonDataLoadingException(
              "CSV header in input file is not proper. Column names in schema and csv header are not "
                  + "the same. Input file : " + CarbonUtil.removeAKSK(csvFile));
        }
      }
  
      // In case of static partition columns just change the name of header if already exists as
      // we should not take the column from csv file and add them as new columns at the end.
      if (staticPartitionCols.size() > 0) {
        List<String> updatedColumns = new ArrayList<>();
        for (int i = 0; i < csvColumns.length; i++) {
          if (staticPartitionCols.contains(csvColumns[i])) {
            updatedColumns.add(csvColumns[i] + "1");
          } else {
            updatedColumns.add(csvColumns[i]);
          }
        }
        updatedColumns.addAll(staticPartitionCols);
        return updatedColumns.toArray(new String[updatedColumns.size()]);
      } else {
        return csvColumns;
      }
    }
  
  }
  