-- Example DDL. Note: This FLIP does not yet expose the constraint via DDL syntax;
-- it only introduces the constraint at the Schema. The example below
-- is provided solely for illustrative purposes.
CREATE TABLE src (
  a int,
  b string,
  c bigint,
  d double,
  PRIMARY KEY (a) NOT ENFORCED,
  CONSTRAINT no_up_cols COLUMNS (b, c) IMMUTABLE NOT ENFORCED
)

For the same PK (e.g., a = 1), once values for columns b and c are written, subsequent updates must not alter them.
Deletion operations are also prohibited (since deletion is equivalent to “clearing” immutable columns).

Examples:
INSERT[1, 's1', 1L, 1.0] → Valid
UPDATE[1, 's1', 2L, 2.0] → Invalid（c changed）
UPDATE[1, 's1', 1L, 4.0] → Valid（b and c unchanged）
DELETE[1, 's1', 1L, 1.0] → Invalid

    /**
     * Type of the constraint.
     *
     * <ul>
     *   <li>Unique constraints:
     *       <ul>
     *         <li>UNIQUE - is satisfied if and only if there do not exist two rows that have same
     *             non-null values in the unique columns
     *         <li>PRIMARY KEY - additionally to UNIQUE constraint, it requires none of the values
     *             in specified columns be a null value. Moreover there can be only a single PRIMARY
     *             KEY defined for a Table.
     *       </ul>
     *   <li>Immutable constraints - is satisfied these specific columns corresponding each pk will
     *       not be modified with new values.
     * </ul>
     */
    @PublicEvolving
    enum ConstraintType {
        PRIMARY_KEY,
        UNIQUE_KEY,
        IMMUTABLE_COLUMNS
    }

/**
 * A constraint for immutable columns. All columns within each pk in this constraint will not be
 * modified with new values.
 *
 * @see ConstraintType
 */
@PublicEvolving
public final class ImmutableColumnsConstraint extends AbstractConstraint {

    private final List<String> columns;

    public ImmutableColumnsConstraint(String name, boolean enforced, List<String> columns) {
        super(name, enforced);
        this.columns = columns;
    }

    /** Creates a non enforced {@link ConstraintType#IMMUTABLE_COLUMNS} constraint. */
    public static ImmutableColumnsConstraint immutableColumns(String name, List<String> columns) {
        return new ImmutableColumnsConstraint(name, false, columns);
    }

    public List<String> getColumns() {
        return columns;
    }

    @Override
    public ConstraintType getType() {
        return ConstraintType.IMMUTABLE_COLUMNS;
    }

    @Override
    public String asSummaryString() {
        return String.format(
                "CONSTRAINT %s COLUMNS (%s) IMMUTABLE%s",
                EncodingUtils.escapeIdentifier(getName()),
                columns.stream()
                        .map(EncodingUtils::escapeIdentifier)
                        .collect(Collectors.joining(", ")),
                isEnforced() ? "" : " NOT ENFORCED");
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        if (!super.equals(o)) {
            return false;
        }
        ImmutableColumnsConstraint that = (ImmutableColumnsConstraint) o;
        return Objects.equals(columns, that.columns);
    }

    @Override
    public int hashCode() {
        return Objects.hash(super.hashCode(), columns);
    }
}

@PublicEvolving
public final class Schema {
  ...
  private final @Nullable UnresolvedImmutableColumns immutableColumns;
  ...
  public Schema(
            List<UnresolvedColumn> columns,
            List<UnresolvedWatermarkSpec> watermarkSpecs,
            @Nullable UnresolvedPrimaryKey primaryKey,
            List<UnresolvedIndex> indexes,
            @Nullable UnresolvedImmutableColumns immutableColumns) {
        this.columns = Collections.unmodifiableList(columns);
        this.watermarkSpecs = Collections.unmodifiableList(watermarkSpecs);
        this.primaryKey = primaryKey;
        this.indexes = Collections.unmodifiableList(indexes);
        this.immutableColumns = immutableColumns;
    }

  ... (Omit `getter`, `toString`, `equals` and `hashCode`)

    /** A builder for constructing an immutable but still unresolved {@link Schema}. */
    @PublicEvolving
    public static final class Builder {

      ...
      private @Nullable UnresolvedImmutableColumns immutableColumns;
      ...

      
        /**
         * Declares an immutable columns constraint for a list of given columns. Immutable columns
         * constraint is used to identify which columns in a table are not allowed to be modified.
         * Currently, this constraint is informational only and is not enforced. It can be utilized
         * for optimization purposes. It is the responsibility of the data owner to ensure that
         * these columns are unmodified.
         *
         * <p>The immutable columns will be assigned a generated name in the format {@code
         * IMMUTABLE_COLUMNS_col1_col2}.
         *
         * @param columnNames columns that form the constraint for immutable columns
         */
        public Builder immutableColumns(String... columnNames) {
            Preconditions.checkNotNull(columnNames, "Immutable column names must not be null.");
            return immutableColumns(Arrays.asList(columnNames));
        }

        /**
         * Declares an immutable columns constraint for a list of given columns. Immutable columns
         * constraint is used to identify which columns in a table are not allowed to be modified.
         * Currently, this constraint is informational only and is not enforced. It can be utilized
         * for optimization purposes. It is the responsibility of the data owner to ensure that
         * these columns are unmodified.
         *
         * <p>The immutable columns will be assigned a generated name in the format {@code
         * IMMUTABLE_COLUMNS_col1_col2}.
         *
         * @param columnNames columns that form the constraint for immutable columns
         */
        public Builder immutableColumns(List<String> columnNames) {
            Preconditions.checkNotNull(columnNames, "Immutable column names must not be null.");
            final String generatedConstraintName =
                    columnNames.stream().collect(Collectors.joining("_", "IMMUTABLE_COLUMNS_", ""));
            return immutableColumnsNamed(generatedConstraintName, columnNames);
        }

