private void createTable(SinkInfo sinkInfo) {
        LOG.info("begin to create SQLServer table for sinkId={}", sinkInfo.getId());
        List<StreamSinkFieldEntity> fieldList = fieldEntityMapper.selectBySinkId(sinkInfo.getId());
        if (CollectionUtils.isEmpty(fieldList)) {
            LOG.warn("no SQLServer fields found, skip to create table for sinkId={}", sinkInfo.getId());
        }
        // set columns
        List<SQLServerColumnInfo> columnList = Lists.newArrayList();
        fieldList.forEach(field -> {
            SQLServerColumnInfo columnInfo = new SQLServerColumnInfo(
                    field.getFieldName(),
                    field.getFieldType(),
                    field.getFieldComment());
            columnList.add(columnInfo);
        });

        final SQLServerSinkDTO sink = SQLServerSinkDTO.getFromJson(sinkInfo.getExtParams());
        final SQLServerTableInfo tableInfo = SQLServerSinkDTO.getTableInfo(sink, columnList);

        try (Connection conn = SQLServerJdbcUtils.getConnection(sink.getJdbcUrl(),
                sink.getUsername(), sink.getPassword())) {

            // 1. create schema if not exists
            SQLServerJdbcUtils.createSchema(conn, tableInfo.getSchemaName());
            // 2. if table not exists, create it
            SQLServerJdbcUtils.createTable(conn, tableInfo);
            // 3. if table exists, add columns - skip the exists columns
            SQLServerJdbcUtils.addColumns(conn, tableInfo.getSchemaName(), tableInfo.getTableName(), columnList);
            // 4. update the sink status to success
            final String info = "success to create SQLServer resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
        } catch (Throwable e) {
            String errMsg = "create SQLServer 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 SQLServer table for data sink [" + sinkInfo.getId() + "]");
    }
Create SQLServer table by SinkInfo.

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