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

        final OracleSinkDTO oracleSink = OracleSinkDTO.getFromJson(sinkInfo.getExtParams());
        final OracleTableInfo tableInfo = OracleSinkDTO.getTableInfo(oracleSink, columnList);

        try (Connection conn = OracleJdbcUtils.getConnection(oracleSink.getJdbcUrl(),
                oracleSink.getUsername(), oracleSink.getPassword())) {

            // In Oracle, there is no need to consider whether the database exists
            // 1.If table not exists, create it
            OracleJdbcUtils.createTable(conn, tableInfo);
            // 2. table exists, add columns - skip the exists columns
            OracleJdbcUtils.addColumns(conn, tableInfo.getTableName(), columnList);
            // 3. update the sink status to success
            final String info = "success to create Oracle resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
        } catch (Throwable e) {
            String errMsg = "create Oracle 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 Oracle table for data sink [" + sinkInfo.getId() + "]");
    }
Create Oracle table by SinkInfo.

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