private boolean move(Connection connection,
                         SegmentSchema schema,
                         SegmentMode mode,
                         AggSchema aggSchema,
                         String tableLocation,
                         String tablePartitionColumn,
                         String rtTableName,
                         String segTmpTableName) throws Exception {
        String selectSql;
        if (this.selectSql == null) {
            String allColumnStr = Joiner.on(",").join(Lists.transform(
                    schema.getColumns(),
                    cs -> "`" + cs.getName() + "`"));
            selectSql = String.format("SELECT %s FROM %s WHERE %s", allColumnStr, TABLE_NAME, WHERE_PREDICATE);
        } else {
            selectSql = this.selectSql;
        }

        if (!hasChild(fileSystem, new Path(tableLocation, "rt"))
                && !hasChild(fileSystem, new Path(tableLocation, "rt2his"))) {
            logger.debug("Table {} has not realtime segments to move", hiveTable);
            return true;
        }

        String rt2hisPath = tableLocation + "/rt2his";
        createTable(connection, rtTableName, true, schema, mode, aggSchema, rt2hisPath, null);
        createTable(connection, segTmpTableName, false, schema, mode, aggSchema, null, tablePartitionColumn);
        String segTmpTableLocation = getHiveTableLocation(getHiveTableCreateSql(connection, segTmpTableName));

        // Move segments in /rt into /rt2his folder.
        if (hasChild(fileSystem, new Path(tableLocation, "rt"))) {
            String importRtTableSql = String.format("LOAD DATA INPATH '%s' INTO TABLE %s", tableLocation + "/rt", rtTableName);
            try (Statement statement = connection.createStatement()) {
                statement.execute(importRtTableSql);
            }
        }

        // Notify indexr to update segments.
        SegmentHelper.notifyUpdate(fileSystem, tableLocation);

        String selectPartitionSql = String.format("SELECT DISTINCT %s FROM %s", segmentPartitionColumn, rtTableName);
        logger.debug("selectPartitionSql: {}", selectPartitionSql);
        List<String> partitions = new ArrayList<>();

        try (Statement statement = connection.createStatement()) {
            ResultSet rs = statement.executeQuery(selectPartitionSql);
            while (rs.next()) {
                partitions.add(rs.getString(1));
            }
            if (partitions.size() == 0) {
                return true;
            }
            rs.close();
        }
        logger.debug("partitions: {}", partitions);

        // Group rows in rt folder into partitions.
        for (String partition : partitions) {
            String selectFromSql = selectSql.replace(TABLE_NAME, rtTableName);
            selectFromSql = selectFromSql.replace(
                    WHERE_PREDICATE,
                    String.format("%s = '%s'", segmentPartitionColumn, partition));

            String insertPartitionSql = String.format(
                    "INSERT INTO TABLE %s PARTITION (`%s` = '%s') %s",
                    segTmpTableName,
                    tablePartitionColumn,
                    partition,
                    selectFromSql);
            logger.debug("insertPartitionSql: {}", insertPartitionSql);

            try (Statement myStatement = connection.createStatement()) {
                myStatement.execute(insertPartitionSql);
            }
        }

        // Move segment files from segTmpTable back into hiveTable.
        List<Path> movedSegmentPaths = new ArrayList<>();
        try {
            String fromPrefix = segTmpTableLocation + "/";
            for (String partition : partitions) {
                String fromPartitionPath = String.format("%s/%s=%s", segTmpTableLocation, tablePartitionColumn, partition);
                String toPartitionPath = String.format("%s/%s=%s", tableLocation, tablePartitionColumn, partition);
                if (!fileSystem.exists(new Path(toPartitionPath))) {
                    fileSystem.mkdirs(new Path(toPartitionPath));
                }
                FileStatus[] files = fileSystem.listStatus(new Path(fromPartitionPath));
                for (FileStatus f : files) {
                    Path from = f.getPath();
                    if (!SegmentHelper.checkSegmentByPath(from)) {
                        continue;
                    }
                    String toFileName = String.format("rt2his.%s.%s.seg", time, UUID.randomUUID().toString());
                    Path to = new Path(toPartitionPath, toFileName);
                    fileSystem.rename(from, to);

                    movedSegmentPaths.add(to);
                }

                // Add partitions manually.
                String addPartition = String.format("ALTER TABLE %s ADD IF NOT EXISTS PARTITION (%s='%s')",
                        hiveTable, tablePartitionColumn, partition);
                try (Statement myStatement = connection.createStatement()) {
                    myStatement.execute(addPartition);
                }
            }
        } catch (Exception e) {
            logger.error("move segment file from segTmpTable failed", e);
            // Try the best to roll back.
            List<Path> remainPaths = new LinkedList<>(movedSegmentPaths);
            Boolean ok = Try.on(() -> {
                        Iterator<Path> it = remainPaths.iterator();
                        while (it.hasNext()) {
                            Path p = it.next();
                            if (fileSystem.exists(p)) {
                                if (fileSystem.delete(p)) {
                                    it.remove();
                                }
                            }
                        }
                        return remainPaths.isEmpty();
                    },
                    5, logger);
            if (ok == null || !ok) {
                logger.error(String.format("Some generated segments failed to move into destination path, data in table [%s] may duplicated. " +
                        "You can try remove those successful segments (%s) manually to fix it. " +
                        "You may need to re-run this task after fix the issue.", hiveTable, remainPaths));
            }

            return false;
        }

        Boolean ok = Try.on(() -> fileSystem.delete(new Path(rt2hisPath), true),
                5, logger,
                String.format("Remove rt2his folder failed. " +
                                "Some old segments failed to deleted, data in table [%s] duplicated. " +
                                "You should delete them manually in [%s] to fix it.",
                        hiveTable, rt2hisPath));
        return ok != null && ok;
    }
A tool which move realtime data in rt folder into history partition.
Only support one partition column.
It only works on indexr tables which has a hive table with location pointed to the same path and with a table partition column.
indexr-server/src/main/java/io/indexr/server/rt2his/Rt2HisOnHive.java