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

        OceanBaseSinkDTO sinkDTO = this.getOceanBaseInfo(sinkInfo);
        OceanBaseTableInfo tableInfo = OceanBaseSinkDTO.getTableInfo(sinkDTO, columnList);
        try (Connection conn = OceanBaseJdbcUtils.getConnection(sinkDTO.getJdbcUrl(), sinkDTO.getUsername(),
                sinkDTO.getPassword())) {
            // 1. create database if not exists
            OceanBaseJdbcUtils.createDb(conn, tableInfo.getDbName());
            // 2. table not exists, create it
            OceanBaseJdbcUtils.createTable(conn, tableInfo);
            // 3. table exists, add columns - skip the exists columns
            OceanBaseJdbcUtils.addColumns(conn, tableInfo.getDbName(), tableInfo.getTableName(), columnList);

            // 4. update the sink status to success
            String info = "success to create OceanBase resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
        } catch (Throwable e) {
            String errMsg = "create OceanBase 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 OceanBase table for data sink [" + sinkInfo.getId() + "]");
    }
Create OceanBase table by SinkInfo.

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