@Override
    public String getMetadataKey(MetaField metaField) {
        String metadataKey;
        switch (metaField) {
            case TABLE_NAME:
                metadataKey = "meta.table_name";
                break;
            case DATABASE_NAME:
                metadataKey = "meta.database_name";
                break;
            case SCHEMA_NAME:
                metadataKey = "meta.schema_name";
                break;
            case OP_TS:
                metadataKey = "meta.op_ts";
                break;
            case OP_TYPE:
                metadataKey = "meta.op_type";
                break;
            case DATA:
            case DATA_BYTES:
                metadataKey = "meta.data";
                break;
            case DATA_CANAL:
            case DATA_BYTES_CANAL:
                metadataKey = "meta.data_canal";
                break;
            case DATA_DEBEZIUM:
            case DATA_BYTES_DEBEZIUM:
                metadataKey = "meta.data_debezium";
                break;
            case IS_DDL:
                metadataKey = "meta.is_ddl";
                break;
            case TS:
                metadataKey = "meta.ts";
                break;
            case SQL_TYPE:
                metadataKey = "meta.sql_type";
                break;
            case PK_NAMES:
                metadataKey = "meta.pk_names";
                break;
            default:
                throw new UnsupportedOperationException(
                        String.format("Unsupport meta field for %s: %s", this.getClass().getSimpleName(), metaField));
        }
        return metadataKey;
    }
generate table options

@return options
@see <a href="https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html">postgres
cdc</a>
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/PostgresExtractNode.java