private void process(HdfsIdFile idFile) throws FileNotFoundException, IOException {
        DistributedFileSystem fs = idFile.getFs();

        // rename files in "intmp" directory to "in" directory.
        FileStatus[] intmpFiles = fs.listStatus(idFile.getIntmpPath());
        long currentTime = System.currentTimeMillis();
        long fileArchiveDelayTime = currentTime
                - context.getFileArchiveDelayMinute() * MINUTE_MS;
        for (FileStatus fileStatus : intmpFiles) {
            if (fileStatus.getModificationTime() > fileArchiveDelayTime) {
                continue;
            }
            Path intmpFilePath = fileStatus.getPath();
            String strIntmpFile = intmpFilePath.getName();
            Path inFilePath = new Path(idFile.getInPath(), strIntmpFile);
            fs.rename(intmpFilePath, inFilePath);
        }

        // clear "outtmp" fiels.
        FileStatus[] inFiles = fs.listStatus(idFile.getInPath());
        for (FileStatus fileStatus : inFiles) {
            Path inFile = fileStatus.getPath();
            if (inFile.getName().lastIndexOf(HdfsIdFile.OUTTMP_FILE_POSTFIX) >= 0) {
                fs.delete(inFile, true);
            }
        }

        // merge and copy files in "in" directory to "outtmp" file.
        long outputFileSize = 0;
        List<Path> concatInFiles = new ArrayList<>();
        for (FileStatus fileStatus : inFiles) {
            if (fileStatus.getLen() <= 0) {
                continue;
            }
            if (outputFileSize < context.getMaxOutputFileSizeGb() * HiveSinkContext.GB_BYTES) {
                concatInFiles.add(fileStatus.getPath());
                continue;
            }
            this.concatInFiles2OuttmpFile(idFile, concatInFiles, fs);
            outputFileSize = 0;
            concatInFiles.clear();
        }
        if (concatInFiles.size() > 0) {
            this.concatInFiles2OuttmpFile(idFile, concatInFiles, fs);
            outputFileSize = 0;
            concatInFiles.clear();
        }

        // rename outtmp file to "out" directory.
        inFiles = fs.listStatus(idFile.getInPath());
        for (FileStatus fileStatus : inFiles) {
            Path inFile = fileStatus.getPath();
            if (inFile.getName().lastIndexOf(HdfsIdFile.OUTTMP_FILE_POSTFIX) >= 0) {
                String strFullFile = inFile.getName();
                String strOuttmpFile = strFullFile.substring(0,
                        strFullFile.length() - HdfsIdFile.OUTTMP_FILE_POSTFIX.length());
                Path outFilePath = new Path(idFile.getOutPath(), strOuttmpFile);
                fs.rename(inFile, outFilePath);
            }
        }

        // delete file in "in" directory.
        for (FileStatus fileStatus : inFiles) {
            fs.delete(fileStatus.getPath(), true);
        }

        // execute the sql of adding partition.
        try (Connection conn = context.getHiveConnection()) {
            Statement stat = conn.createStatement();
            String partitionSqlPattern = "ALTER TABLE %s.%s ADD IF NOT EXISTS PARTITION (dt='%s') LOCATION '%s'";
            String partitionSql = String.format(partitionSqlPattern,
                    context.getHiveDatabase(),
                    idConfig.getHiveTableName(),
                    this.strPartitionValue,
                    idFile.getOutPath().toString());
            LOG.info("create partition sql:{}", partitionSql);
            stat.executeUpdate(partitionSql);
            stat.close();
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }
process

@param  idFile
@throws IOException
@throws FileNotFoundException
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/hive/PartitionCreateRunnable.java