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.sdk.file;
  
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  
  import org.apache.carbondata.common.exceptions.sql.InvalidLoadOptionException;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.metadata.datatype.DataType;
  import org.apache.carbondata.core.metadata.datatype.Field;
  import org.apache.carbondata.core.scan.expression.ColumnExpression;
  import org.apache.carbondata.core.scan.expression.Expression;
  import org.apache.carbondata.core.scan.expression.LiteralExpression;
  import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression;
  import org.apache.carbondata.core.scan.expression.logical.AndExpression;
  import org.apache.carbondata.core.scan.expression.logical.OrExpression;
  import org.apache.carbondata.hadoop.api.CarbonTableOutputFormat;
  import org.apache.carbondata.hadoop.internal.ObjectArrayWritable;
  
  import org.apache.commons.lang3.StringUtils;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.io.NullWritable;
  import org.apache.hadoop.mapreduce.RecordWriter;
  
  public class CarbonIUD {
  
    private final Configuration configuration;
    private final Map<String, Map<String, Set<String>>> filterColumnToValueMappingForDelete;
    private final Map<String, Map<String, Set<String>>> filterColumnToValueMappingForUpdate;
    private final Map<String, Map<String, String>> updateColumnToValueMapping;
  
    private CarbonIUD(Configuration conf) {
      configuration = conf;
      filterColumnToValueMappingForDelete = new HashMap<>();
      filterColumnToValueMappingForUpdate = new HashMap<>();
      updateColumnToValueMapping = new HashMap<>();
    }
  
    /**
     * @return CarbonIUD object
     */
    public static CarbonIUD getInstance(Configuration conf) {
      return new CarbonIUD(conf);
    }
  
    public static CarbonIUD getInstance() {
      return new CarbonIUD(null);
    }
  
    /**
     * @param path   is the table path on which delete is performed
     * @param column is the columnName on which records have to be deleted
     * @param value  of column on which the records have to be deleted
     * @return CarbonIUD object
     */
    public CarbonIUD delete(String path, String column, String value) {
      prepareDelete(path, column, value, filterColumnToValueMappingForDelete);
      return this;
    }
  
    /**
     * This method deletes the rows at given path by applying the filterExpression
     *
     * @param path             is the table path on which delete is performed
     * @param filterExpression is the expression to delete the records
     * @throws IOException
     * @throws InterruptedException
     */
    public void delete(String path, Expression filterExpression)
        throws IOException, InterruptedException {
      CarbonReader reader = CarbonReader.builder(path)
          .projection(new String[] { CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID })
          .withHadoopConf(configuration)
          .filter(filterExpression).build();
  
      RecordWriter<NullWritable, ObjectArrayWritable> deleteDeltaWriter =
          CarbonTableOutputFormat.getDeleteDeltaRecordWriter(path);
      ObjectArrayWritable writable = new ObjectArrayWritable();
      while (reader.hasNext()) {
        Object[] row = (Object[]) reader.readNextRow();
        writable.set(row);
        deleteDeltaWriter.write(NullWritable.get(), writable);
      }
      deleteDeltaWriter.close(null);
      reader.close();
    }
  
    /**
     * Calling this method will start the execution of delete process
     *
     * @throws IOException
     * @throws InterruptedException
     */
    private void closeDelete() throws IOException, InterruptedException {
      for (Map.Entry<String, Map<String, Set<String>>> path : this.filterColumnToValueMappingForDelete
          .entrySet()) {
        deleteExecution(path.getKey());
        createEmptyMetadataFile(path.getKey());
      }
    }
  
    private void createEmptyMetadataFile(String path) throws IOException {
      if (!StringUtils.isEmpty(path)) {
        path = path + CarbonCommonConstants.FILE_SEPARATOR +
            CarbonCommonConstants.CARBON_SDK_EMPTY_METADATA_PATH;
        CarbonFile emptySdkDirectory = FileFactory.getCarbonFile(path);
        if (!emptySdkDirectory.exists()) {
          emptySdkDirectory.mkdirs();
        }
      }
    }
  
    /**
     * @param path      is the table path on which update is performed
     * @param column    is the columnName on which records have to be updated
     * @param value     of column on which the records have to be updated
     * @param updColumn is the name of updatedColumn
     * @param updValue  is the value of updatedColumn
     * @return CarbonUID
     */
    public CarbonIUD update(String path, String column, String value, String updColumn,
        String updValue) {
      prepareUpdate(path, column, value, updColumn, updValue);
      return this;
    }
  
    /**
     * This method updates the rows at given path by applying the filterExpression
     *
     * @param path                        is the table path on which update is performed.
     * @param filterExpression            is the expression object to update the records
     * @param updatedColumnToValueMapping contains the mapping of updatedColumns to updatedValues
     * @throws IOException
     * @throws InterruptedException
     * @throws InvalidLoadOptionException
     */
    public void update(String path, Expression filterExpression,
        Map<String, String> updatedColumnToValueMapping)
        throws IOException, InterruptedException, InvalidLoadOptionException {
      List<String> indexFiles = getCarbonIndexFile(path);
      Schema schema = CarbonSchemaReader.readSchema(indexFiles.get(0)).asOriginOrder();
      Field[] fields = schema.getFields();
      String[] projectionColumns = new String[fields.length + 1];
      for (int i = 0; i < fields.length; i++) {
        projectionColumns[i] = (fields[i].getFieldName());
      }
      projectionColumns[projectionColumns.length - 1] =
          CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID;
      CarbonWriter writer =
          CarbonWriter.builder().outputPath(path)
              .withHadoopConf(configuration)
              .withCsvInput(schema)
              .writtenBy("CarbonIUD")
              .build();
      CarbonReader reader =
          CarbonReader.builder(path).projection(projectionColumns)
              .withHadoopConf(configuration)
              .filter(filterExpression).build();
      RecordWriter<NullWritable, ObjectArrayWritable> deleteDeltaWriter =
          CarbonTableOutputFormat.getDeleteDeltaRecordWriter(path);
      ObjectArrayWritable writable = new ObjectArrayWritable();
      while (reader.hasNext()) {
        Object[] row = (Object[]) reader.readNextRow();
        writable.set(Arrays.copyOfRange(row, row.length - 1, row.length));
        for (Map.Entry<String, String> column : updatedColumnToValueMapping.entrySet()) {
          row[getColumnIndex(fields, column.getKey())] = column.getValue();
        }
        writer.write(Arrays.copyOfRange(row, 0, row.length - 1));
        deleteDeltaWriter.write(NullWritable.get(), writable);
      }
      deleteDeltaWriter.close(null);
      writer.close();
      reader.close();
    }
  
    /**
     * Calling this method will start the execution of update process
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws InvalidLoadOptionException
     */
    private void closeUpdate() throws IOException, InterruptedException, InvalidLoadOptionException {
      for (Map.Entry<String, Map<String, Set<String>>> path : this.filterColumnToValueMappingForUpdate
          .entrySet()) {
        if (this.updateColumnToValueMapping.containsKey(path.getKey())) {
          updateExecution(path.getKey());
          createEmptyMetadataFile(path.getKey());
        }
      }
    }
  
    /**
     * Calling this method will execute delete and update operation in one statement.
     * This method will first perform delete operation and then update operation
     * (update operation is only performed if the rows to be
     * updated are not deleted while delete operation)
     * For eg:
     * CarbonIUD.getInstance().delete().delete().update()
     *
     * @throws IOException
     * @throws InterruptedException
     * @throws InvalidLoadOptionException
     */
    public void commit() throws IOException, InterruptedException, InvalidLoadOptionException {
      if (filterColumnToValueMappingForDelete.size() != 0) {
        closeDelete();
      }
      if (filterColumnToValueMappingForUpdate.size() != 0 && !ifRowsDeleted()) {
        closeUpdate();
      }
    }
  
    private void updateExecution(String path)
        throws IOException, InterruptedException, InvalidLoadOptionException {
      Expression filterExpression =
          getExpression(path, this.filterColumnToValueMappingForUpdate.get(path));
      update(path, filterExpression, this.updateColumnToValueMapping.get(path));
    }
  
    private void deleteExecution(String path) throws IOException, InterruptedException {
      Expression filterExpression =
          getExpression(path, this.filterColumnToValueMappingForDelete.get(path));
      delete(path, filterExpression);
    }
  
    /**
     * This method prepares the column to value mapping for update operation
     * for eg: UPDATE updColumn = updValue WHERE column = value
     */
    private void prepareUpdate(String path, String column, String value, String updColumn,
        String updValue) {
      prepareDelete(path, column, value, filterColumnToValueMappingForUpdate);
      updColumn = updColumn.toLowerCase().trim();
      if (this.updateColumnToValueMapping.containsKey(path)) {
        this.updateColumnToValueMapping.get(path).put(updColumn, updValue);
      } else {
        Map<String, String> columnToValue = new HashMap<>();
        columnToValue.put(updColumn, updValue);
        this.updateColumnToValueMapping.put(path, columnToValue);
      }
    }
  
    /**
     * This method prepares the column to value mapping for delete operation
     * for eg: DELETE WHERE column = value
     */
    private void prepareDelete(String path, String column, String value,
        Map<String, Map<String, Set<String>>> filterColumnToValueMapping) {
      column = column.toLowerCase().trim();
      if (filterColumnToValueMapping.containsKey(path)) {
        Map<String, Set<String>> columnToValueMapping = filterColumnToValueMapping.get(path);
        if (columnToValueMapping.containsKey(column)) {
          columnToValueMapping.get(column).add(value);
        } else {
          Set<String> columnValues = new HashSet<>();
          columnValues.add(value);
          columnToValueMapping.put(column, columnValues);
        }
      } else {
        Map<String, Set<String>> columnToValueMapping = new HashMap<>();
        Set<String> columnValues = new HashSet<>();
        columnValues.add(value);
        columnToValueMapping.put(column, columnValues);
        filterColumnToValueMapping.put(path, columnToValueMapping);
      }
    }
  
    /**
     * This method will convert the given columnToValue mapping into expression object
     * If columnToValue mapping have following entries:
     * name --> {karan, kunal, vikram}
     * age --> {24}
     * the expression will look like this for above entries:
     * ((name = karan || name = kunal || name = vikram) && (age = 24))
     */
    private Expression getExpression(String path, Map<String, Set<String>> columnToValueMapping)
        throws IOException {
      List<String> indexFiles = getCarbonIndexFile(path);
      Schema schema = CarbonSchemaReader.readSchema(indexFiles.get(0)).asOriginOrder();
      Field[] fields = schema.getFields();
      List<Expression> listOfExpressions = new ArrayList<>();
      for (Map.Entry<String, Set<String>> column : columnToValueMapping.entrySet()) {
        DataType dataType = getColumnDataType(fields, column.getKey());
        List<Expression> listOfOrExpressions = new ArrayList<>();
        for (String value : column.getValue()) {
          listOfOrExpressions.add(
              new EqualToExpression(new ColumnExpression(column.getKey(), dataType),
                  new LiteralExpression(value, dataType)));
        }
        Expression orFilterExpression = null;
        if (listOfOrExpressions.size() > 0) {
          orFilterExpression = listOfOrExpressions.get(0);
        }
        for (int i = 1; i < listOfOrExpressions.size(); i++) {
          orFilterExpression = new OrExpression(orFilterExpression, listOfOrExpressions.get(i));
        }
        listOfExpressions.add(orFilterExpression);
      }
      Expression filterExpression = null;
      if (listOfExpressions.size() > 0) {
        filterExpression = listOfExpressions.get(0);
      }
      for (int i = 1; i < listOfExpressions.size(); i++) {
        filterExpression = new AndExpression(filterExpression, listOfExpressions.get(i));
      }
      return filterExpression;
    }
  
    private int getColumnIndex(Field[] fields, String column) {
      int index = -1;
      for (Field field : fields) {
        if (field.getFieldName().equals(column)) {
          index = field.getSchemaOrdinal();
          break;
        }
      }
      if (index == -1) {
        throw new RuntimeException("ColumnName doesn't exists");
      }
      return index;
    }
  
    private List<String> getCarbonIndexFile(String path) {
      List<String> indexFiles = null;
      try (Stream<Path> walk = Files.walk(Paths.get(path))) {
        indexFiles = walk.map(x -> x.toString())
            .filter(f -> f.endsWith(CarbonCommonConstants.UPDATE_INDEX_FILE_EXT))
            .collect(Collectors.toList());
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      if (indexFiles == null || indexFiles.size() < 1) {
        throw new RuntimeException("Carbon index file does not exists.");
      }
      return indexFiles;
    }
  
    private DataType getColumnDataType(Field[] fields, String column) {
      DataType type = null;
      for (Field field : fields) {
        if (field.getFieldName().equals(column)) {
          type = field.getDataType();
          break;
        }
      }
      if (null == type) {
        throw new RuntimeException("ColumnName doesn't exists");
      }
      if (type.isComplexType()) {
        throw new RuntimeException("IUD operation not supported for Complex data types");
      }
      return type;
    }
  
    private boolean ifRowsDeleted() {
      for (Map.Entry<String, Map<String, Set<String>>> path : this.filterColumnToValueMappingForUpdate
          .entrySet()) {
        if (!this.filterColumnToValueMappingForDelete.containsKey(path.getKey())) {
          return false;
        } else {
          for (Map.Entry<String, Set<String>> column : this.filterColumnToValueMappingForDelete
              .get(path.getKey()).entrySet()) {
            if (!this.filterColumnToValueMappingForUpdate.get(path.getKey())
                .containsKey(column.getKey())) {
              return false;
            } else {
              for (String value : this.filterColumnToValueMappingForUpdate.get(path.getKey())
                  .get(column.getKey())) {
                if (!column.getValue().contains(value)) {
                  return false;
                }
              }
            }
          }
        }
      }
      return true;
    }
  }
  