private void createTable(SinkInfo sinkInfo) {
        LOG.info("begin to create starRocks table for sinkId={}", sinkInfo.getId());
        List<StreamSinkFieldEntity> fieldList = fieldEntityMapper.selectBySinkId(sinkInfo.getId());
        if (CollectionUtils.isEmpty(fieldList)) {
            LOG.warn("no starRocks fields found, skip to create table for sinkId={}", sinkInfo.getId());
        }
        // get columns
        List<StarRocksColumnInfo> columnList = getStarRocksColumnInfoFromSink(fieldList);

        StarRocksSinkDTO sinkDTO = getStarRocksInfo(sinkInfo);
        StarRocksTableInfo tableInfo = StarRocksSinkDTO.getTableInfo(sinkDTO, columnList);
        String url = sinkDTO.getJdbcUrl();
        String username = sinkDTO.getUsername();
        String password = sinkDTO.getPassword();
        String dbName = sinkDTO.getDatabaseName();
        String tableName = sinkDTO.getTableName();
        try {
            // 1. create database if not exists
            StarRocksJdbcUtils.createDb(url, username, password, dbName);
            String dbUrl = url + "/" + dbName;
            // 2. table not exists, create it
            StarRocksJdbcUtils.createTable(dbUrl, username, password, tableInfo);
            // 3. table exists, add columns - skip the exists columns
            StarRocksJdbcUtils.addColumns(dbUrl, username, password, dbName, tableName, columnList);
            // 4. update the sink status to success
            String info = "success to create StarRocks resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
        } catch (Throwable e) {
            String errMsg = "create StarRocks table failed: " + e.getMessage();
            LOG.error(errMsg, e);
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_FAILED.getCode(), errMsg);
            throw new WorkflowException(errMsg);
        }
        LOG.info("success create StarRocks table for data sink [" + sinkInfo.getId() + "]");
    }
Create starRocks table by SinkInfo.

@param sinkInfo {@link SinkInfo}
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/starrocks/StarRocksResourceOperator.java