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.presto;
  
  import java.io.IOException;
  import java.io.UncheckedIOException;
  import java.util.Arrays;
  import java.util.List;
  import java.util.Properties;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.hadoop.api.CarbonTableOutputFormat;
  import org.apache.carbondata.hive.CarbonHiveSerDe;
  import org.apache.carbondata.hive.MapredCarbonOutputFormat;
  import org.apache.carbondata.presto.impl.CarbonTableConfig;
  
  import com.google.common.collect.ImmutableList;
  import io.prestosql.plugin.hive.FileWriter;
  import io.prestosql.plugin.hive.HiveType;
  import io.prestosql.plugin.hive.util.HiveWriteUtils;
  import io.prestosql.spi.Page;
  import io.prestosql.spi.PrestoException;
  import io.prestosql.spi.block.Block;
  import io.prestosql.spi.type.Type;
  import io.prestosql.spi.type.TypeManager;
  import org.apache.commons.lang3.StringUtils;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.hive.conf.HiveConf;
  import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
  import org.apache.hadoop.hive.ql.io.IOConstants;
  import org.apache.hadoop.hive.serde2.SerDeException;
  import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.SettableStructObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.StructField;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.Reporter;
  import org.apache.log4j.Logger;
  
  import static com.google.common.collect.ImmutableList.toImmutableList;
  import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_WRITER_CLOSE_ERROR;
  import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_WRITER_DATA_ERROR;
  import static java.util.Objects.requireNonNull;
  import static java.util.stream.Collectors.toList;
  import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.COMPRESSRESULT;
  import static org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory.getStandardStructObjectInspector;
  
  /**
   * This class implements HiveFileWriter and it creates the carbonFileWriter to write the page data
   * sent from presto.
   */
  public class CarbonDataFileWriter implements FileWriter {
  
    private static final Logger LOG =
        LogServiceFactory.getLogService(CarbonDataFileWriter.class.getName());
  
    private final JobConf configuration;
    private final Path outPutPath;
    private final FileSinkOperator.RecordWriter recordWriter;
    private final CarbonHiveSerDe serDe;
    private final int fieldCount;
    private final Object row;
    private final SettableStructObjectInspector tableInspector;
    private final List<StructField> structFields;
    private final HiveWriteUtils.FieldSetter[] setters;
  
    private boolean isCommitDone;
  
    public CarbonDataFileWriter(Path outPutPath, List<String> inputColumnNames, Properties properties,
        JobConf configuration, TypeManager typeManager) throws SerDeException {
      requireNonNull(outPutPath, "path is null");
      // take the outputPath same as location in compliance with the carbon store folder structure.
      this.outPutPath = new Path(properties.getProperty("location"));
      this.configuration = requireNonNull(configuration, "conf is null");
      List<String> columnNames = Arrays
          .asList(properties.getProperty(IOConstants.COLUMNS, "").split(CarbonCommonConstants.COMMA));
      List<Type> fileColumnTypes =
          HiveType.toHiveTypes(properties.getProperty(IOConstants.COLUMNS_TYPES, "")).stream()
              .map(hiveType -> hiveType.getType(typeManager)).collect(toList());
      this.fieldCount = columnNames.size();
      this.serDe = new CarbonHiveSerDe();
      serDe.initialize(configuration, properties);
      List<ObjectInspector> objectInspectors = fileColumnTypes.stream()
          .map(HiveWriteUtils::getRowColumnInspector)
          .collect(toList());
      this.tableInspector = getStandardStructObjectInspector(columnNames, objectInspectors);
  
      this.structFields =
          ImmutableList.copyOf(inputColumnNames.stream().map(tableInspector::getStructFieldRef)
              .collect(toImmutableList()));
  
      this.row = tableInspector.create();
  
      this.setters = new HiveWriteUtils.FieldSetter[structFields.size()];
      for (int i = 0; i < setters.length; i++) {
        setters[i] = HiveWriteUtils.createFieldSetter(tableInspector, row, structFields.get(i),
            fileColumnTypes.get(structFields.get(i).getFieldID()));
      }
      String encodedLoadModel = this.configuration.get(CarbonTableConfig.CARBON_PRESTO_LOAD_MODEL);
      if (StringUtils.isNotEmpty(encodedLoadModel)) {
        this.configuration.set(CarbonTableOutputFormat.LOAD_MODEL, encodedLoadModel);
      }
      try {
        boolean compress = HiveConf.getBoolVar(this.configuration, COMPRESSRESULT);
        Object writer =
            Class.forName(MapredCarbonOutputFormat.class.getName()).getConstructor().newInstance();
        this.recordWriter = ((MapredCarbonOutputFormat<?>) writer)
            .getHiveRecordWriter(this.configuration, this.outPutPath, Text.class, compress,
                properties, Reporter.NULL);
      } catch (Exception e) {
        LOG.error("error while initializing writer", e);
        throw new RuntimeException("writer class not found");
      }
    }
  
    @Override
    public long getWrittenBytes() {
      if (isCommitDone) {
        try {
          return outPutPath.getFileSystem(configuration).getFileStatus(outPutPath).getLen();
        } catch (IOException e) {
          throw new UncheckedIOException(e);
        }
      }
      return 0;
    }
  
    @Override
    public long getSystemMemoryUsage() {
      // TODO: need to support this, Jira CARBONDATA-4038 is created
      return 0;
    }
  
    @Override
    public void appendRows(Page dataPage) {
      for (int position = 0; position < dataPage.getPositionCount(); position++) {
        appendRow(dataPage, position);
      }
    }
  
    private void appendRow(Page dataPage, int position) {
      for (int field = 0; field < fieldCount; field++) {
        Block block = dataPage.getBlock(field);
        if (block.isNull(position)) {
          tableInspector.setStructFieldData(row, structFields.get(field), null);
        } else {
          setters[field].setField(block, position);
        }
      }
      try {
        recordWriter.write(serDe.serialize(row, tableInspector));
      } catch (SerDeException | IOException e) {
        throw new PrestoException(HIVE_WRITER_DATA_ERROR, e);
      }
    }
  
    @Override
    public void commit() {
      try {
        recordWriter.close(false);
      } catch (Exception ex) {
        LOG.error("Error while closing the record writer", ex);
        throw new RuntimeException(ex);
      }
      isCommitDone = true;
    }
  
    @Override
    public void rollback() {
      try {
        recordWriter.close(true);
      } catch (Exception e) {
        LOG.error("Error while closing the record writer during rollback", e);
        throw new PrestoException(HIVE_WRITER_CLOSE_ERROR, "Error rolling back write to Hive", e);
      }
    }
  
    @Override
    public long getValidationCpuNanos() {
      // TODO: need to support this, Jira CARBONDATA-4038 is created
      return 0;
    }
  }
  