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 com.alibaba.datax.plugin.writer.doriswriter;
  
  import com.alibaba.datax.common.element.Record;
  import com.alibaba.datax.common.exception.DataXException;
  import com.alibaba.datax.common.plugin.RecordReceiver;
  import com.alibaba.datax.common.spi.Writer;
  import com.alibaba.datax.common.util.Configuration;
  import com.alibaba.datax.plugin.rdbms.util.DBUtil;
  import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
  import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.sql.Connection;
  import java.util.ArrayList;
  import java.util.List;
  
  /**
   * doris data writer
   */
  public class DorisWriter extends Writer {
  
      public static class Job extends Writer.Job {
  
          private static final Logger LOG = LoggerFactory.getLogger(Job.class);
          private Configuration originalConfig = null;
          private Keys options;
  
          @Override
          public void init() {
              this.originalConfig = super.getPluginJobConf();
              options = new Keys (super.getPluginJobConf());
              options.doPretreatment();
          }
  
          @Override
          public void preCheck(){
              this.init();
              DorisUtil.preCheckPrePareSQL(options);
              DorisUtil.preCheckPostSQL(options);
          }
  
          @Override
          public void prepare() {
              String username = options.getUsername();
              String password = options.getPassword();
              String jdbcUrl = options.getJdbcUrl();
              List<String> renderedPreSqls = DorisUtil.renderPreOrPostSqls(options.getPreSqlList(), options.getTable());
              if (null != renderedPreSqls && !renderedPreSqls.isEmpty()) {
                  Connection conn = DBUtil.getConnection(DataBaseType.MySql, jdbcUrl, username, password);
                  LOG.info("Begin to execute preSqls:[{}]. context info:{}.", String.join(";", renderedPreSqls), jdbcUrl);
                  DorisUtil.executeSqls(conn, renderedPreSqls);
                  DBUtil.closeDBResources(null, null, conn);
              }
          }
  
          @Override
          public List<Configuration> split(int mandatoryNumber) {
              List<Configuration> configurations = new ArrayList<>(mandatoryNumber);
              for (int i = 0; i < mandatoryNumber; i++) {
                  configurations.add(originalConfig);
              }
              return configurations;
          }
  
          @Override
          public void post() {
              String username = options.getUsername();
              String password = options.getPassword();
              String jdbcUrl = options.getJdbcUrl();
              List<String> renderedPostSqls = DorisUtil.renderPreOrPostSqls(options.getPostSqlList(), options.getTable());
              if (null != renderedPostSqls && !renderedPostSqls.isEmpty()) {
                  Connection conn = DBUtil.getConnection(DataBaseType.MySql, jdbcUrl, username, password);
                  LOG.info("Start to execute preSqls:[{}]. context info:{}.", String.join(";", renderedPostSqls), jdbcUrl);
                  DorisUtil.executeSqls(conn, renderedPostSqls);
                  DBUtil.closeDBResources(null, null, conn);
              }
          }
  
          @Override
          public void destroy() {
          }
  
      }
  
      public static class Task extends Writer.Task {
          private DorisWriterManager writerManager;
          private Keys options;
          private DorisCodec rowCodec;
  
          @Override
          public void init() {
              options = new Keys (super.getPluginJobConf());
              if (options.isWildcardColumn()) {
                  Connection conn = DBUtil.getConnection(DataBaseType.MySql, options.getJdbcUrl(), options.getUsername(), options.getPassword());
                  List<String> columns = DorisUtil.getDorisTableColumns(conn, options.getDatabase(), options.getTable());
                  options.setInfoCchemaColumns(columns);
              }
              writerManager = new DorisWriterManager(options);
              rowCodec = DorisCodecFactory.createCodec(options);
          }
  
          @Override
          public void prepare() {
          }
  
          public void startWrite(RecordReceiver recordReceiver) {
              try {
                  Record record;
                  while ((record = recordReceiver.getFromReader()) != null) {
                      if (record.getColumnNumber() != options.getColumns().size()) {
                          throw DataXException
                                  .asDataXException(
                                          DBUtilErrorCode.CONF_ERROR,
                                          String.format(
                                                  "There is an error in the column configuration information. " +
                                                  "This is because you have configured a task where the number of fields to be read from the source:%s " +
                                                  "is not equal to the number of fields to be written to the destination table:%s. " +
                                                  "Please check your configuration and make changes.",
                                                  record.getColumnNumber(),
                                                  options.getColumns().size()));
                      }
                      writerManager.writeRecord(rowCodec.codec(record));
                  }
              } catch (Exception e) {
                  throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
              }
          }
  
          @Override
          public void post() {
              try {
                  writerManager.close();
              } catch (Exception e) {
                  throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
              }
          }
  
          @Override
          public void destroy() {}
  
          @Override
          public boolean supportFailOver(){
              return false;
          }
      }
  }
  