private TableChange getTableChange(SourceRecord record) throws Exception {
        Envelope.Operation op = Envelope.operationFor(record);
        Schema valueSchema;
        if (op == Envelope.Operation.DELETE) {
            valueSchema = record.valueSchema().field(Envelope.FieldName.BEFORE).schema();
        } else {
            valueSchema = record.valueSchema().field(FieldName.AFTER).schema();
        }
        List<Field> fields = valueSchema.fields();

        TableId tableId = RecordUtils.getTableId(record);
        String schema = tableId.schema();
        String table = tableId.table();
        String id = tableId.identifier();

        TableChange tableChange;
        TableChangeHolder holder;
        if (!tableChangeMap.containsKey(id)) {
            List<Map<String, Object>> columns = this.postgreSQLQueryVisitor.getTableColumnsMetaData(schema, table);
            LOG.info("columns: {}", columns);
            tableChange = initTableChange(tableId, TableChangeType.CREATE, columns);
            holder = new TableChangeHolder();
            holder.tableChange = tableChange;
            holder.timestamp = System.currentTimeMillis();
            tableChangeMap.put(id, holder);
        } else {
            holder = tableChangeMap.get(id);
            tableChange = holder.tableChange;
            // diff fieldMap and columns of tableChange, to see if new column has been added
            if (System.currentTimeMillis() - holder.timestamp > DDL_UPDATE_INTERVAL
                    && holder.tableChange.getTable() instanceof TableImpl) {
                TableImpl tableImpl = (TableImpl) holder.tableChange.getTable();
                List<Column> columnList = tableImpl.columns();
                // only support add or remove columns
                if (columnList.size() != fields.size()) {
                    List<Map<String, Object>> columns = this.postgreSQLQueryVisitor.getTableColumnsMetaData(schema,
                            table);
                    LOG.info("columns: {}", columns);
                    tableChange = initTableChange(tableId, TableChangeType.CREATE, columns);
                    // update tableChange
                    holder.tableChange = tableChange;
                }
                holder.timestamp = System.currentTimeMillis();
                tableChangeMap.put(id, holder);
            }
        }
        return tableChange;
    }
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