Repository: apache/flink
File: flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/Schema.java (Java)

Code before:
        }

        // ----------------------------------------------------------------------------------------

        private void addResolvedColumns(List<Column> columns) {
            for (Column c : columns) {
                if (c instanceof PhysicalColumn) {
                    PhysicalColumn physicalColumn = (PhysicalColumn) c;
                    final Builder builder =
                            column(physicalColumn.getName(), physicalColumn.getDataType());
                    c.getComment().ifPresent(builder::withComment);
                } else if (c instanceof ComputedColumn) {
                    ComputedColumn computedColumn = (ComputedColumn) c;
                    final Builder builder =
                            columnByExpression(
                                    computedColumn.getName(), computedColumn.getExpression());
                    c.getComment().ifPresent(builder::withComment);
                } else if (c instanceof MetadataColumn) {
                    MetadataColumn metadataColumn = (MetadataColumn) c;
                    final Builder builder =
                            columnByMetadata(
                                    metadataColumn.getName(),
                                    metadataColumn.getDataType(),
                                    metadataColumn.getMetadataKey().orElse(null),
                                    metadataColumn.isVirtual());
                    c.getComment().ifPresent(builder::withComment);
                }
            }
        }

        private void addResolvedWatermarkSpec(List<WatermarkSpec> specs) {
            specs.forEach(
                    s ->
                            watermarkSpecs.add(
                                    new UnresolvedWatermarkSpec(
                                            s.getRowtimeAttribute(), s.getWatermarkExpression())));
        }

        private void addResolvedConstraint(UniqueConstraint constraint) {
            if (constraint.getType() == Constraint.ConstraintType.PRIMARY_KEY) {
                primaryKeyNamed(constraint.getName(), constraint.getColumns());
            } else {
                throw new IllegalArgumentException("Unsupported constraint type.");
            }
        }

        private void addResolvedIndexes(List<Index> resolvedIndexes) {
            resolvedIndexes.forEach(
                    index -> indexes.add(new UnresolvedIndex(index.getName(), index.getColumns())));
        }
    }

Review comment (refactor):
This line is duplicated in all 3 conditions , are you happy to move it after the if else's ? As these 3 subclasses are the only types of columns. 

Code after:
        }

        // ----------------------------------------------------------------------------------------

        private void addResolvedColumns(List<Column> columns) {
            for (Column c : columns) {
                if (c instanceof PhysicalColumn) {
                    PhysicalColumn physicalColumn = (PhysicalColumn) c;
                    column(physicalColumn.getName(), physicalColumn.getDataType());
                    c.getComment().ifPresent(this::withComment);
                } else if (c instanceof ComputedColumn) {
                    ComputedColumn computedColumn = (ComputedColumn) c;
                    columnByExpression(computedColumn.getName(), computedColumn.getExpression());
                    c.getComment().ifPresent(this::withComment);
                } else if (c instanceof MetadataColumn) {
                    MetadataColumn metadataColumn = (MetadataColumn) c;
                    columnByMetadata(
                            metadataColumn.getName(),
                            metadataColumn.getDataType(),
                            metadataColumn.getMetadataKey().orElse(null),
                            metadataColumn.isVirtual());
                    c.getComment().ifPresent(this::withComment);
                }
            }
        }

        private void addResolvedWatermarkSpec(List<WatermarkSpec> specs) {
            specs.forEach(
                    s ->
                            watermarkSpecs.add(
                                    new UnresolvedWatermarkSpec(
                                            s.getRowtimeAttribute(), s.getWatermarkExpression())));
        }

        private void addResolvedConstraint(UniqueConstraint constraint) {
            if (constraint.getType() == Constraint.ConstraintType.PRIMARY_KEY) {
                primaryKeyNamed(constraint.getName(), constraint.getColumns());
            } else {
                throw new IllegalArgumentException("Unsupported constraint type.");
            }
        }

        private void addResolvedIndexes(List<Index> resolvedIndexes) {
            resolvedIndexes.forEach(
                    index -> indexes.add(new UnresolvedIndex(index.getName(), index.getColumns())));
        }
    }

    // --------------------------------------------------------------------------------------------
    // Helper classes for representing the schema
    // --------------------------------------------------------------------------------------------