        /**
         * Declares an immutable columns constraint for a list of given columns. Immutable columns
         * constraint is used to identify which columns in a table are not allowed to be modified.
         * Currently, this constraint is informational only and is not enforced. It can be utilized
         * for optimization purposes. It is the responsibility of the data owner to ensure that
         * these columns are unmodified.
         *
         * @param constraintName name for the immutable columns constraint, can be used to reference
         *     this constraint
         * @param columnNames columns that form the constraint for immutable columns
         */
        public Builder immutableColumnsNamed(String constraintName, String... columnNames) {
            Preconditions.checkNotNull(columnNames, "Immutable column names must not be null.");
            return immutableColumnsNamed(constraintName, Arrays.asList(columnNames));
        }

        /**
         * Declares an immutable columns constraint for a list of given columns. Immutable columns
         * constraint is used to identify which columns in a table are not allowed to be modified.
         * Currently, this constraint is informational only and is not enforced. It can be utilized
         * for optimization purposes. It is the responsibility of the data owner to ensure that
         * these columns are unmodified.
         *
         * @param constraintName name for the immutable columns constraint, can be used to reference
         *     this constraint
         * @param columnNames columns that form the constraint for immutable columns
         */
        public Builder immutableColumnsNamed(String constraintName, List<String> columnNames) {
            Preconditions.checkState(
                    immutableColumns == null,
                    "Multiple constraints for immutable columns are not supported.");
            Preconditions.checkNotNull(
                    constraintName, "Immutable column constraint name must not be null.");
            Preconditions.checkArgument(
                    !StringUtils.isNullOrWhitespaceOnly(constraintName),
                    "The constraint name for immutable columns must not be empty.");
            Preconditions.checkArgument(
                    columnNames != null && !columnNames.isEmpty(),
                    "The constraint for immutable columns must be defined for at least a single column.");
            immutableColumns = new UnresolvedImmutableColumns(constraintName, columnNames);
            return this;
        }

        /** Returns an instance of an unresolved {@link Schema}. */
        public Schema build() {
            return new Schema(columns, watermarkSpecs, primaryKey, indexes, immutableColumns);
        }
      
      ...
    }
  
  ...
  

    /**
     * Declaration of a list of immutable columns that will be resolved to {@link
     * ImmutableColumnsConstraint} during schema resolution.
     */
    @PublicEvolving
    public static final class UnresolvedImmutableColumns extends UnresolvedConstraint {

        private final List<String> columnNames;

        public UnresolvedImmutableColumns(String constraintName, List<String> columnNames) {
            super(constraintName);
            this.columnNames = columnNames;
        }

        public List<String> getColumnNames() {
            return columnNames;
        }

        @Override
        public String toString() {
            return String.format(
                    "%s COLUMNS (%s) IMMUTABLE NOT ENFORCED",
                    super.toString(),
                    columnNames.stream()
                            .map(EncodingUtils::escapeIdentifier)
                            .collect(Collectors.joining(", ")));
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            if (!super.equals(o)) {
                return false;
            }
            UnresolvedImmutableColumns that = (UnresolvedImmutableColumns) o;
            return columnNames.equals(that.columnNames);
        }

        @Override
        public int hashCode() {
            return Objects.hash(super.hashCode(), columnNames);
        }
    }
}

@PublicEvolving
public final class ResolvedSchema {
    ...
    private final @Nullable ImmutableColumnsConstraint immutableColumns;
    ...

    public ResolvedSchema(
            List<Column> columns,
            List<WatermarkSpec> watermarkSpecs,
            @Nullable UniqueConstraint primaryKey,
            List<Index> indexes,
            @Nullable ImmutableColumnsConstraint immutableColumns) {
        this.columns = Preconditions.checkNotNull(columns, "Columns must not be null.");
        this.watermarkSpecs =
                Preconditions.checkNotNull(watermarkSpecs, "Watermark specs must not be null.");
        this.primaryKey = primaryKey;
        this.indexes = Preconditions.checkNotNull(indexes, "Indexes must not be null.");
        this.immutableColumns = immutableColumns;

        Preconditions.checkArgument(
                primaryKey != null || immutableColumns == null,
                "Constraint for immutable columns must be defined on pk table");
    }

    ...
    /** Returns the constraint about immutable columns if it has been defined. */
    public Optional<ImmutableColumnsConstraint> getImmutableColumns() {
        return Optional.ofNullable(immutableColumns);
    }

    ...
    /**
     * Returns the indexes of fields about the immutable constraint in the {@link
     * #toPhysicalRowDataType()}, if any, otherwise returns an empty array.
     */
    public int[] getImmutableColumnIndexes() {
        final List<String> columns =
                getColumns().stream()
                        .filter(Column::isPhysical)
                        .map(Column::getName)
                        .collect(Collectors.toList());
        return getImmutableColumns()
                .map(ImmutableColumnsConstraint::getColumns)
                .map(
                        immutableColumns ->
                                immutableColumns.stream().mapToInt(columns::indexOf).toArray())
                .orElseGet(() -> new int[] {});
    }

    ... (Omit `getter`, `toString`, `equals` and `hashCode`)

}

[CONSTRAINT no_up_cols] CHECK ON UPDATE (NEW.col1 = OLD.col1, NEW.col2 = OLD.col2 ...) NOT ENFORCED