@Override
    public void finishSink(String dbName, String tableName, List<TSinkCommitInfo> commitInfos, String branch) {
        if (commitInfos.isEmpty()) {
            LOG.warn("No commit info on {}.{} after hive sink", dbName, tableName);
            return;
        }
        HiveTable table = (HiveTable) getTable(dbName, tableName);
        String stagingDir = commitInfos.get(0).getStaging_dir();
        boolean isOverwrite = commitInfos.get(0).isIs_overwrite();

        List<PartitionUpdate> partitionUpdates = commitInfos.stream()
                .map(TSinkCommitInfo::getHive_file_info)
                .map(fileInfo -> PartitionUpdate.get(fileInfo, stagingDir, table.getTableLocation()))
                .collect(Collectors.collectingAndThen(Collectors.toList(), PartitionUpdate::merge));

        List<String> partitionColNames = table.getPartitionColumnNames();
        for (PartitionUpdate partitionUpdate : partitionUpdates) {
            PartitionUpdate.UpdateMode mode;
            if (table.isUnPartitioned()) {
                mode = isOverwrite ? UpdateMode.OVERWRITE : UpdateMode.APPEND;
                partitionUpdate.setUpdateMode(mode);
                break;
            } else {
                List<String> partitionValues = toPartitionValues(partitionUpdate.getName());
                Preconditions.checkState(partitionColNames.size() == partitionValues.size(),
                        "Partition columns names size doesn't equal partition values size. %s vs %s",
                        partitionColNames.size(), partitionValues.size());
                if (hmsOps.partitionExists(table, partitionValues)) {
                    mode = isOverwrite ? UpdateMode.OVERWRITE : UpdateMode.APPEND;
                } else {
                    mode = PartitionUpdate.UpdateMode.NEW;
                }
                partitionUpdate.setUpdateMode(mode);
            }
        }

        HiveCommitter committer = new HiveCommitter(
                hmsOps, fileOps, updateExecutor, refreshOthersFeExecutor, table, new Path(stagingDir));
        try (Timer ignored = Tracers.watchScope(EXTERNAL, "HIVE.SINK.commit")) {
            committer.commit(partitionUpdates);
        }
    }
@Test(expected = StarRocksConnectorException.class)
    public void testFinishSink() {
        String stagingDir = "hdfs://127.0.0.1:10000/tmp/starrocks/queryid";
        THiveFileInfo fileInfo = new THiveFileInfo();
        fileInfo.setFile_name("myfile.parquet");
        fileInfo.setPartition_path("hdfs://127.0.0.1:10000/tmp/starrocks/queryid/col1=2");
        fileInfo.setRecord_count(10);
        fileInfo.setFile_size_in_bytes(100);
        TSinkCommitInfo tSinkCommitInfo = new TSinkCommitInfo();
        tSinkCommitInfo.setStaging_dir(stagingDir);
        tSinkCommitInfo.setIs_overwrite(false);
        tSinkCommitInfo.setHive_file_info(fileInfo);
        hiveMetadata.finishSink("hive_db", "hive_table", Lists.newArrayList(), null);
        hiveMetadata.finishSink("hive_db", "hive_table", Lists.newArrayList(tSinkCommitInfo), null);
    }