private ColsAndTypes deriveFileSinkColTypes(
      RowResolver inputRR, List<String> partitionColumnNames, List<String> sortColumnNames, List<String> distributeColumnNames,
      List<FieldSchema> columns, List<FieldSchema> partitionColumns, List<FieldSchema> sortColumns, List<FieldSchema> distributeColumns,
      List<ColumnInfo> fileSinkColInfos, List<ColumnInfo> sortColInfos, List<ColumnInfo> distributeColInfos) throws SemanticException {
    ColsAndTypes result = new ColsAndTypes("", "");
    List<String> allColumns = new ArrayList<>();
    List<ColumnInfo> colInfos = inputRR.getColumnInfos();
    List<ColumnInfo> nonPartColInfos = new ArrayList<>();
    SortedMap<Integer, Pair<FieldSchema, ColumnInfo>> partColInfos = new TreeMap<>();
    SortedMap<Integer, Pair<FieldSchema, ColumnInfo>> sColInfos = new TreeMap<>();
    SortedMap<Integer, Pair<FieldSchema, ColumnInfo>> dColInfos = new TreeMap<>();
    boolean first = true;
    int numNonPartitionedCols = colInfos.size() - partitionColumnNames.size();
    if (numNonPartitionedCols <= 0) {
      throw new SemanticException("Too many partition columns declared");
    }
    for (ColumnInfo colInfo : colInfos) {
      String[] nm = inputRR.reverseLookup(colInfo.getInternalName());

      if (nm[1] != null) { // non-null column alias
        colInfo.setAlias(nm[1]);
      }

      boolean isPartitionCol = false;
      String colName = colInfo.getInternalName();  //default column name
      if (columns != null) {
        FieldSchema col = new FieldSchema();
        if (!("".equals(nm[0])) && nm[1] != null) {
          colName = unescapeIdentifier(colInfo.getAlias()).toLowerCase(); // remove ``
        }
        colName = fixCtasColumnName(colName);
        col.setName(colName);
        allColumns.add(colName);
        String typeName = colInfo.getType().getTypeName();
        // CTAS should NOT create a VOID type
        if (typeName.equals(serdeConstants.VOID_TYPE_NAME)) {
          throw new SemanticException(ErrorMsg.CTAS_CREATES_VOID_TYPE.getMsg(colName));
        }
        col.setType(typeName);
        int idx = partitionColumnNames.indexOf(colName);
        if (idx >= 0) {
          partColInfos.put(idx, Pair.of(col, colInfo));
          isPartitionCol = true;
        } else {
          if (sortColumnNames != null) {
            idx = sortColumnNames.indexOf(colName);
            if (idx >= 0) {
              sColInfos.put(idx, Pair.of(col, colInfo));
            }
          }
          if (distributeColumnNames != null) {
            idx = distributeColumnNames.indexOf(colName);
            if (idx >= 0) {
              dColInfos.put(idx, Pair.of(col, colInfo));
            }
          }
          columns.add(col);
          nonPartColInfos.add(colInfo);
        }
      }

      if (!isPartitionCol) {
        if (!first) {
          result.cols = result.cols.concat(",");
          result.colTypes = result.colTypes.concat(":");
        }

        first = false;
        result.cols = result.cols.concat(colName);

        // Replace VOID type with string when the output is a temp table or
        // local files.
        // A VOID type can be generated under the query:
        //
        // select NULL from tt;
        // or
        // insert overwrite local directory "abc" select NULL from tt;
        //
        // where there is no column type to which the NULL value should be
        // converted.
        //
        String tName = colInfo.getType().getTypeName();
        if (tName.equals(serdeConstants.VOID_TYPE_NAME)) {
          result.colTypes = result.colTypes.concat(serdeConstants.STRING_TYPE_NAME);
        } else {
          result.colTypes = result.colTypes.concat(tName);
        }
      }

    }

    if (partColInfos.size() != partitionColumnNames.size()) {
      throw new SemanticException("Table declaration contains partition columns that are not present " +
        "in query result schema. " +
        "Query columns: " + allColumns + ". " +
        "Partition columns: " + partitionColumnNames);
    }

    if (sortColumnNames != null && sColInfos.size() != sortColumnNames.size()) {
      throw new SemanticException("Table declaration contains cluster/sort columns that are not present " +
          "in query result schema. " +
          "Query columns: " + allColumns + ". " +
          "Organization columns: " + sortColumnNames);
    }

    if (distributeColumnNames != null && dColInfos.size() != distributeColumnNames.size()) {
      throw new SemanticException("Table declaration contains cluster/distribute columns that are not present " +
          "in query result schema. " +
          "Query columns: " + allColumns + ". " +
          "Organization columns: " + distributeColumnNames);
    }

    // FileSinkColInfos comprise nonPartCols followed by partCols
    fileSinkColInfos.addAll(nonPartColInfos);
    partitionColumns.addAll(partColInfos.values().stream().map(Pair::getLeft).collect(Collectors.toList()));
    fileSinkColInfos.addAll(partColInfos.values().stream().map(Pair::getRight).collect(Collectors.toList()));
    // data org columns
    if (sortColumnNames != null) {
      sortColumns.addAll(sColInfos.values().stream().map(Pair::getLeft).collect(Collectors.toList()));
      sortColInfos.addAll(sColInfos.values().stream().map(Pair::getRight).collect(Collectors.toList()));
    }
    if (distributeColumnNames != null) {
      distributeColumns.addAll(dColInfos.values().stream().map(Pair::getLeft).collect(Collectors.toList()));
      distributeColInfos.addAll(dColInfos.values().stream().map(Pair::getRight).collect(Collectors.toList()));
    }

    return result;
  }
We will directly insert to the final destination in the following cases:
1. Non native table
2. Micro-managed (insert only table)
3. Full ACID table and operation type is INSERT
ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java