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

Code before:
 *
 * @see ContextResolvedTableJsonSerializer for the reverse operation
 */
@Internal
final class ContextResolvedTableJsonDeserializer extends StdDeserializer<ContextResolvedTable> {
    private static final long serialVersionUID = 1L;

    private static final JsonPointer optionsPointer =
            JsonPointer.compile("/" + FIELD_NAME_CATALOG_TABLE + "/" + OPTIONS);

    ContextResolvedTableJsonDeserializer() {
        super(ContextResolvedTable.class);
    }

    @Override
    public ContextResolvedTable deserialize(JsonParser jsonParser, DeserializationContext ctx)
            throws IOException {
        final CatalogPlanRestore planRestoreOption =
                SerdeContext.get(ctx).getConfiguration().get(PLAN_RESTORE_CATALOG_OBJECTS);
        final CatalogManager catalogManager =
                SerdeContext.get(ctx).getFlinkContext().getCatalogManager();
        final ObjectNode objectNode = jsonParser.readValueAsTree();

        // Deserialize the two fields, if available
        final ObjectIdentifier identifier =
                JsonSmileSerdeUtil.deserializeOptionalField(
                                objectNode,
                                FIELD_NAME_IDENTIFIER,
                                ObjectIdentifier.class,
                                jsonParser.getCodec(),
                                ctx)
                        .orElse(null);
        final ResolvedCatalogTable resolvedCatalogTable =
                JsonSmileSerdeUtil.deserializeOptionalField(
                                objectNode,
                                FIELD_NAME_CATALOG_TABLE,
                                ResolvedCatalogTable.class,
                                jsonParser.getCodec(),
                                ctx)
                        .orElse(null);

        if (identifier == null && resolvedCatalogTable == null) {
            throw new TableException(
                    String.format(
                            "The input JSON is invalid because it does neither contain '%s' nor '%s'.",
                            FIELD_NAME_IDENTIFIER, FIELD_NAME_CATALOG_TABLE));
        }

        if (identifier == null) {
            return ContextResolvedTable.anonymous(resolvedCatalogTable);
        }

Review comment (suggestion):
```suggestion
                CompiledPlanSerdeUtil.deserializeOptionalField(
```

Code after:
 *
 * @see ContextResolvedTableJsonSerializer for the reverse operation
 */
@Internal
final class ContextResolvedTableJsonDeserializer extends StdDeserializer<ContextResolvedTable> {
    private static final long serialVersionUID = 1L;

    private static final JsonPointer optionsPointer =
            JsonPointer.compile("/" + FIELD_NAME_CATALOG_TABLE + "/" + OPTIONS);

    ContextResolvedTableJsonDeserializer() {
        super(ContextResolvedTable.class);
    }

    @Override
    public ContextResolvedTable deserialize(JsonParser jsonParser, DeserializationContext ctx)
            throws IOException {
        final CatalogPlanRestore planRestoreOption =
                SerdeContext.get(ctx).getConfiguration().get(PLAN_RESTORE_CATALOG_OBJECTS);
        final CatalogManager catalogManager =
                SerdeContext.get(ctx).getFlinkContext().getCatalogManager();
        final ObjectNode objectNode = jsonParser.readValueAsTree();

        // Deserialize the two fields, if available
        final ObjectIdentifier identifier =
                CompiledPlanSerdeUtil.deserializeOptionalField(
                                objectNode,
                                FIELD_NAME_IDENTIFIER,
                                ObjectIdentifier.class,
                                jsonParser.getCodec(),
                                ctx)
                        .orElse(null);
        final ResolvedCatalogTable resolvedCatalogTable =
                CompiledPlanSerdeUtil.deserializeOptionalField(
                                objectNode,
                                FIELD_NAME_CATALOG_TABLE,
                                ResolvedCatalogTable.class,
                                jsonParser.getCodec(),
                                ctx)
                        .orElse(null);

        if (identifier == null && resolvedCatalogTable == null) {
            throw new TableException(
                    String.format(
                            "The input JSON is invalid because it does neither contain '%s' nor '%s'.",
                            FIELD_NAME_IDENTIFIER, FIELD_NAME_CATALOG_TABLE));
        }

        if (identifier == null) {
            return ContextResolvedTable.anonymous(resolvedCatalogTable);
        }