@SuppressWarnings("nls")
  protected Operator genFileSinkPlan(String dest, QB qb, Operator input)
      throws SemanticException {

    RowResolver inputRR = opParseCtx.get(input).getRowResolver();
    QBMetaData qbm = qb.getMetaData();
    Integer destType = qbm.getDestTypeForAlias(dest);

    Table destinationTable = null; // destination table if any
    boolean destTableIsTransactional;     // true for full ACID table and MM table
    boolean destTableIsFullAcid; // should the destination table be written to using ACID
    boolean isDirectInsert = false; // should we add files directly to the final path
    AcidUtils.Operation acidOperation = null;
    boolean destTableIsTemporary = false;
    boolean destTableIsMaterialization = false;
    Partition destinationPartition = null;// destination partition if any
    Path queryTmpdir = null; // the intermediate destination directory
    String moveTaskId = null;
    Path destinationPath = null; // the final destination directory
    TableDesc tableDescriptor = null;
    StructObjectInspector specificRowObjectInspector = null;
    int currentTableId = 0;
    boolean isLocal = false;
    SortBucketRSCtx rsCtx = new SortBucketRSCtx();
    DynamicPartitionCtx dpCtx = null;
    LoadTableDesc ltd = null;
    ListBucketingCtx lbCtx = null;
    Map<String, String> partSpec = null;
    boolean isMmTable = false, isMmCreate = false, isNonNativeTable = false, isAlreadyContainsPartCols = false;
    Long writeId = null;
    HiveTxnManager txnMgr = getTxnMgr();

    switch (destType.intValue()) {
    case QBMetaData.DEST_TABLE: {

      destinationTable = qbm.getDestTableForAlias(dest);
      destTableIsTransactional = AcidUtils.isTransactionalTable(destinationTable);
      destTableIsFullAcid = AcidUtils.isFullAcidTable(destinationTable);
      destTableIsTemporary = destinationTable.isTemporary();

      // Is the user trying to insert into a external tables
      checkExternalTable(destinationTable);

      partSpec = qbm.getPartSpecForAlias(dest);
      destinationPath = destinationTable.getPath();

      checkImmutableTable(qb, destinationTable, destinationPath, false);

      // Check for dynamic partitions.
      dpCtx = checkDynPart(qb, qbm, destinationTable, partSpec, dest);

      isNonNativeTable = destinationTable.isNonNative();
      isMmTable = AcidUtils.isInsertOnlyTable(destinationTable.getParameters());
      AcidUtils.Operation acidOp = AcidUtils.Operation.NOT_ACID;
      // this table_desc does not contain the partitioning columns
      tableDescriptor = Utilities.getTableDesc(destinationTable);

      if (!isNonNativeTable) {
        if (destTableIsTransactional) {
          acidOp = getAcidType(tableDescriptor.getOutputFileFormatClass(), dest, isMmTable);
        }
      }
      isDirectInsert = isDirectInsert(destTableIsFullAcid, acidOp);
      acidOperation = acidOp;
      queryTmpdir = getTmpDir(isNonNativeTable, isMmTable, isDirectInsert, destinationPath, dpCtx);
      moveTaskId = getMoveTaskId();
      if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
        Utilities.FILE_OP_LOGGER.trace("create filesink w/DEST_TABLE specifying " + queryTmpdir
            + " from " + destinationPath);
      }
      if (dpCtx != null) {
        // set the root of the temporary path where dynamic partition columns will populate
        dpCtx.setRootPath(queryTmpdir);
      }

      if (!qb.getIsQuery()) {
        isAlreadyContainsPartCols = Optional.ofNullable(destinationTable)
            .map(Table::hasNonNativePartitionSupport)
            .orElse(Boolean.FALSE);
        if (!updating(dest) && !deleting(dest) && isAlreadyContainsPartCols
            && destinationTable != null && partSpec != null) {
          // Do not specify partition spec again since its already added in the earlier.
          input = genConversionSelectOperatorByAddPartition(dest, qb, input, destinationTable.getDeserializer(),
                  destinationTable, partSpec);
        } else {
          input = genConversionSelectOperator(dest, qb, input, destinationTable.getDeserializer(),
                  dpCtx, destinationTable.getPartitionKeys(), destinationTable);
        }
      }

      // Add NOT NULL constraint check
      input = genConstraintsPlan(dest, qb, input);

      if (destinationTable.isMaterializedView() &&
          mvRebuildMode == MaterializationRebuildMode.INSERT_OVERWRITE_REBUILD) {
        // Data organization (DISTRIBUTED, SORTED, CLUSTERED) for materialized view
        // TODO: We only do this for a full rebuild
        String sortColsStr = destinationTable.getProperty(Constants.MATERIALIZED_VIEW_SORT_COLUMNS);
        String distributeColsStr = destinationTable.getProperty(Constants.MATERIALIZED_VIEW_DISTRIBUTE_COLUMNS);
        if (sortColsStr != null || distributeColsStr != null) {
          input = genMaterializedViewDataOrgPlan(destinationTable, sortColsStr, distributeColsStr, inputRR, input);
        }
      } else {
        // Add sorting/bucketing if needed
        input = genBucketingSortingDest(dest, input, qb, tableDescriptor, destinationTable, rsCtx);
      }

      idToTableNameMap.put(String.valueOf(destTableId), destinationTable.getTableName());
      currentTableId = destTableId;
      destTableId++;
      // Create the work for moving the table
      // NOTE: specify Dynamic partitions in dest_tab for WriteEntity
      if (!isNonNativeTable || destinationTable.getStorageHandler().commitInMoveTask()) {
        if (destTableIsTransactional) {
          acidOp = getAcidType(tableDescriptor.getOutputFileFormatClass(), dest, isMmTable);
          checkAcidConstraints();
        } else {
          lbCtx = constructListBucketingCtx(destinationTable.getSkewedColNames(),
              destinationTable.getSkewedColValues(), destinationTable.getSkewedColValueLocationMaps(),
              destinationTable.isStoredAsSubDirectories());
        }
        writeId = allocateTableWriteId(destinationTable.getFullTableName(), isMmTable || acidOp != Operation.NOT_ACID);
        
        boolean isReplace = !qb.getParseInfo().isInsertIntoTable(
            destinationTable.getDbName(), destinationTable.getTableName(), destinationTable.getSnapshotRef());
        ltd = new LoadTableDesc(queryTmpdir, tableDescriptor, dpCtx, acidOp, isReplace, writeId);
        if (writeId != null) {
          ltd.setStmtId(txnMgr.getCurrentStmtId());
        }
        ltd.setMoveTaskId(moveTaskId);
        // For Acid table, Insert Overwrite shouldn't replace the table content. We keep the old
        // deltas and base and leave them up to the cleaner to clean up
        boolean isInsertInto = qb.getParseInfo().isInsertIntoTable(
            destinationTable.getDbName(), destinationTable.getTableName(), destinationTable.getSnapshotRef());
        LoadFileType loadType;
        if (isDirectInsert) {
          loadType = LoadFileType.IGNORE;
        } else if (!isInsertInto && !destTableIsTransactional) {
          loadType = LoadFileType.REPLACE_ALL;
        } else {
          loadType = LoadFileType.KEEP_EXISTING;
        }
        ltd.setLoadFileType(loadType);
        ltd.setInsertOverwrite(!isInsertInto);
        ltd.setIsDirectInsert(isDirectInsert);
        ltd.setLbCtx(lbCtx);
        loadTableWork.add(ltd);
      } else {
        // This is a non-native table.
        // We need to set stats as inaccurate.
        setStatsForNonNativeTable(destinationTable.getDbName(), destinationTable.getTableName());
        // true if it is insert overwrite.
        boolean overwrite = !qb.getParseInfo().isInsertIntoTable(destinationTable.getDbName(), destinationTable.getTableName(),
            destinationTable.getSnapshotRef());
        createPreInsertDesc(destinationTable, overwrite);

        ltd = new LoadTableDesc(queryTmpdir, tableDescriptor, partSpec == null ? ImmutableMap.of() : partSpec);
        ltd.setInsertOverwrite(overwrite);
        ltd.setLoadFileType(overwrite ? LoadFileType.REPLACE_ALL : LoadFileType.KEEP_EXISTING);
      }

      if (destinationTable.isMaterializedView()) {
        materializedViewUpdateDesc = new MaterializedViewUpdateDesc(
            destinationTable.getFullyQualifiedName(), false, false, true);
      }

      WriteEntity output = generateTableWriteEntity(dest, destinationTable, partSpec, ltd, dpCtx);
      ctx.getLoadTableOutputMap().put(ltd, output);
      break;
    }
    case QBMetaData.DEST_PARTITION: {

      destinationPartition = qbm.getDestPartitionForAlias(dest);
      destinationTable = destinationPartition.getTable();
      destTableIsTransactional = AcidUtils.isTransactionalTable(destinationTable);
      destTableIsFullAcid = AcidUtils.isFullAcidTable(destinationTable);

      checkExternalTable(destinationTable);

      Path partPath = destinationPartition.getDataLocation();

      checkImmutableTable(qb, destinationTable, partPath, true);

      // Previous behavior (HIVE-1707) used to replace the partition's dfs with the table's dfs.
      // The changes in HIVE-19891 appears to no longer support that behavior.
      destinationPath = partPath;

      if (MetaStoreUtils.isArchived(destinationPartition.getTPartition())) {
        try {
          String conflictingArchive = ArchiveUtils.conflictingArchiveNameOrNull(
                  db, destinationTable, destinationPartition.getSpec());
          String message = String.format("Insert conflict with existing archive: %s",
                  conflictingArchive);
          throw new SemanticException(message);
        } catch (SemanticException err) {
          throw err;
        } catch (HiveException err) {
          throw new SemanticException(err);
        }
      }

      isNonNativeTable = destinationTable.isNonNative();
      isMmTable = AcidUtils.isInsertOnlyTable(destinationTable.getParameters());
      AcidUtils.Operation acidOp = AcidUtils.Operation.NOT_ACID;
      // this table_desc does not contain the partitioning columns
      tableDescriptor = Utilities.getTableDesc(destinationTable);

      if (!isNonNativeTable) {
        if (destTableIsTransactional) {
          acidOp = getAcidType(tableDescriptor.getOutputFileFormatClass(), dest, isMmTable);
        }
      }
      isDirectInsert = isDirectInsert(destTableIsFullAcid, acidOp);
      acidOperation = acidOp;
      queryTmpdir = getTmpDir(isNonNativeTable, isMmTable, isDirectInsert, destinationPath, null);
      moveTaskId = getMoveTaskId();
      if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
        Utilities.FILE_OP_LOGGER.trace("create filesink w/DEST_PARTITION specifying "
            + queryTmpdir + " from " + destinationPath);
      }

      if (destinationTable.hasNonNativePartitionSupport()) {
        partSpec = qbm.getPartSpecForAlias(dest);
      }

      if (!qb.getIsQuery()) {
        isAlreadyContainsPartCols = Optional.ofNullable(destinationTable)
                .map(Table::hasNonNativePartitionSupport)
                .orElse(Boolean.FALSE);
        if (!updating(dest) && !deleting(dest) && isAlreadyContainsPartCols
                && destinationTable != null && partSpec != null) {
          input = genConversionSelectOperatorByAddPartition(dest, qb, input, destinationTable.getDeserializer(),
                  destinationTable, partSpec);
        } else {
          input = genConversionSelectOperator(dest, qb, input, destinationTable.getDeserializer(),
                  dpCtx, null, destinationTable);
        }
      }

      // Add NOT NULL constraint check
      input = genConstraintsPlan(dest, qb, input);

      if (destinationTable.isMaterializedView() &&
          mvRebuildMode == MaterializationRebuildMode.INSERT_OVERWRITE_REBUILD) {
        // Data organization (DISTRIBUTED, SORTED, CLUSTERED) for materialized view
        // TODO: We only do this for a full rebuild
        String sortColsStr = destinationTable.getProperty(Constants.MATERIALIZED_VIEW_SORT_COLUMNS);
        String distributeColsStr = destinationTable.getProperty(Constants.MATERIALIZED_VIEW_DISTRIBUTE_COLUMNS);
        if (sortColsStr != null || distributeColsStr != null) {
          input = genMaterializedViewDataOrgPlan(destinationTable, sortColsStr, distributeColsStr, inputRR, input);
        }
      } else {
        // Add sorting/bucketing if needed
        input = genBucketingSortingDest(dest, input, qb, tableDescriptor, destinationTable, rsCtx);
      }

      idToTableNameMap.put(String.valueOf(destTableId), destinationTable.getTableName());
      currentTableId = destTableId;
      destTableId++;

      if (destTableIsTransactional) {
        acidOp = getAcidType(tableDescriptor.getOutputFileFormatClass(), dest, isMmTable);
        checkAcidConstraints();
      } else {
        // Transactional tables can't be list bucketed or have skewed cols
        lbCtx = constructListBucketingCtx(destinationPartition.getSkewedColNames(),
            destinationPartition.getSkewedColValues(), destinationPartition.getSkewedColValueLocationMaps(),
            destinationPartition.isStoredAsSubDirectories());
      }
      writeId = allocateTableWriteId(destinationTable.getFullTableName(), isMmTable || acidOp != Operation.NOT_ACID);
      
      ltd = new LoadTableDesc(queryTmpdir, tableDescriptor, destinationPartition.getSpec(), acidOp, writeId);
      if (writeId != null) {
        ltd.setStmtId(txnMgr.getCurrentStmtId());
      }
      // For the current context for generating File Sink Operator, it is either INSERT INTO or INSERT OVERWRITE.
      // So the next line works.
      boolean isInsertInto = !qb.getParseInfo().isDestToOpTypeInsertOverwrite(dest);
      // For Acid table, Insert Overwrite shouldn't replace the table content. We keep the old
      // deltas and base and leave them up to the cleaner to clean up
      LoadFileType loadType;
      if (isDirectInsert) {
        loadType = LoadFileType.IGNORE;
      } else if (!isInsertInto && !destTableIsTransactional) {
        loadType = LoadFileType.REPLACE_ALL;
      } else {
        loadType = LoadFileType.KEEP_EXISTING;
      }
      ltd.setLoadFileType(loadType);
      ltd.setInsertOverwrite(!isInsertInto);
      ltd.setIsDirectInsert(isDirectInsert);
      ltd.setLbCtx(lbCtx);
      ltd.setMoveTaskId(moveTaskId);

      loadTableWork.add(ltd);

      if (destinationTable.hasNonNativePartitionSupport()) {
        // HMS does not know about this partition
        // but the underlying storage format knows about it.
        DummyPartition dummyPartition;
        try {
          String partName = Warehouse.makePartName(partSpec, false);
          dummyPartition = new DummyPartition(destinationTable, partName, partSpec);
        } catch (MetaException e) {
          throw new SemanticException("Unable to construct name for dummy partition due to: ", e);
        }
        if (!outputs.add(new WriteEntity(dummyPartition, determineWriteType(ltd, dest)))) {
          throw new SemanticException(ErrorMsg.OUTPUT_SPECIFIED_MULTIPLE_TIMES
                  .getMsg(destinationTable.getTableName() + "@" + dummyPartition.getName()));
        }
      } else {
        if (!outputs.add(new WriteEntity(destinationPartition, determineWriteType(ltd, dest)))) {
          throw new SemanticException(ErrorMsg.OUTPUT_SPECIFIED_MULTIPLE_TIMES
                  .getMsg(destinationTable.getTableName() + "@" + destinationPartition.getName()));
        }
      }
     break;
    }
    case QBMetaData.DEST_LOCAL_FILE:
      isLocal = true;
      // fall through
    case QBMetaData.DEST_DFS_FILE: {
      destinationPath = getDestinationFilePath(qb, qbm.getDestFileForAlias(dest), isMmTable);

      // CTAS case: the file output format and serde are defined by the create
      // table command rather than taking the default value
      List<FieldSchema> fieldSchemas = null;
      List<FieldSchema> partitionColumns = null;
      List<String> partitionColumnNames = null;
      List<FieldSchema> sortColumns = null;
      List<String> sortColumnNames = null;
      List<FieldSchema> distributeColumns = null;
      List<String> distributeColumnNames = null;
      List<ColumnInfo> fileSinkColInfos = null;
      List<ColumnInfo> sortColInfos = null;
      List<ColumnInfo> distributeColInfos = null;
      Map<String, String> tblProps = null;
      CreateTableDesc tblDesc = qb.getTableDesc();
      CreateMaterializedViewDesc viewDesc = qb.getViewDesc();
      boolean createTableUseSuffix = false;
      
      DDLDescWithTableProperties ddlDesc = (tblDesc != null) ? tblDesc : viewDesc;
      if (ddlDesc != null) {
        fieldSchemas = new ArrayList<>();
        partitionColumns = new ArrayList<>();
        partitionColumnNames = ddlDesc.getPartColNames();
        fileSinkColInfos = new ArrayList<>();
        destTableIsTemporary = ddlDesc.isTemporary();
        destTableIsMaterialization = ddlDesc.isMaterialization();
        tblProps = ddlDesc.getTblProps();

        // Add suffix only when required confs are present
        // and user has not specified a location to the table.
        createTableUseSuffix = (HiveConf.getBoolVar(conf, ConfVars.HIVE_ACID_CREATE_TABLE_USE_SUFFIX)
            || HiveConf.getBoolVar(conf, ConfVars.HIVE_ACID_LOCKLESS_READS_ENABLED))
          && ddlDesc.getLocation() == null;
      }
      if (viewDesc != null) {
        sortColumns = new ArrayList<>();
        sortColumnNames = viewDesc.getSortColNames();
        distributeColumns = new ArrayList<>();
        distributeColumnNames = viewDesc.getDistributeColNames();
        sortColInfos = new ArrayList<>();
        distributeColInfos = new ArrayList<>();
      }

      destTableIsTransactional = tblProps != null && AcidUtils.isTablePropertyTransactional(tblProps);
      if (destTableIsTransactional) {
        isNonNativeTable = MetaStoreUtils.isNonNativeTable(tblProps);
        isMmTable = isMmCreate = AcidUtils.isInsertOnlyTable(tblProps);
        
        writeId = allocateTableWriteId(ddlDesc.getFullTableName(), 0L);
        
        if (!isNonNativeTable && !destTableIsTemporary && (qb.isCTAS() || qb.isMaterializedView())) {
          destTableIsFullAcid = AcidUtils.isFullAcidTable(tblProps);
          acidOperation = getAcidType(dest);
          isDirectInsert = isDirectInsert(destTableIsFullAcid, acidOperation);
          
          // Set the location in context for possible rollback.
          ctx.setLocation(getCtasOrCMVLocation(tblDesc, viewDesc, createTableUseSuffix));
          
          if (isDirectInsert || isMmTable) {
            destinationPath = ctx.getLocation();
            if (createTableUseSuffix) {
              ddlDesc.getTblProps().put(SOFT_DELETE_TABLE, Boolean.TRUE.toString());
            }
            // Setting the location so that metadata transformers
            // does not change the location later while creating the table.
            ddlDesc.setLocation(destinationPath.toString());
          }
        }
        if (isMmTable || isDirectInsert) {
          ddlDesc.setInitialWriteId(writeId);
        }
      }

      // Check for dynamic partitions.
      final String cols, colTypes;
      final boolean isPartitioned;
      if (dpCtx != null) {
        throw new SemanticException("Dynamic partition context has already been created, this should not happen");
      }
      if (!CollectionUtils.isEmpty(partitionColumnNames)) {
        ColsAndTypes ct = deriveFileSinkColTypes(
            inputRR, partitionColumnNames, sortColumnNames, distributeColumnNames, fieldSchemas, partitionColumns,
            sortColumns, distributeColumns, fileSinkColInfos, sortColInfos, distributeColInfos);
        cols = ct.cols;
        colTypes = ct.colTypes;
        dpCtx = new DynamicPartitionCtx(partitionColumnNames,
            conf.getVar(HiveConf.ConfVars.DEFAULT_PARTITION_NAME),
            conf.getIntVar(HiveConf.ConfVars.DYNAMIC_PARTITION_MAX_PARTS_PER_NODE));
        qbm.setDPCtx(dest, dpCtx);
        isPartitioned = true;
      } else {
        ColsAndTypes ct = deriveFileSinkColTypes(
            inputRR, sortColumnNames, distributeColumnNames, fieldSchemas, sortColumns, distributeColumns,
            sortColInfos, distributeColInfos);
        cols = ct.cols;
        colTypes = ct.colTypes;
        isPartitioned = false;
      }

      if (isLocal) {
        assert !isMmTable;
        // for local directory - we always write to map-red intermediate
        // store and then copy to local fs
        queryTmpdir = ctx.getMRTmpPath();
        if (dpCtx != null && dpCtx.getSPPath() != null) {
          queryTmpdir = new Path(queryTmpdir, dpCtx.getSPPath());
        }
      } else {
        // otherwise write to the file system implied by the directory
        // no copy is required. we may want to revisit this policy in future
        try {
          Path qPath = FileUtils.makeQualified(destinationPath, conf);
          queryTmpdir = getTmpDir(false, isMmTable, isDirectInsert, qPath, dpCtx);
        } catch (Exception e) {
          throw new SemanticException("Error creating "
              + destinationPath, e);
        }
      }
      // set the root of the temporary path where dynamic partition columns will populate
      if (dpCtx != null) {
        dpCtx.setRootPath(queryTmpdir);
      }

      if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
        Utilities.FILE_OP_LOGGER.trace("Setting query directory " + queryTmpdir
            + " from " + destinationPath + " (" + isMmTable + ")");
      }

      // update the create table descriptor with the resulting schema.
      if (tblDesc != null) {
        tblDesc.setCols(new ArrayList<>(fieldSchemas));
        tblDesc.setPartCols(new ArrayList<>(partitionColumns));
      } else if (viewDesc != null) {
        viewDesc.setCols(new ArrayList<>(fieldSchemas));
        viewDesc.setPartCols(new ArrayList<>(partitionColumns));
        if (viewDesc.isOrganized()) {
          viewDesc.setSortCols(new ArrayList<>(sortColumns));
          viewDesc.setDistributeCols(new ArrayList<>(distributeColumns));
        }
      }

      boolean isDestTempFile = true;
      if (!ctx.isMRTmpFileURI(destinationPath.toUri().toString())
          && !ctx.isResultCacheDir(destinationPath)) {
        // not a temp dir and not a result cache dir
        idToTableNameMap.put(String.valueOf(destTableId), destinationPath.toUri().toString());
        currentTableId = destTableId;
        destTableId++;
        isDestTempFile = false;
      }

      try {
        if (tblDesc == null) {
          if (viewDesc != null) {
            if (viewDesc.getStorageHandler() != null) {
              viewDesc.setLocation(getCtasOrCMVLocation(tblDesc, viewDesc, createTableUseSuffix).toString());
            }
            tableDescriptor = PlanUtils.getTableDesc(viewDesc, cols, colTypes);
          } else if (qb.getIsQuery()) {
            Class<? extends Deserializer> serdeClass = LazySimpleSerDe.class;
            String fileFormat = conf.getResultFileFormat().toString();
            if (SessionState.get().getIsUsingThriftJDBCBinarySerDe()) {
              serdeClass = ThriftJDBCBinarySerDe.class;
              fileFormat = ResultFileFormat.SEQUENCEFILE.toString();
              // Set the fetch formatter to be a no-op for the ListSinkOperator, since we'll
              // write out formatted thrift objects to SequenceFile
              conf.set(SerDeUtils.LIST_SINK_OUTPUT_FORMATTER, NoOpFetchFormatter.class.getName());
            } else if (fileFormat.equals(PlanUtils.LLAP_OUTPUT_FORMAT_KEY)) {
              serdeClass = LazyBinarySerDe2.class;
            }
            tableDescriptor = PlanUtils.getDefaultQueryOutputTableDesc(cols, colTypes, fileFormat,
                serdeClass);
          } else {
            tableDescriptor = PlanUtils.getDefaultTableDesc(qb.getDirectoryDesc(), cols, colTypes);
          }
        } else {
          if (tblDesc.isCTAS() && tblDesc.getStorageHandler() != null) {
            tblDesc.toTable(conf).getStorageHandler().setTableLocationForCTAS(
                tblDesc, getCtasOrCMVLocation(tblDesc, viewDesc, false).toString());
          }
          tableDescriptor = PlanUtils.getTableDesc(tblDesc, cols, colTypes);
        }
      } catch (HiveException e) {
        throw new SemanticException(e);
      }

      // We need a specific rowObjectInspector in this case
      try {
        specificRowObjectInspector =
            (StructObjectInspector) tableDescriptor.getDeserializer(conf).getObjectInspector();
      } catch (Exception e) {
        throw new SemanticException(e.getMessage(), e);
      }

      boolean isDfsDir = (destType == QBMetaData.DEST_DFS_FILE);

      try {
        if (tblDesc != null) {
          Table t = tblDesc.toTable(conf);
          destinationTable = tblDesc.isMaterialization() ? t : db.getTranslateTableDryrun(t.getTTable());
        } else {
          destinationTable = viewDesc != null ? viewDesc.toTable(conf) : null;
        }
      } catch (HiveException e) {
        throw new SemanticException(e);
      }

      destTableIsFullAcid = AcidUtils.isFullAcidTable(destinationTable);

      // Data organization (DISTRIBUTED, SORTED, CLUSTERED) for materialized view
      if (viewDesc != null && viewDesc.isOrganized()) {
        input = genMaterializedViewDataOrgPlan(sortColInfos, distributeColInfos, inputRR, input);
      }

      moveTaskId = getMoveTaskId();

      if (isPartitioned) {
        // Create a SELECT that may reorder the columns if needed
        RowResolver rowResolver = new RowResolver();
        List<ExprNodeDesc> columnExprs = new ArrayList<>();
        List<String> colNames = new ArrayList<>();
        Map<String, ExprNodeDesc> colExprMap = new HashMap<>();
        for (int i = 0; i < fileSinkColInfos.size(); i++) {
          ColumnInfo ci = fileSinkColInfos.get(i);
          ExprNodeDesc columnExpr = new ExprNodeColumnDesc(ci);
          String name = getColumnInternalName(i);
          rowResolver.put("", name, new ColumnInfo(name, columnExpr.getTypeInfo(), "", false));
          columnExprs.add(columnExpr);
          colNames.add(name);
          colExprMap.put(name, columnExpr);
        }
        input = putOpInsertMap(OperatorFactory.getAndMakeChild(
            new SelectDesc(columnExprs, colNames), new RowSchema(rowResolver
                .getColumnInfos()), input), rowResolver);
        input.setColumnExprMap(colExprMap);

        // If this is a partitioned CTAS or MV statement, we are going to create a LoadTableDesc
        // object. Although the table does not exist in metastore, we will swap the CreateTableTask
        // and MoveTask resulting from this LoadTable so in this specific case, first we create
        // the metastore table, then we move and commit the partitions. At least for the time being,
        // this order needs to be enforced because metastore expects a table to exist before we can
        // add any partitions to it.
        isNonNativeTable = tableDescriptor.isNonNative();
        if (!isNonNativeTable || destinationTable.getStorageHandler().commitInMoveTask()) {
          AcidUtils.Operation acidOp = AcidUtils.Operation.NOT_ACID;
          if (destTableIsTransactional) {
            acidOp = getAcidType(tableDescriptor.getOutputFileFormatClass(), dest, isMmTable);
            checkAcidConstraints();
          }
          // isReplace = false in case concurrent operation is executed
          ltd = new LoadTableDesc(queryTmpdir, tableDescriptor, dpCtx, acidOp, false, writeId);
          if (writeId != null) {
            ltd.setStmtId(txnMgr.getCurrentStmtId());
          }
          ltd.setLoadFileType(LoadFileType.KEEP_EXISTING);
          ltd.setInsertOverwrite(false);
          ltd.setIsDirectInsert(isDirectInsert);
          loadTableWork.add(ltd);
        } else {
          // This is a non-native table.
          // We need to set stats as inaccurate.
          setStatsForNonNativeTable(tableDescriptor.getDbName(), tableDescriptor.getTableName());
          ltd = new LoadTableDesc(queryTmpdir, tableDescriptor, dpCtx.getPartSpec());
          ltd.setInsertOverwrite(false);
          ltd.setLoadFileType(LoadFileType.KEEP_EXISTING);
        }
        ltd.setMoveTaskId(moveTaskId);
        ltd.setMdTable(destinationTable);
        WriteEntity output = generateTableWriteEntity(dest, destinationTable, dpCtx.getPartSpec(), ltd, dpCtx);
        ctx.getLoadTableOutputMap().put(ltd, output);
      } else {
        // Create LFD even for MM CTAS - it's a no-op move, but it still seems to be used for stats.
        LoadFileDesc loadFileDesc = new LoadFileDesc(tblDesc, viewDesc, queryTmpdir, destinationPath, isDfsDir, cols,
            colTypes,
            destTableIsFullAcid ?//there is a change here - prev version had 'transactional', one before 'acid'
                Operation.INSERT : Operation.NOT_ACID,
            isMmCreate);
        loadFileDesc.setMoveTaskId(moveTaskId);
        loadFileWork.add(loadFileDesc);
        try {
          FileSystem fs = isDfsDir ?  destinationPath.getFileSystem(conf) : FileSystem.getLocal(conf);
          Path qualifiedPath = conf.getBoolVar(ConfVars.HIVE_RANGER_USE_FULLY_QUALIFIED_URL) ?
              fs.makeQualified(destinationPath) : destinationPath;
          if (!outputs.add(new WriteEntity(qualifiedPath, !isDfsDir, isDestTempFile))) {
            throw new SemanticException(ErrorMsg.OUTPUT_SPECIFIED_MULTIPLE_TIMES
                    .getMsg(destinationPath.toUri().toString()));
          }
        } catch (IOException ex) {
          throw new SemanticException("Error while getting the full qualified path for the given directory: " + ex.getMessage());
        }
      }
      break;
    }
    default:
      throw new SemanticException("Unknown destination type: " + destType);
    }

    if (!(destType == QBMetaData.DEST_DFS_FILE && qb.getIsQuery())
            && destinationTable != null && destinationTable.getStorageHandler() != null) {
      try {
        if (!updating(dest) && !deleting(dest) && isAlreadyContainsPartCols && MapUtils.isNotEmpty(partSpec)) {
          input = genConversionSelectOperatorByAddPartition(dest, qb, input, destinationTable.getDeserializer(),
                  destinationTable, null);
        } else {
          input = genConversionSelectOperator(dest, qb, input, destinationTable.getDeserializer(),
                  dpCtx, null, destinationTable);
        }
      } catch (Exception e) {
        throw new SemanticException(e);
      }
    }

    inputRR = opParseCtx.get(input).getRowResolver();

    List<ColumnInfo> vecCol = new ArrayList<ColumnInfo>();
    
    if (updating(dest) || deleting(dest) || merging(dest)) {
      if (AcidUtils.isNonNativeAcidTable(destinationTable)) {
        destinationTable.getStorageHandler().acidVirtualColumns().stream()
            .map(col -> new ColumnInfo(col.getName(), col.getTypeInfo(), "", true))
            .forEach(vecCol::add);
      } else {
        vecCol.add(new ColumnInfo(VirtualColumn.ROWID.getName(), VirtualColumn.ROWID.getTypeInfo(),
            "", true));
      }
    } else {
      try {
        // If we already have a specific inspector (view or directory as a target) use that
        // Otherwise use the table deserializer to get the inspector
        StructObjectInspector rowObjectInspector = specificRowObjectInspector != null ? specificRowObjectInspector :
            (StructObjectInspector) destinationTable.getDeserializer().getObjectInspector();
        List<? extends StructField> fields = rowObjectInspector
            .getAllStructFieldRefs();
        for (StructField field : fields) {
          vecCol.add(new ColumnInfo(field.getFieldName(), TypeInfoUtils
              .getTypeInfoFromObjectInspector(field
                  .getFieldObjectInspector()), "", false));
        }
      } catch (Exception e) {
        throw new SemanticException(e.getMessage(), e);
      }
    }

    RowSchema fsRS = new RowSchema(vecCol);

    // The output files of a FileSink can be merged if they are either not being written to a table
    // or are being written to a table which is not bucketed
    // and table the table is not sorted
    boolean canBeMerged = (destinationTable == null || !((destinationTable.getNumBuckets() > 0) ||
        (destinationTable.getSortCols() != null && destinationTable.getSortCols().size() > 0)));

    // If this table is working with ACID semantics or
    // if its a delete, update, merge operation that supports merge task, turn off merging
    canBeMerged &= !destTableIsFullAcid;
    if (destinationTable != null && destinationTable.getStorageHandler() != null) {
      canBeMerged &= destinationTable.getStorageHandler().supportsMergeFiles();
      // TODO: Support for merge task for update and merge queries
      //  when storage handler supports it.
      if (Context.Operation.DELETE.equals(ctx.getOperation()) && !deleting(dest)) {
        canBeMerged = true;
      }
      if (Context.Operation.UPDATE.equals(ctx.getOperation())
              || Context.Operation.MERGE.equals(ctx.getOperation())) {
        canBeMerged = false;
      }
    }

    // Generate the partition columns from the parent input
    if (destType == QBMetaData.DEST_TABLE || destType == QBMetaData.DEST_PARTITION) {
      genPartnCols(dest, input, qb, tableDescriptor, destinationTable, rsCtx);
    }

    FileSinkDesc fileSinkDesc = createFileSinkDesc(dest, tableDescriptor, destinationPartition,
        destinationPath, currentTableId, destTableIsFullAcid, destTableIsTemporary,//this was 1/4 acid
        destTableIsMaterialization, queryTmpdir, rsCtx, dpCtx, lbCtx, fsRS,
        canBeMerged, destinationTable, isMmCreate, destType, qb, isDirectInsert, acidOperation, moveTaskId);
    if (isMmCreate || (qb.isCTAS() || qb.isMaterializedView()) && isDirectInsert) {
      // Add FSD so that the LoadTask compilation could fix up its path to avoid the move.
      if (tableDesc != null) {
        tableDesc.setWriter(fileSinkDesc);
      } else {
        createVwDesc.setWriter(fileSinkDesc);
      }
    }

    if (fileSinkDesc.getInsertOverwrite()) {
      if (ltd != null) {
        ltd.setInsertOverwrite(true);
      }
    }
    if (null != tableDescriptor && useBatchingSerializer(tableDescriptor.getSerdeClassName())) {
      fileSinkDesc.setIsUsingBatchingSerDe(true);
    } else {
      fileSinkDesc.setIsUsingBatchingSerDe(false);
    }

    Operator output = putOpInsertMap(OperatorFactory.getAndMakeChild(
        fileSinkDesc, fsRS, input), inputRR);

    // In the MoveTask the lineage information is not set in case of delete and update as the
    // columns are not matching. So we only need the lineage information for insert.
    // If directInsert=false, adding the lineage info here for the other operations is
    // ok, as the paths are different. But if directInsert=true, the path for all
    // operation is the same (the table location) and this can lead to invalid lineage information
    // in case of a merge statement.
    if (!isDirectInsert || acidOperation == AcidUtils.Operation.INSERT) {
      handleLineage(destinationTable, ltd, output);
    }
    setWriteIdForSurrogateKeys(ltd, input);

    LOG.debug("Created FileSink Plan for clause: {}dest_path: {} row schema: {}", dest, destinationPath, inputRR);

    FileSinkOperator fso = (FileSinkOperator) output;
    fso.getConf().setTable(destinationTable);
    // the following code is used to collect column stats when
    // hive.stats.autogather=true
    // and it is an insert overwrite or insert into table
    if (conf.getBoolVar(ConfVars.HIVE_STATS_AUTOGATHER)
        && conf.getBoolVar(ConfVars.HIVE_STATS_COL_AUTOGATHER)
        && enableColumnStatsCollecting()
        && destinationTable != null
        && (!destinationTable.isNonNative() || destinationTable.getStorageHandler().commitInMoveTask())
        && !destTableIsTemporary && !destTableIsMaterialization
        && ColumnStatsAutoGatherContext.canRunAutogatherStats(fso)) {
      if (destType == QBMetaData.DEST_TABLE) {
        genAutoColumnStatsGatheringPipeline(destinationTable, partSpec, input,
            qb.getParseInfo().isInsertIntoTable(destinationTable.getDbName(), destinationTable.getTableName(),
                destinationTable.getSnapshotRef()), false);
      } else if (destType == QBMetaData.DEST_PARTITION) {
        genAutoColumnStatsGatheringPipeline(destinationTable, destinationPartition.getSpec(), input,
            qb.getParseInfo().isInsertIntoTable(destinationTable.getDbName(), destinationTable.getTableName(),
                destinationTable.getSnapshotRef()), false);
      } else if (destType == QBMetaData.DEST_LOCAL_FILE || destType == QBMetaData.DEST_DFS_FILE) {
        // CTAS or CMV statement
        genAutoColumnStatsGatheringPipeline(destinationTable, null, input,
            false, true);
      }
    }
    return output;
  }
@param totalFiles
         the totalFiles to set
ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java