protected RowType extractSchemaNode(JsonNode schema, JsonNode dialectSchema, List<String> pkNames) {
        Iterator<Entry<String, JsonNode>> schemaFields = schema.fields();
        List<RowField> fields = new ArrayList<>();
        while (schemaFields.hasNext()) {
            Entry<String, JsonNode> entry = schemaFields.next();
            String name = entry.getKey();
            LogicalType type = sqlType2FlinkType(entry.getValue().asInt());
            String dialectType = null;
            if (dialectSchema != null && dialectSchema.get(name) != null) {
                dialectType = dialectSchema.get(name).asText();
            }
            try {
                type = handleDialectSqlType(type, dialectType);
            } catch (Exception e) {
                throw new RuntimeException(
                        String.format("Handle dialect sql type failed, the field: %s, the dialect type: %s",
                                name, dialectType),
                        e);
            }
            if (pkNames.contains(name)) {
                type = type.copy(false);
            }
            fields.add(new RowField(name, type));
        }
        return new RowType(fields);
    }
Get opType of data

@param root The json root node
@return The opType of data
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/format/JsonDynamicSchemaFormat.java