[core][spark] Support generated columns

docs/docs/concepts/spec/schema.md

@@ -115,10 +115,63 @@ A DataField describes one column, including fields nested inside a row type.
 | `type` | String or JSON type object | Logical type, including nullability and nested type information. |
 | `description` | String, optional | Field comment. |
 | `defaultValue` | String, optional | Stored default value definition. Engine support determines how it is used. |
+| `generationExpression` | String, optional | SQL scalar expression generating a physical top-level column. Mutually exclusive with `defaultValue`. |
 
 Primitive types commonly use strings such as `BIGINT NOT NULL`. Nested types need their
 structured type representation. See [Data Types](../data-types) for the logical type reference.
 
+### Generated Columns
+
+A top-level field, including a partition field, may define `generationExpression`, for example:
+
+```json
+{
+  "id": 2,
+  "name": "dt",
+  "type": "DATE",
+  "generationExpression": "CAST(event_time AS DATE)"
+}
+```
+
+Writers compile expressions for omitted columns into Paimon `Transform` operations once at
+initialization. These operations define the automatically generated value, including CAST
+formatting and invalid-input behavior; the write engine's ANSI mode or implicit CAST rules do
+not change that definition. Spark fills omitted columns through the same evaluator, including
+source-column defaults, before writing.
+Expressions may reference ordinary top-level fields and compose supported scalar calls, such as
+`YEAR(event_time)`, `UPPER(name)`, `CONCAT(UPPER(name), '-', name)`, and
+`DATE_ADD(CAST(event_time AS DATE), days)`. The expression result type must match the declared
+column type. Nested field references, references to other generated columns, arbitrary SQL,
+UDFs, and nondeterministic functions are not supported. `TIMESTAMP WITH LOCAL TIME ZONE` inputs
+and cast targets are rejected because their results can depend on a writer's default time zone.
+
+Generated columns are physical fields; readers use the stored values without evaluating the
+expression. A cast from `TIMESTAMP WITHOUT TIME ZONE` to `DATE` uses the timestamp's calendar
+date, independently of the writer's default time zone.
+
+This is a write-time column-filling feature, not a strict computed-column constraint. Adapted
+writers evaluate an expression only when the input schema omits its column. If the input schema
+includes the column, its supplied value is preserved without evaluating or checking the expression,
+including an explicit NULL. Normal type, NOT NULL, primary-key, and partition constraints still
+apply. Stored values may therefore differ from the expression result; readers and optimizers must
+not assume that the expression is an invariant or infer partition filters from it. Updating a source
+column does not recompute a generated column that is already supplied in the write row.
+
+Definitions are supported at table creation. Changing a definition or renaming, dropping, or
+changing the type of any source is not supported. A generated column cannot be used as
+`rowkind.field`.
+
+All primary-key merge engines are supported. Column filling happens before merging; the merge
+engine applies its normal rules to generated columns without recomputing expressions after a merge.
+For example, a partial-update merge can retain an existing generated value when its source changes.
+Data Evolution partial-column updates write only the selected columns. Unselected generated
+columns keep their existing values, and selected generated columns accept the supplied values
+without requiring their expression's source columns.
+
+Use clients supporting this attribute for writes; older clients may ignore it. Its presence does
+not establish compatibility with older writers. See [Spark DDL](../../spark/sql-ddl#generated-columns)
+for SQL examples.
+
 ## Update Schema
 
 A schema update writes a new schema file rather than replacing the old definition:

docs/docs/spark/sql-ddl.md

@@ -90,6 +90,66 @@ TBLPROPERTIES ('primary-key' = 'dt,hh,user_id');
 Omit `primary-key` to create an append table. For storage layout and table options, see
 [Primary Key Tables](../primary-key-table/) and [Append Tables](../append-table/).
 
+### Generated Columns
+
+On Spark 3.5 and Spark 4 with `org.apache.paimon.spark.SparkCatalog`, define physical top-level columns
+with `GENERATED ALWAYS AS`:
+
+```sql
+CREATE TABLE events (
+    id BIGINT,
+    name STRING,
+    event_time TIMESTAMP_NTZ,
+    dt DATE GENERATED ALWAYS AS (CAST(event_time AS DATE)),
+    event_year INT GENERATED ALWAYS AS (YEAR(event_time)),
+    label STRING GENERATED ALWAYS AS (UPPER(name))
+) USING paimon PARTITIONED BY (dt);
+
+INSERT INTO events (id, name, event_time)
+VALUES (1, 'event', TIMESTAMP_NTZ '2026-10-30 12:00:00');
+```
+
+The writer fills omitted generated columns before partition routing. Explicitly supplied values,
+including NULL, are preserved without evaluating or checking the expression. Normal type and
+NOT NULL constraints still apply. Although Spark uses `GENERATED ALWAYS AS` syntax, Paimon
+provides automatic filling of omitted columns, not a strict computed-column constraint: stored
+values need not match the expression, and changing a source does not recompute a supplied column.
+Expressions are stored in the column's schema definition and evaluated with Paimon transforms.
+Spark uses the same evaluator as the core writer for omitted columns, applying source-column
+defaults first. CAST formatting, conversion errors, and argument bounds follow Paimon transforms
+rather than session-dependent Spark rules. The two-argument TRIM/LTRIM/RTRIM form takes
+`(characters, string)`.
+Ordinary append inserts, all primary-key merge engines, and Data Evolution partial-column updates
+are supported; existing primary-key, partition, and merge-engine rules still apply. Generated
+values are filled before primary-key merging and are not recomputed after the merge. A partial
+update preserves unselected generated columns, even when their source columns change.
+
+Supported calls include `CAST`, `YEAR`, `MONTH`, `DAY`/`DAYOFMONTH`, `HOUR`, `MINUTE`, `SECOND`,
+`QUARTER`, `DAYOFWEEK`, `DAYOFYEAR`, `WEEKDAY`, `WEEKOFYEAR`, `DATE_ADD`, `DATEDIFF`/`DATE_DIFF`,
+`TRUNC`, `UPPER`, `LOWER`, `CONCAT`, `CONCAT_WS`, `SUBSTRING`/`SUBSTR`, `LENGTH`/`CHAR_LENGTH`,
+`BIT_LENGTH`, `TRANSLATE`, `OVERLAY`, `LPAD`, `RPAD`, `TRIM`, `LTRIM`, and `RTRIM`.
+Calls use comma-separated arguments; CAST uses `CAST(expression AS type)`. String and integer
+literals, NULL arguments, DATE literals, and nested calls are supported. Only ordinary top-level
+columns may be referenced. The result type must match the declared column type. Nested field
+references, generated-column dependencies, nondeterministic functions, and `TIMESTAMP_LTZ`
+inputs or cast targets are not supported. A generated column cannot be used as `rowkind.field`.
+Spark also rejects automatic generation when the result contains a TIMESTAMP with precision
+greater than 6, including inside a top-level complex column, because Spark cannot preserve its
+nanoseconds. Automatic generation of BLOB results is also rejected because the Spark row
+representation cannot preserve the distinction between a BLOB descriptor and inline content
+through this evaluator.
+
+The Java schema builder exposes `generatedColumn(name, type, expression)`. Java writers can call
+`withWriteType` with all ordinary fields to omit generated fields from input rows. The
+writer fills in the generated fields and writes the full table schema. If ordinary fields are
+also omitted, `withWriteType` uses the underlying writer's partial-column behavior and generates
+no omitted columns. For Data Evolution partial updates, this preserves unselected column files.
+Flink's existing virtual computed columns retain their original semantics; this feature does not
+add Flink SQL syntax for physical generated columns.
+
+Generated definitions and the names and types of their dependencies cannot be changed after
+creation. Use writers that understand this schema attribute; older writers may ignore it.
+
 ### Manage Format Table Partitions
 
 See [Format Table Partitions](./format-table) for partition registration, custom locations,

paimon-api/src/main/java/org/apache/paimon/schema/Schema.java

@@ -185,7 +185,8 @@ private static List<DataField> normalizeFields(
                                 field.name(),
                                 field.type().copy(primaryKeyNullable),
                                 field.description(),
-                                field.defaultValue()));
+                                field.defaultValue(),
+                                field.generationExpression()));
             } else {
                 newFields.add(field);
             }
