Repository: apache/flink
File: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/serde/CompiledPlanSerdeUtil.java (Java)

Code before:
            Class<? extends V> valueType,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return ctx.readValue(
                    traverse(objectNode.get(fieldName), codec),
                    ctx.getTypeFactory().constructMapType(Map.class, keyType, valueType));
        }
        return Map.of();
    }

    static <T> Optional<T> deserializeOptionalField(
            ObjectNode objectNode,
            String fieldName,
            JavaType type,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return Optional.of(ctx.readValue(traverse(objectNode.get(fieldName), codec), type));
        }
        return Optional.empty();
    }

    static <T> T deserializeFieldOrNull(
            ObjectNode objectNode,
            String fieldName,
            Class<T> type,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return ctx.readValue(traverse(objectNode.get(fieldName), codec), type);
        }
        return null;
    }

    static boolean areColumnsEqual(
            ResolvedSchema schemaFromPlan, ResolvedSchema schemaFromCatalog) {
        // For schema equality we check:
        //  * Columns size and order
        //  * For each column: name, kind (class) and type
        final List<Column> columnsFromPlan = schemaFromPlan.getColumns();
        final List<Column> columnsFromCatalog = schemaFromCatalog.getColumns();

        if (columnsFromPlan.size() != columnsFromCatalog.size()) {
            return false;
        }

        for (int i = 0; i < columnsFromPlan.size(); i++) {

Review comment (suggestion):
```suggestion
    static <T> @Nullable T deserializeFieldOrNull(
```

Code after:
            String fieldName,
            Class<? extends K> keyType,
            Class<? extends V> valueType,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return ctx.readValue(
                    traverse(objectNode.get(fieldName), codec),
                    ctx.getTypeFactory().constructMapType(Map.class, keyType, valueType));
        }
        return Map.of();
    }

    static <T> Optional<T> deserializeOptionalField(
            ObjectNode objectNode,
            String fieldName,
            JavaType type,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return Optional.of(ctx.readValue(traverse(objectNode.get(fieldName), codec), type));
        }
        return Optional.empty();
    }

    static <T> @Nullable T deserializeFieldOrNull(
            ObjectNode objectNode,
            String fieldName,
            Class<T> type,
            ObjectCodec codec,
            DeserializationContext ctx)
            throws IOException {
        if (objectNode.hasNonNull(fieldName)) {
            return ctx.readValue(traverse(objectNode.get(fieldName), codec), type);
        }
        return null;
    }

    static boolean areColumnsEqual(
            ResolvedSchema schemaFromPlan, ResolvedSchema schemaFromCatalog) {
        // For schema equality we check:
        //  * Columns size and order
        //  * For each column: name, kind (class) and type
        final List<Column> columnsFromPlan = schemaFromPlan.getColumns();
        final List<Column> columnsFromCatalog = schemaFromCatalog.getColumns();

        if (columnsFromPlan.size() != columnsFromCatalog.size()) {
            return false;
        }