public boolean handle() {
        Connection connection = null;
        String tableLocation = null;
        String rtTableName = null;
        String segTmpTableName = null;
        try {
            Class.forName("org.apache.hive.jdbc.HiveDriver");
            connection = DriverManager.getConnection(hiveConnect, user, pwd);

            // Looks like hive could produce errors when running concurrent sql throught HiveServer2 in local mode.
            // We disable local mode here to avoid this issue.
            try (Statement stmt = connection.createStatement()) {
                stmt.execute("set hive.exec.mode.local.auto = false");
            }

            String createTable = getHiveTableCreateSql(connection, String.format("%s", hiveTable));
            logger.debug("createTable: {}", createTable);
            tableLocation = getHiveTableLocation(createTable);
            String tablePartitionColumn = getHiveTablePartitionColumn(createTable);
            SegmentSchema schema = getSchema(createTable);
            Map<String, String> properties = getTableProperties(createTable);
            SegmentMode mode = SegmentMode.fromName(properties.get(HiveHelper.KEY_SEGMENT_MODE));

            String sortColumnsStr = properties.getOrDefault(HiveHelper.KEY_SORT_COLUMNS, "");
            String aggGroupingStr = properties.getOrDefault(HiveHelper.KEY_AGG_GROUPING, "false");
            String aggDimsStr = properties.getOrDefault(HiveHelper.KEY_AGG_DIMS, "");
            String aggMetricsStr = properties.getOrDefault(HiveHelper.KEY_AGG_METRICS, "");

            boolean grouping = Boolean.parseBoolean(aggGroupingStr.trim());
            List<String> sortColumns = Trick.split(sortColumnsStr, ",", String::trim);
            List<String> dims = Trick.split(aggDimsStr, ",", String::trim);
            if (dims.isEmpty()) {
                dims = sortColumns;
            }
            List<Metric> metrics = Trick.split(aggMetricsStr, ",", s -> {
                String[] ss = s.trim().split(":", 2);
                return new Metric(ss[0].trim(), ss[1].trim());
            });
            AggSchema aggSchema = new AggSchema(
                    grouping,
                    dims,
                    metrics);
            String error = RealtimeHelper.validateSetting(schema.columns, dims, metrics, grouping);
            Preconditions.checkState(error == null, error);

            logger.debug("tableLocation: {}", tableLocation);
            logger.debug("tablePartitionColumn: {}", tablePartitionColumn);
            logger.debug("schema: {}", schema);
            logger.debug("mode: {}", mode);
            logger.debug("aggSchema: {}", aggSchema);
            logger.debug("properties: {}", JsonUtil.toJson(properties));

            rtTableName = String.format(
                    "indexr_rt2his_rt_%s_%s_%s",
                    hiveTable,
                    time,
                    RandomStringUtils.randomAlphabetic(16));
            segTmpTableName = String.format(
                    "indexr_rt2his_seg_%s_%s_%s",
                    hiveTable,
                    time,
                    RandomStringUtils.randomAlphabetic(16));

            return move(connection, schema, mode, aggSchema, tableLocation, tablePartitionColumn, rtTableName, segTmpTableName);
        } catch (Exception e) {
            logger.error("", e);
            return false;
        } finally {
            if (tableLocation != null) {
                final String finalTableLocation = tableLocation;
                Try.on(() -> SegmentHelper.notifyUpdate(fileSystem, finalTableLocation),
                        1, logger,
                        "Notify segment update failed");
            }

            if (connection != null) {
                try (Statement statement = connection.createStatement()) {
                    String dropSegTmpSql = String.format("DROP TABLE %s", segTmpTableName);
                    logger.debug("dropSegTmpSql: {}", dropSegTmpSql);
                    Try.on(() -> statement.execute(dropSegTmpSql),
                            1, logger,
                            "Drop segment tmp table failed");

                    String dropRTTableSql = String.format("DROP TABLE %s", rtTableName);
                    logger.debug("dropRTTableSql: {}", dropRTTableSql);
                    Try.on(() -> {
                                statement.execute(dropRTTableSql);
                            }, 5, logger,
                            "Drop rt2his tmp table failed");
                } catch (Exception e) {
                    logger.warn("Drop tmp tables failed! You can safely remove them manually.", e);
                }
            }

            IOUtil.closeQuietly(connection);
        }
    }
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