@@ -332,12 +333,36 @@ public Builder column(
                 DataType dataType,
                 @Nullable String description,
                 @Nullable String defaultValue) {
+            return column(columnName, dataType, description, defaultValue, null);
+        }
+
+        /** Declares a physical generated top-level column. */
+        public Builder generatedColumn(
+                String columnName, DataType dataType, String generationExpression) {
+            Preconditions.checkNotNull(
+                    generationExpression, "Generation expression must not be null.");
+            return column(columnName, dataType, null, null, generationExpression);
+        }
+
+        public Builder column(
+                String columnName,
+                DataType dataType,
+                @Nullable String description,
+                @Nullable String defaultValue,
+                @Nullable String generationExpression) {
             Preconditions.checkNotNull(columnName, "Column name must not be null.");
             Preconditions.checkNotNull(dataType, "Data type must not be null.");
 
             int id = highestFieldId.incrementAndGet();
             DataType reassignDataType = ReassignFieldId.reassign(dataType, highestFieldId);
-            columns.add(new DataField(id, columnName, reassignDataType, description, defaultValue));
+            columns.add(
+                    new DataField(
+                            id,
+                            columnName,
+                            reassignDataType,
+                            description,
+                            defaultValue,
+                            generationExpression));
             return this;
         }
 

paimon-api/src/main/java/org/apache/paimon/types/DataField.java

