private TableChange initTableChange(TableId tableId, TableChangeType type, List<Map<String, Object>> columns) {
        List<Column> sortedColumns = new ArrayList<>();
        List<String> pkColumnNames = new ArrayList<>();
        String defaultCharsetName = "utf-8";
        for (Map<String, Object> column : columns) {
            // TODO query charset from table information_schema
            int columnLength = -1;
            Integer characterMaxLen = (Integer) column.get("character_maximum_length");
            if (characterMaxLen != null) {
                columnLength = characterMaxLen;
            }
            Integer columnScale = -1;
            Integer numPrecision = (Integer) column.get("numeric_precision");
            if (numPrecision != null) {
                columnScale = numPrecision;
            }
            boolean optional = "YES".equals(column.get("is_nullable"));
            boolean autoIncremented = false;
            boolean generated = false;

            String columnName = (String) column.get("column_name");
            String constraintType = (String) column.get("constraint_type");
            if (constraintType != null && constraintType.equalsIgnoreCase("PRIMARY KEY")) {
                pkColumnNames.add(columnName);
            }
            int position = (Integer) column.get("ordinal_position");
            String typeName = (String) column.get("data_type");
            int jdbcType = getJdbcType(typeName);
            int componentType = jdbcType;
            String charsetName = "utf-8";
            ColumnImpl columnImpl = new ColumnImpl(columnName, position, jdbcType, componentType, typeName, null,
                    charsetName, defaultCharsetName, columnLength, columnScale, optional, autoIncremented, generated);
            sortedColumns.add(columnImpl);
        }
        TableImpl tableImpl = new TableImpl(tableId, sortedColumns, pkColumnNames, defaultCharsetName);
        return new TableChange(type, tableImpl);
    }
The metric indicates delay from data generation to leaving the source operator.

<p>Note: the metric is available during the binlog phase. Use 0 to indicate the metric is
unavailable.</p>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/debezium/internal/DebeziumChangeFetcher.java