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

        GreenplumSinkDTO greenplumSink = GreenplumSinkDTO.getFromJson(sinkInfo.getExtParams());
        GreenplumTableInfo tableInfo = GreenplumSinkDTO.getTableInfo(greenplumSink, columnList);
        if (StringUtils.isEmpty(tableInfo.getSchemaName())) {
            tableInfo.setSchemaName(GREENPLUM_DEFAULT_SCHEMA);
        }
        try (Connection conn = GreenplumJdbcUtils.getConnection(greenplumSink.getJdbcUrl(), greenplumSink.getUsername(),
                greenplumSink.getPassword())) {
            // 1.If schema not exists,create it
            GreenplumJdbcUtils.createSchema(conn, tableInfo.getTableName(), tableInfo.getUserName());
            // 2.If table not exists, create it
            GreenplumJdbcUtils.createTable(conn, tableInfo);
            // 3.Table exists, add columns - skip the exists columns
            GreenplumJdbcUtils.addColumns(conn, tableInfo.getSchemaName(), tableInfo.getTableName(), columnList);
            // 4.Update the sink status to success
            final String info = "success to create Greenplum resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
            // 4. close connection.
        } catch (Throwable e) {
            String errMsg = "create Greenplum 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 Greenplum table for data sink [" + sinkInfo.getId() + "]");
    }
Create Greenplum table by SinkInfo.

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