@@ -47,6 +47,7 @@ public final class DataField implements Serializable {
     private final DataType type;
     private final @Nullable String description;
     private final @Nullable String defaultValue;
+    private final @Nullable String generationExpression;
 
     public DataField(int id, String name, DataType dataType) {
         this(id, name, dataType, null, null);
@@ -62,11 +63,22 @@ public DataField(
             DataType type,
             @Nullable String description,
             @Nullable String defaultValue) {
+        this(id, name, type, description, defaultValue, null);
+    }
+
+    public DataField(
+            int id,
+            String name,
+            DataType type,
+            @Nullable String description,
+            @Nullable String defaultValue,
+            @Nullable String generationExpression) {
         this.id = id;
         this.name = name;
         this.type = type;
         this.description = description;
         this.defaultValue = defaultValue;
+        this.generationExpression = generationExpression;
     }
 
     public int id() {
@@ -82,23 +94,27 @@ public DataType type() {
     }
 
     public DataField newId(int newId) {
-        return new DataField(newId, name, type, description, defaultValue);
+        return new DataField(newId, name, type, description, defaultValue, generationExpression);
     }
 
     public DataField newName(String newName) {
-        return new DataField(id, newName, type, description, defaultValue);
+        return new DataField(id, newName, type, description, defaultValue, generationExpression);
     }
 
     public DataField newType(DataType newType) {
-        return new DataField(id, name, newType, description, defaultValue);
+        return new DataField(id, name, newType, description, defaultValue, generationExpression);
     }
 
     public DataField newDescription(String newDescription) {
-        return new DataField(id, name, type, newDescription, defaultValue);
+        return new DataField(id, name, type, newDescription, defaultValue, generationExpression);
     }
 
     public DataField newDefaultValue(String newDefaultValue) {
-        return new DataField(id, name, type, description, newDefaultValue);
+        return new DataField(id, name, type, description, newDefaultValue, generationExpression);
+    }
+
+    public DataField newGenerationExpression(@Nullable String expression) {
+        return new DataField(id, name, type, description, defaultValue, expression);
     }
 
     @Nullable
@@ -111,17 +127,28 @@ public String defaultValue() {
         return defaultValue;
     }
 
+    /** Returns the expression generating this physical column, or null for an ordinary column. */
+    @Nullable
+    public String generationExpression() {
+        return generationExpression;
+    }
+
     public DataField copy() {
-        return new DataField(id, name, type.copy(), description, defaultValue);
+        return new DataField(
+                id, name, type.copy(), description, defaultValue, generationExpression);
     }
 
     public DataField copy(boolean isNullable) {
-        return new DataField(id, name, type.copy(isNullable), description, defaultValue);
+        return new DataField(
+                id, name, type.copy(isNullable), description, defaultValue, generationExpression);
     }
 
     public String asSQLString() {
         StringBuilder sb = new StringBuilder();
         sb.append(escapeIdentifier(name)).append(" ").append(type.asSQLString());
+        if (generationExpression != null) {
+            sb.append(" GENERATED ALWAYS AS (").append(generationExpression).append(")");
+        }
         if (StringUtils.isNotEmpty(description)) {
             sb.append(" COMMENT '").append(escapeSingleQuotes(description)).append("'");
         }
@@ -143,6 +170,9 @@ public void serializeJson(JsonGenerator generator) throws IOException {
         if (defaultValue() != null) {
             generator.writeStringField("defaultValue", defaultValue());
         }
+        if (generationExpression != null) {
+            generator.writeStringField("generationExpression", generationExpression);
+        }
         generator.writeEndObject();
     }
 
@@ -159,7 +189,8 @@ public boolean equals(Object o) {
                 && Objects.equals(name, field.name)
                 && Objects.equals(type, field.type)
                 && Objects.equals(description, field.description)
-                && Objects.equals(defaultValue, field.defaultValue);
+                && Objects.equals(defaultValue, field.defaultValue)
+                && Objects.equals(generationExpression, field.generationExpression);
     }
 
     public boolean equalsIgnoreFieldId(DataField other) {
@@ -172,7 +203,8 @@ public boolean equalsIgnoreFieldId(DataField other) {
         return Objects.equals(name, other.name)
                 && type.equalsIgnoreFieldId(other.type)
                 && Objects.equals(description, other.description)
-                && Objects.equals(defaultValue, other.defaultValue);
+                && Objects.equals(defaultValue, other.defaultValue)
+                && Objects.equals(generationExpression, other.generationExpression);
     }
 
     public boolean isPrunedFrom(DataField other) {
@@ -186,12 +218,13 @@ public boolean isPrunedFrom(DataField other) {
                 && Objects.equals(name, other.name)
                 && type.isPrunedFrom(other.type)
                 && Objects.equals(description, other.description)
-                && Objects.equals(defaultValue, other.defaultValue);
+                && Objects.equals(defaultValue, other.defaultValue)
+                && Objects.equals(generationExpression, other.generationExpression);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, name, type, description, defaultValue);
+        return Objects.hash(id, name, type, description, defaultValue, generationExpression);
     }
 
     @Override
@@ -210,7 +243,9 @@ public static boolean dataFieldEqualsIgnoreId(DataField dataField1, DataField da
             return Objects.equals(dataField1.name(), dataField2.name())
                     && Objects.equals(dataField1.type(), dataField2.type())
                     && Objects.equals(dataField1.description(), dataField2.description())
-                    && Objects.equals(dataField1.defaultValue(), dataField2.defaultValue());
+                    && Objects.equals(dataField1.defaultValue(), dataField2.defaultValue())
+                    && Objects.equals(
+                            dataField1.generationExpression(), dataField2.generationExpression());
         } else {
             return false;
         }

paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java

@@ -70,7 +70,16 @@ public static DataField parseDataField(JsonNode json, AtomicInteger fieldId) {
         if (defaultValueNode != null) {
             defaultValue = defaultValueNode.asText();
         }
-        return new DataField(id, name, type, description, defaultValue);
+        JsonNode generationExpression = json.get("generationExpression");
+        return new DataField(
+                id,
+                name,
+                type,
+                description,
+                defaultValue,
+                generationExpression == null || generationExpression.isNull()
+                        ? null
+                        : generationExpression.asText());
     }
 
     public static DataType parseDataType(JsonNode json) {

paimon-api/src/main/java/org/apache/paimon/types/ReassignFieldId.java

@@ -70,7 +70,8 @@ public DataType visit(RowType rowType) {
                                         f.name(),
                                         f.type().accept(this),
                                         f.description(),
-                                        f.defaultValue()));
+                                        f.defaultValue(),
+                                        f.generationExpression()));
         return builder.build();
     }
 

paimon-api/src/main/java/org/apache/paimon/types/RowType.java

@@ -644,9 +644,23 @@ public Builder field(
                 DataType type,
                 @Nullable String description,
                 @Nullable String defaultValue) {
+            return field(name, type, description, defaultValue, null);
+        }
+
+        public Builder field(
+                String name,
+                DataType type,
+                @Nullable String description,
+                @Nullable String defaultValue,
+                @Nullable String generationExpression) {
             fields.add(
                     new DataField(
-                            fieldId.incrementAndGet(), name, type, description, defaultValue));
+                            fieldId.incrementAndGet(),
+                            name,
+                            type,
+                            description,
+                            defaultValue,
+                            generationExpression));
             return this;
         }