private TableChange readTableSchema(TableId tableId) throws SQLException {

        final PostgresOffsetContext offsetContext =
                PostgresOffsetContext.initialContext(dbzConfig, jdbcConnection, Clock.SYSTEM);

        // set the events to populate proper sourceInfo into offsetContext
        offsetContext.event(tableId, Instant.now());

        Tables tables = new Tables();
        try {
            jdbcConnection.readSchema(
                    tables,
                    dbzConfig.databaseName(),
                    tableId.schema(),
                    dbzConfig.getTableFilters().dataCollectionFilter(),
                    null,
                    false);
        } catch (SQLException e) {
            throw new FlinkRuntimeException("Failed to read schema", e);
        }

        Table table = Objects.requireNonNull(tables.forTable(tableId));

        // TODO: check whether we always set isFromSnapshot = true
        SchemaChangeEvent schemaChangeEvent =
                new SchemaChangeEvent(
                        offsetContext.getPartition(),
                        offsetContext.getOffset(),
                        offsetContext.getSourceInfo(),
                        dbzConfig.databaseName(),
                        tableId.schema(),
                        null,
                        table,
                        SchemaChangeEvent.SchemaChangeEventType.CREATE,
                        true);

        for (TableChanges.TableChange tableChange : schemaChangeEvent.getTableChanges()) {
            this.schemasByTableId.put(tableId, tableChange);
        }
        return this.schemasByTableId.get(tableId);
    }
A PgSchema similar to PostgresSchema with customization.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/utils/PgSchema.java