diff --git a/docs/docs/concepts/spec/schema.md b/docs/docs/concepts/spec/schema.md index bf007bfbf5aa..44d7ae65f7d0 100644 --- a/docs/docs/concepts/spec/schema.md +++ b/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: diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md index 08d72707ce05..ba7cefc1d94c 100644 --- a/docs/docs/spark/sql-ddl.md +++ b/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, diff --git a/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java b/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java index 5b8be18e4510..17b844ba8536 100644 --- a/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java +++ b/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java @@ -185,7 +185,8 @@ private static List 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; } diff --git a/paimon-api/src/main/java/org/apache/paimon/types/DataField.java b/paimon-api/src/main/java/org/apache/paimon/types/DataField.java index 968de13e15a7..4bf7ad30ea38 100644 --- a/paimon-api/src/main/java/org/apache/paimon/types/DataField.java +++ b/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; } diff --git a/paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java b/paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java index 73ef59e49fe9..ce0be8d89a01 100644 --- a/paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java +++ b/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) { diff --git a/paimon-api/src/main/java/org/apache/paimon/types/ReassignFieldId.java b/paimon-api/src/main/java/org/apache/paimon/types/ReassignFieldId.java index 85f104d99320..21ebf49974d7 100644 --- a/paimon-api/src/main/java/org/apache/paimon/types/ReassignFieldId.java +++ b/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(); } diff --git a/paimon-api/src/main/java/org/apache/paimon/types/RowType.java b/paimon-api/src/main/java/org/apache/paimon/types/RowType.java index a99c966ee5f7..8e4fda242205 100644 --- a/paimon-api/src/main/java/org/apache/paimon/types/RowType.java +++ b/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; } diff --git a/paimon-api/src/test/java/org/apache/paimon/types/GeneratedDataFieldTest.java b/paimon-api/src/test/java/org/apache/paimon/types/GeneratedDataFieldTest.java new file mode 100644 index 000000000000..0be0b1014d34 --- /dev/null +++ b/paimon-api/src/test/java/org/apache/paimon/types/GeneratedDataFieldTest.java @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.types; + +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.TableSchema; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests that physical generated columns retain their definition across schema operations. */ +class GeneratedDataFieldTest { + + @Test + void testSchemaRoundTripAndCopies() { + Schema schema = + Schema.newBuilder() + .column("ts", DataTypes.TIMESTAMP(3)) + .generatedColumn("dt", DataTypes.DATE(), "CAST(ts AS DATE)") + .partitionKeys("dt") + .primaryKey("ts", "dt") + .build(); + DataField generated = schema.fields().get(1); + assertThat(generated.generationExpression()).isEqualTo("CAST(ts AS DATE)"); + TableSchema tableSchema = + new TableSchema( + 0, + schema.fields(), + 1, + schema.partitionKeys(), + schema.primaryKeys(), + Collections.emptyMap(), + null); + assertThat(TableSchema.fromJson(tableSchema.toString())).isEqualTo(tableSchema); + assertThat(generated.copy()).isEqualTo(generated); + assertThat(generated.copy(true).generationExpression()).isEqualTo("CAST(ts AS DATE)"); + assertThat(generated.newId(99).generationExpression()).isEqualTo("CAST(ts AS DATE)"); + assertThat(generated.newName("day").generationExpression()).isEqualTo("CAST(ts AS DATE)"); + assertThat(generated.newType(DataTypes.DATE()).generationExpression()) + .isEqualTo("CAST(ts AS DATE)"); + assertThat(generated.newDescription("partition").generationExpression()) + .isEqualTo("CAST(ts AS DATE)"); + assertThat(generated.asSQLString()).contains("GENERATED ALWAYS AS (CAST(ts AS DATE))"); + RowType reassigned = + (RowType) ReassignFieldId.reassign(schema.rowType(), new AtomicInteger(20)); + assertThat(reassigned.getField("dt").generationExpression()).isEqualTo("CAST(ts AS DATE)"); + assertThat(generated).isNotEqualTo(generated.newGenerationExpression(null)); + } + + @Test + void testLegacySchemaHasNoGenerationExpression() { + TableSchema schema = + new TableSchema( + 0, + Collections.singletonList(new DataField(0, "id", DataTypes.INT())), + 0, + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyMap(), + null); + assertThat(schema.toString()).doesNotContain("generationExpression"); + assertThat(TableSchema.fromJson(schema.toString()).fields().get(0).generationExpression()) + .isNull(); + } + + @Test + void testExplicitNullGenerationExpression() { + TableSchema schema = + TableSchema.fromJson( + "{\"version\":3,\"id\":0,\"fields\":[{\"id\":0,\"name\":\"id\",\"type\":\"INT\"," + + "\"generationExpression\":null}],\"highestFieldId\":0,\"partitionKeys\":[]," + + "\"primaryKeys\":[],\"options\":{},\"timeMillis\":0}"); + assertThat(schema.fields().get(0).generationExpression()).isNull(); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumn.java b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumn.java new file mode 100644 index 000000000000..f376d51867df --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumn.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.utils; + +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** A generation expression compiled once for a physical top-level column. */ +public final class GeneratedColumn { + + private final int targetIndex; + private final GeneratedExpression expression; + + private GeneratedColumn(int targetIndex, GeneratedExpression expression) { + this.targetIndex = targetIndex; + this.expression = expression; + } + + public static GeneratedColumn create(RowType rowType, DataField field) { + GeneratedExpression expression = + GeneratedExpression.parse( + field.generationExpression(), rowType, Function.identity()); + checkArgument( + field.type().copy(true).equalsIgnoreFieldId(expression.outputType().copy(true)), + "Generation expression type %s does not match column '%s' type %s.", + expression.outputType(), + field.name(), + field.type()); + return new GeneratedColumn(rowType.getFieldIndex(field.name()), expression); + } + + /** Resolves SQL identifiers with an engine's case-sensitivity rules. */ + public static String normalizeExpression( + String expression, RowType rowType, Function resolver) { + return GeneratedExpression.parse(expression, rowType, resolver).normalized(); + } + + public int targetIndex() { + return targetIndex; + } + + public List sourceNames() { + return expression.references().stream().map(FieldRef::name).collect(Collectors.toList()); + } + + @Nullable + public Object evaluate(InternalRow row) { + return expression.evaluate(row); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumnRow.java b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumnRow.java new file mode 100644 index 000000000000..37ca1fe78928 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedColumnRow.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.utils; + +import org.apache.paimon.casting.DefaultValueRow; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import java.util.ArrayList; +import java.util.List; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Maps input fields to a full row and generates only omitted physical columns. */ +public final class GeneratedColumnRow { + + private final RowType rowType; + private final int inputArity; + private final int[] mapping; + private final InternalRow.FieldGetter[] getters; + private final List columns = new ArrayList<>(); + private final DefaultValueRow defaults; + + public GeneratedColumnRow(RowType rowType, RowType inputType) { + this.rowType = rowType; + this.inputArity = inputType.getFieldCount(); + this.mapping = new int[rowType.getFieldCount()]; + this.getters = new InternalRow.FieldGetter[rowType.getFieldCount()]; + List inputFields = new ArrayList<>(); + for (DataField input : inputType.getFields()) { + checkArgument( + rowType.containsField(input.name()), + "Unknown input column '%s'.", + input.name()); + DataField target = rowType.getField(input.name()); + checkArgument( + target.type().copy(true).equals(input.type().copy(true)), + "Input type for column '%s' does not match the table.", + input.name()); + inputFields.add(target); + } + this.defaults = DefaultValueRow.create(new RowType(inputFields)); + for (int i = 0; i < mapping.length; i++) { + DataField field = rowType.getFields().get(i); + mapping[i] = inputType.getFieldIndex(field.name()); + checkArgument( + mapping[i] >= 0 || field.generationExpression() != null, + "Missing input column '%s'. Only generated columns may be omitted.", + field.name()); + if (mapping[i] >= 0) { + getters[i] = + InternalRowUtils.createNullCheckingFieldGetter(field.type(), mapping[i]); + } + if (mapping[i] < 0) { + columns.add(GeneratedColumn.create(rowType, field)); + } + } + } + + public InternalRow replaceRow(InternalRow input) { + checkArgument( + input.getFieldCount() == inputArity, + "Input row has %s fields, expected %s.", + input.getFieldCount(), + inputArity); + InternalRow source = defaults == null ? input : defaults.replaceRow(input); + GenericRow result = new GenericRow(input.getRowKind(), rowType.getFieldCount()); + for (int i = 0; i < mapping.length; i++) { + if (mapping[i] >= 0) { + result.setField(i, getters[i].getFieldOrNull(source)); + } + } + for (GeneratedColumn column : columns) { + result.setField(column.targetIndex(), column.evaluate(result)); + } + return result; + } + + public static boolean hasGeneratedColumns(RowType rowType) { + return rowType.getFields().stream().anyMatch(f -> f.generationExpression() != null); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedExpression.java b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedExpression.java new file mode 100644 index 000000000000..fbc702631145 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/GeneratedExpression.java @@ -0,0 +1,415 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.utils; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.predicate.BitLengthTransform; +import org.apache.paimon.predicate.CastTransform; +import org.apache.paimon.predicate.ConcatTransform; +import org.apache.paimon.predicate.ConcatWsTransform; +import org.apache.paimon.predicate.DateAddTransform; +import org.apache.paimon.predicate.DateDiffTransform; +import org.apache.paimon.predicate.DateTruncTransform; +import org.apache.paimon.predicate.DayOfWeekTransform; +import org.apache.paimon.predicate.DayOfYearTransform; +import org.apache.paimon.predicate.DayTransform; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.FieldTransform; +import org.apache.paimon.predicate.HourTransform; +import org.apache.paimon.predicate.LengthTransform; +import org.apache.paimon.predicate.LowerTransform; +import org.apache.paimon.predicate.MinuteTransform; +import org.apache.paimon.predicate.MonthTransform; +import org.apache.paimon.predicate.OverlayTransform; +import org.apache.paimon.predicate.PadTransform; +import org.apache.paimon.predicate.QuarterTransform; +import org.apache.paimon.predicate.SecondTransform; +import org.apache.paimon.predicate.SubstringTransform; +import org.apache.paimon.predicate.Transform; +import org.apache.paimon.predicate.TranslateTransform; +import org.apache.paimon.predicate.TrimTransform; +import org.apache.paimon.predicate.UpperTransform; +import org.apache.paimon.predicate.WeekTransform; +import org.apache.paimon.predicate.WeekdayTransform; +import org.apache.paimon.predicate.YearTransform; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeFamily; +import org.apache.paimon.types.DataTypeJsonParser; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.MultisetType; +import org.apache.paimon.types.RowType; + +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.function.Function; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Compiles the supported SQL scalar expressions into existing storage transforms. */ +final class GeneratedExpression { + + private final int fieldCount; + private final List steps; + private final List references; + private final String normalized; + + private GeneratedExpression(Parser parser) { + this.fieldCount = parser.rowType.getFieldCount(); + this.steps = parser.steps; + this.references = new ArrayList<>(parser.references.values()); + this.normalized = parser.normalized.toString(); + } + + static GeneratedExpression parse( + String expression, RowType rowType, Function resolver) { + Parser parser = new Parser(expression, rowType, resolver); + Object root = parser.value(); + checkArgument(root instanceof FieldRef, "A generation expression must reference a column."); + if (((FieldRef) root).index() < rowType.getFieldCount()) { + parser.emit(new FieldTransform((FieldRef) root)); + } + checkArgument(parser.token == null, "Unsupported generation expression: %s", expression); + return new GeneratedExpression(parser); + } + + DataType outputType() { + return steps.get(steps.size() - 1).outputType(); + } + + List references() { + return references; + } + + String normalized() { + return normalized; + } + + Object evaluate(InternalRow row) { + if (steps.size() == 1) { + return steps.get(0).transform(row); + } + // Nested calls use temporary field references, without changing the table schema. + GenericRow inputs = new GenericRow(fieldCount + steps.size()); + for (FieldRef ref : references) { + inputs.setField(ref.index(), InternalRowUtils.get(row, ref.index(), ref.type())); + } + for (int i = 0; i < steps.size(); i++) { + inputs.setField(fieldCount + i, steps.get(i).transform(inputs)); + } + return inputs.getField(fieldCount + steps.size() - 1); + } + + private static final class Parser { + + private static final Pattern TOKEN = + Pattern.compile( + "\\s*(`(?:``|[^`])+`|'(?:''|[^'])*'|[+-]?\\d+|[a-zA-Z_][a-zA-Z_0-9]*|[(),])"); + + private final String expression; + private final RowType rowType; + private final Function resolver; + private final List steps = new ArrayList<>(); + private final Map references = new LinkedHashMap<>(); + private final StringBuilder normalized = new StringBuilder(); + private int position; + private String token; + private int tokenStart; + + private Parser(String expression, RowType rowType, Function resolver) { + this.expression = expression; + this.rowType = rowType; + this.resolver = resolver; + next(); + } + + private void next() { + Matcher matcher = TOKEN.matcher(expression); + matcher.region(position, expression.length()); + if (matcher.lookingAt()) { + tokenStart = matcher.start(1); + normalized.append(expression, position, tokenStart); + token = matcher.group(1); + position = matcher.end(); + } else { + checkArgument( + expression.substring(position).trim().isEmpty(), + "Unsupported generation expression near '%s'. Only top-level field references and supported scalar calls are allowed.", + expression.substring(position)); + normalized.append(expression.substring(position)); + position = expression.length(); + token = null; + } + } + + private String take() { + String result = token; + checkArgument(result != null, "Incomplete generation expression: %s", expression); + normalized.append(result); + next(); + return result; + } + + private void expect(String expected) { + checkArgument( + expected.equalsIgnoreCase(token), + "Expected '%s' in generation expression: %s", + expected, + expression); + take(); + } + + private Object value() { + checkArgument(token != null, "Incomplete generation expression: %s", expression); + String name = token; + if (name.startsWith("'")) { + take(); + return BinaryString.fromString( + name.substring(1, name.length() - 1).replace("''", "'")); + } + if (name.matches("[+-]?\\d+")) { + take(); + return Integer.valueOf(name); + } + if (name.equalsIgnoreCase("NULL")) { + take(); + return null; + } + // An identifier may name a function, a date literal, or a top-level column. + int start = tokenStart; + int end = position; + take(); + if ("(".equals(token) && !name.startsWith("`")) { + take(); + if (name.equalsIgnoreCase("CAST")) { + Object source = value(); + checkArgument( + source instanceof FieldRef, "CAST requires a column or scalar call."); + expect("AS"); + StringBuilder type = new StringBuilder(); + int depth = 0; + while (token != null && (depth > 0 || !token.equals(")"))) { + if (token.equals("(")) { + depth++; + } + if (token.equals(")")) { + depth--; + } + type.append(take()).append(' '); + } + expect(")"); + DataType target = + DataTypeJsonParser.parseAtomicTypeSQLString( + type.toString() + .trim() + .toUpperCase(Locale.ROOT) + .replace("TIMESTAMP_NTZ", "TIMESTAMP")); + checkArgument( + timeZoneIndependent(target), + "Generated expressions cannot use TIMESTAMP WITH LOCAL TIME ZONE."); + return emit(new CastTransform((FieldRef) source, target)); + } + List inputs = new ArrayList<>(); + if (!")".equals(token)) { + inputs.add(value()); + while (",".equals(token)) { + take(); + inputs.add(value()); + } + } + expect(")"); + return emit(function(name.toUpperCase(Locale.ROOT), inputs)); + } + if (name.equalsIgnoreCase("DATE") && token != null && token.startsWith("'")) { + String date = take(); + return Math.toIntExact( + LocalDate.parse(date.substring(1, date.length() - 1)).toEpochDay()); + } + String unquoted = + name.startsWith("`") + ? name.substring(1, name.length() - 1).replace("``", "`") + : name; + String resolved = resolver.apply(unquoted); + checkArgument( + rowType.containsField(resolved), + "Generation expression references unknown column '%s'.", + resolved); + DataField field = rowType.getField(resolved); + checkArgument( + field.generationExpression() == null, + "Generated columns cannot reference generated column '%s'.", + resolved); + checkArgument( + timeZoneIndependent(field.type()), + "Generated expressions cannot use TIMESTAMP WITH LOCAL TIME ZONE."); + // next() may have appended whitespace following this identifier. + if (!resolved.equals(unquoted)) { + int normalizedEnd = + normalized.length() + - (token == null ? expression.length() - end : tokenStart - end); + normalized.replace( + normalizedEnd - (end - start), + normalizedEnd, + EncodingUtils.escapeIdentifier(resolved)); + } + FieldRef ref = new FieldRef(rowType.getFieldIndex(resolved), resolved, field.type()); + references.put(resolved, ref); + return ref; + } + + private FieldRef emit(Transform transform) { + FieldRef result = + new FieldRef( + rowType.getFieldCount() + steps.size(), "", transform.outputType()); + steps.add(transform); + return result; + } + + private Transform function(String name, List inputs) { + switch (name) { + case "CONCAT": + return new ConcatTransform(inputs); + case "CONCAT_WS": + return new ConcatWsTransform(inputs); + case "UPPER": + return new UpperTransform(inputs); + case "LOWER": + return new LowerTransform(inputs); + case "SUBSTRING": + case "SUBSTR": + Transform substring = new SubstringTransform(inputs); + Object source = inputs.get(0); + checkArgument( + !(source instanceof FieldRef) + || ((FieldRef) source) + .type() + .is(DataTypeFamily.CHARACTER_STRING), + "SUBSTRING source must be a string."); + for (int i = 1; i < inputs.size(); i++) { + Object position = inputs.get(i); + checkArgument( + position == null + || position instanceof Integer + || (position instanceof FieldRef + && ((FieldRef) position) + .type() + .is(DataTypeFamily.INTEGER_NUMERIC)), + "SUBSTRING positions must be integers."); + } + return substring; + case "LENGTH": + case "CHAR_LENGTH": + return new LengthTransform(inputs); + case "BIT_LENGTH": + return new BitLengthTransform(inputs); + case "TRANSLATE": + return new TranslateTransform(inputs); + case "OVERLAY": + return new OverlayTransform(inputs); + case "LPAD": + return new PadTransform(inputs, PadTransform.Direction.LEFT); + case "RPAD": + return new PadTransform(inputs, PadTransform.Direction.RIGHT); + case "TRIM": + return new TrimTransform(trimInputs(inputs), TrimTransform.Flag.BOTH); + case "LTRIM": + return new TrimTransform(trimInputs(inputs), TrimTransform.Flag.LEADING); + case "RTRIM": + return new TrimTransform(trimInputs(inputs), TrimTransform.Flag.TRAILING); + case "DATE_ADD": + return new DateAddTransform(inputs); + case "DATEDIFF": + case "DATE_DIFF": + return new DateDiffTransform(inputs); + case "TRUNC": + return new DateTruncTransform(inputs); + default: + } + checkArgument( + inputs.size() == 1 && inputs.get(0) instanceof FieldRef, + "Unsupported generation function or arguments: %s", + name); + FieldRef field = (FieldRef) inputs.get(0); + switch (name) { + case "YEAR": + return new YearTransform(field); + case "MONTH": + return new MonthTransform(field); + case "DAY": + case "DAYOFMONTH": + return new DayTransform(field); + case "HOUR": + return new HourTransform(field); + case "MINUTE": + return new MinuteTransform(field); + case "SECOND": + return new SecondTransform(field); + case "QUARTER": + return new QuarterTransform(field); + case "DAYOFWEEK": + return new DayOfWeekTransform(field); + case "DAYOFYEAR": + return new DayOfYearTransform(field); + case "WEEKDAY": + return new WeekdayTransform(field); + case "WEEKOFYEAR": + return new WeekTransform(field); + default: + throw new IllegalArgumentException("Unsupported generation function: " + name); + } + } + + private List trimInputs(List inputs) { + // SQL's two-argument form is TRIM(characters, string); Transform takes the source + // first. + return inputs.size() == 2 + ? java.util.Arrays.asList(inputs.get(1), inputs.get(0)) + : inputs; + } + + private boolean timeZoneIndependent(DataType type) { + switch (type.getTypeRoot()) { + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return false; + case ROW: + return ((RowType) type) + .getFields().stream() + .allMatch(field -> timeZoneIndependent(field.type())); + case ARRAY: + return timeZoneIndependent(((ArrayType) type).getElementType()); + case MAP: + return timeZoneIndependent(((MapType) type).getKeyType()) + && timeZoneIndependent(((MapType) type).getValueType()); + case MULTISET: + return timeZoneIndependent(((MultisetType) type).getElementType()); + default: + return true; + } + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/GeneratedColumnRowTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/GeneratedColumnRowTest.java new file mode 100644 index 000000000000..a1dfb2076ffd --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/utils/GeneratedColumnRowTest.java @@ -0,0 +1,363 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.utils; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests omitted-column generation and preservation of supplied values before partition routing. */ +class GeneratedColumnRowTest { + + @Test + void testRejectTimeZoneDependentCastsInsideTopLevelComplexFields() { + for (org.apache.paimon.types.DataType sourceType : + Arrays.asList( + DataTypes.ROW(new DataField(0, "ts", DataTypes.TIMESTAMP_LTZ_MILLIS())), + DataTypes.ARRAY(DataTypes.TIMESTAMP_LTZ_MILLIS()), + DataTypes.MAP(DataTypes.STRING(), DataTypes.TIMESTAMP_LTZ_MILLIS()), + DataTypes.MULTISET(DataTypes.TIMESTAMP_LTZ_MILLIS()))) { + RowType type = + RowType.of( + new DataField(0, "source", sourceType), + new DataField(1, "text", DataTypes.STRING()) + .newGenerationExpression("CAST(source AS STRING)")); + assertThatThrownBy(() -> GeneratedColumn.create(type, type.getField("text"))) + .hasMessageContaining("TIMESTAMP WITH LOCAL TIME ZONE"); + } + } + + private final RowType rowType = + RowType.of( + new DataField(0, "ts", DataTypes.TIMESTAMP(9)), + new DataField(1, "dt", DataTypes.DATE()) + .newGenerationExpression("CAST(ts AS DATE)"), + new DataField(2, "id", DataTypes.INT())); + + @Test + void testOmittedPartitionAndReorderedInput() { + RowType inputType = RowType.of(rowType.getField("id"), rowType.getField("ts")); + GeneratedColumnRow generator = new GeneratedColumnRow(rowType, inputType); + InternalRow row = + generator.replaceRow( + GenericRow.of( + 1, + Timestamp.fromLocalDateTime( + LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999999)))); + assertThat(row.getInt(1)).isEqualTo(-1); + assertThat(row.getInt(2)).isEqualTo(1); + assertThat(generator.replaceRow(GenericRow.of(2, null)).isNullAt(1)).isTrue(); + } + + @Test + void testExplicitValuesAndNulls() { + GeneratedColumnRow generator = new GeneratedColumnRow(rowType, rowType); + Timestamp timestamp = Timestamp.fromEpochMillis(0); + assertThat(generator.replaceRow(GenericRow.of(timestamp, 0, 1)).getInt(1)).isZero(); + assertThat(generator.replaceRow(GenericRow.of(timestamp, 1, 1)).getInt(1)).isEqualTo(1); + assertThat(generator.replaceRow(GenericRow.of(timestamp, null, 1)).isNullAt(1)).isTrue(); + assertThat(generator.replaceRow(GenericRow.of(null, 0, 1)).getInt(1)).isZero(); + assertThat(generator.replaceRow(GenericRow.of(null, null, 1)).isNullAt(1)).isTrue(); + } + + @Test + void testGenerateOnlyOmittedColumnsWithoutEvaluatingSuppliedExpressions() { + RowType type = + RowType.of( + new DataField(0, "text", DataTypes.STRING()), + new DataField(1, "number", DataTypes.INT()) + .newGenerationExpression("CAST(text AS INT)"), + new DataField(2, "label", DataTypes.STRING()) + .newGenerationExpression("UPPER(text)")); + InternalRow row = + new GeneratedColumnRow(type, type.project("number", "text")) + .replaceRow(GenericRow.of(42, BinaryString.fromString("invalid"))); + assertThat(row.getInt(1)).isEqualTo(42); + assertThat(row.getString(2).toString()).isEqualTo("INVALID"); + assertThat( + new GeneratedColumnRow(type, type) + .replaceRow( + GenericRow.of( + BinaryString.fromString("invalid"), null, null)) + .isNullAt(1)) + .isTrue(); + assertThatThrownBy( + () -> + new GeneratedColumnRow(type, type.project("text")) + .replaceRow( + GenericRow.of(BinaryString.fromString("invalid")))) + .isInstanceOf(NumberFormatException.class); + } + + @Test + void testComplexIdentityAndSuppliedValuesWithDifferentFieldIds() { + RowType sourceType = + RowType.of(new DataField(10, "values", DataTypes.ARRAY(DataTypes.INT()))); + RowType targetType = + RowType.of(new DataField(20, "values", DataTypes.ARRAY(DataTypes.INT()))); + RowType type = + RowType.of( + new DataField(0, "source", sourceType), + new DataField(1, "copy", targetType).newGenerationExpression("source")); + GenericRow source = GenericRow.of(new GenericArray(new int[] {1, 2})); + InternalRow generated = + new GeneratedColumnRow(type, type.project("source")) + .replaceRow(GenericRow.of(source)); + assertThat(generated.getRow(1, 1).getArray(0).getInt(1)).isEqualTo(2); + GenericRow supplied = GenericRow.of(new GenericArray(new int[] {7})); + InternalRow explicit = + new GeneratedColumnRow(type, type).replaceRow(GenericRow.of(source, supplied)); + assertThat(explicit.getRow(1, 1).getArray(0).getInt(0)).isEqualTo(7); + assertThat(explicit.getRow(1, 1).getArray(0).size()).isEqualTo(1); + } + + @Test + void testPreservesDeletePartitionAndRowKind() { + GenericRow delete = GenericRow.of(null, 42, 1); + delete.setRowKind(RowKind.DELETE); + InternalRow result = new GeneratedColumnRow(rowType, rowType).replaceRow(delete); + assertThat(result.getRowKind()).isEqualTo(RowKind.DELETE); + assertThat(result.getInt(1)).isEqualTo(42); + } + + @Test + void testOnlyGeneratedFieldsMayBeOmitted() { + assertThatThrownBy( + () -> new GeneratedColumnRow(rowType, RowType.of(rowType.getField("ts")))) + .hasMessageContaining("Missing input column 'id'"); + } + + @Test + void testTransformsAndNestedCallsOnTopLevelFields() { + RowType type = + RowType.of( + new DataField(0, "name", DataTypes.STRING()), + new DataField(1, "ts", DataTypes.TIMESTAMP(9)), + new DataField(2, "days", DataTypes.INT()), + new DataField(3, "label", DataTypes.STRING()) + .newGenerationExpression( + "CONCAT(UPPER(name), '-', SUBSTRING(name, 2, 2))"), + new DataField(4, "year", DataTypes.INT()) + .newGenerationExpression("YEAR(ts)"), + new DataField(5, "date", DataTypes.DATE()) + .newGenerationExpression("DATE_ADD(CAST(ts AS DATE), days)"), + new DataField(6, "wide", DataTypes.BIGINT()) + .newGenerationExpression("CAST(days AS BIGINT)")); + GeneratedColumnRow generator = + new GeneratedColumnRow(type, type.project("days", "ts", "name")); + InternalRow result = + generator.replaceRow( + GenericRow.of( + 2, Timestamp.fromEpochMillis(-1), BinaryString.fromString("abc"))); + assertThat(result.getString(3).toString()).isEqualTo("ABC-bc"); + assertThat(result.getInt(4)).isEqualTo(1969); + assertThat(result.getInt(5)).isEqualTo(1); + assertThat(result.getLong(6)).isEqualTo(2L); + assertThat(new GeneratedColumnRow(type, type).replaceRow(result).getInt(5)).isEqualTo(1); + GenericRow wrong = + GenericRow.of( + BinaryString.fromString("abc"), + Timestamp.fromEpochMillis(-1), + 2, + BinaryString.fromString("wrong"), + 1969, + 1, + 2L); + assertThat(new GeneratedColumnRow(type, type).replaceRow(wrong).getString(3).toString()) + .isEqualTo("wrong"); + InternalRow nulls = generator.replaceRow(GenericRow.of(null, null, null)); + assertThat(nulls.isNullAt(3)).isTrue(); + assertThat(nulls.isNullAt(4)).isTrue(); + assertThat(nulls.isNullAt(5)).isTrue(); + assertThat(nulls.isNullAt(6)).isTrue(); + } + + @Test + void testUnsupportedExpressionsAndTypes() { + for (String expression : + Arrays.asList( + "current_date", + "CAST(id AS DATE)", + "CAST(missing AS DATE)", + "RAND()", + "CAST(ts AS DATE) trailing", + "ts.value", + "UPPER(ts)", + "YEAR(ts, id)", + "CONCAT(id, 'x')", + "YEAR(dt)", + "SUBSTRING(ts, 1)", + "SUBSTRING('x', 'bad')")) { + RowType invalid = + rowType.copy( + Arrays.asList( + rowType.getField("ts"), + rowType.getField("dt").newGenerationExpression(expression), + rowType.getField("id"))); + assertThatThrownBy(() -> GeneratedColumn.create(invalid, invalid.getField("dt"))) + .isInstanceOf(IllegalArgumentException.class); + } + RowType zoned = + rowType.copy( + Arrays.asList( + rowType.getField("ts").newType(DataTypes.TIMESTAMP_LTZ_MILLIS()), + rowType.getField("dt"), + rowType.getField("id"))); + assertThatThrownBy(() -> GeneratedColumn.create(zoned, zoned.getField("dt"))) + .hasMessageContaining("TIMESTAMP WITH LOCAL TIME ZONE"); + } + + @Test + void testNormalizeAllReferencesWithoutChangingLiterals() { + RowType type = RowType.of(new DataField(0, "name", DataTypes.STRING())); + String expression = "CONCAT(UPPER(`NAME`), 'NAME''s', NAME) "; + String normalized = + GeneratedColumn.normalizeExpression(expression, type, String::toLowerCase); + RowType generated = + RowType.of( + type.getField("name"), + new DataField(1, "label", DataTypes.STRING()) + .newGenerationExpression(normalized)); + InternalRow row = + new GeneratedColumnRow(generated, type) + .replaceRow(GenericRow.of(BinaryString.fromString("ab"))); + assertThat(row.getString(1).toString()).isEqualTo("ABNAME'sab"); + assertThat(GeneratedColumn.create(generated, generated.getField("label")).sourceNames()) + .containsExactly("name"); + } + + @Test + void testIdentityGeneratesOmittedBinaryValuesAndPreservesSuppliedValues() { + RowType type = + RowType.of( + new DataField(0, "bytes", DataTypes.BYTES()), + new DataField(1, "copy", DataTypes.BYTES()) + .newGenerationExpression("bytes")); + GeneratedColumnRow generator = new GeneratedColumnRow(type, type); + assertThat( + generator + .replaceRow(GenericRow.of(new byte[] {1, 2}, new byte[] {1, 2})) + .getBinary(1)) + .containsExactly(1, 2); + assertThat(generator.replaceRow(GenericRow.of(new byte[] {1}, new byte[] {2})).getBinary(1)) + .containsExactly(2); + assertThat( + new GeneratedColumnRow(type, type.project("bytes")) + .replaceRow(GenericRow.of(new byte[] {1, 2})) + .getBinary(1)) + .containsExactly(1, 2); + } + + @Test + void testDateAndStringLiteralArguments() { + RowType type = + RowType.of( + new DataField(0, "days", DataTypes.INT()), + new DataField(1, "name", DataTypes.STRING()), + new DataField(2, "date", DataTypes.DATE()) + .newGenerationExpression("DATE_ADD(DATE '1970-01-01', days)"), + new DataField(3, "label", DataTypes.STRING()) + .newGenerationExpression("LPAD(SUBSTRING(name, 1, 2), 4, '_')")); + InternalRow row = + new GeneratedColumnRow(type, type.project("days", "name")) + .replaceRow(GenericRow.of(2, BinaryString.fromString("abc"))); + assertThat(row.getInt(2)).isEqualTo(2); + assertThat(row.getString(3).toString()).isEqualTo("__ab"); + } + + @Test + void testRejectGeneratedDependenciesWithMatchingTypes() { + RowType type = + RowType.of( + rowType.getField("ts"), + rowType.getField("dt"), + new DataField(3, "year", DataTypes.INT()) + .newGenerationExpression("YEAR(dt)")); + assertThatThrownBy(() -> GeneratedColumn.create(type, type.getField("year"))) + .hasMessageContaining("cannot reference generated column 'dt'"); + RowType self = + RowType.of( + new DataField(0, "copy", DataTypes.INT()).newGenerationExpression("copy")); + assertThatThrownBy(() -> GeneratedColumn.create(self, self.getField("copy"))) + .hasMessageContaining("cannot reference generated column 'copy'"); + } + + @Test + void testSqlTwoArgumentTrimOrder() { + RowType type = + RowType.of( + new DataField(0, "name", DataTypes.STRING()), + new DataField(1, "trimmed", DataTypes.STRING()) + .newGenerationExpression("TRIM('x', name)"), + new DataField(2, "left_trimmed", DataTypes.STRING()) + .newGenerationExpression("LTRIM('x', name)"), + new DataField(3, "right_trimmed", DataTypes.STRING()) + .newGenerationExpression("RTRIM('x', name)")); + GeneratedColumnRow generator = new GeneratedColumnRow(type, type.project("name")); + InternalRow row = generator.replaceRow(GenericRow.of(BinaryString.fromString("xabcxx"))); + assertThat(row.getString(1).toString()).isEqualTo("abc"); + assertThat(row.getString(2).toString()).isEqualTo("abcxx"); + assertThat(row.getString(3).toString()).isEqualTo("xabc"); + assertThat(new GeneratedColumnRow(type, type).replaceRow(row).getString(1).toString()) + .isEqualTo("abc"); + InternalRow nulls = generator.replaceRow(GenericRow.of((Object) null)); + assertThat(nulls.isNullAt(1)).isTrue(); + assertThat(nulls.isNullAt(2)).isTrue(); + assertThat(nulls.isNullAt(3)).isTrue(); + } + + @Test + void testTimestampNtzCastAlias() { + RowType type = + RowType.of( + new DataField(0, "event_date", DataTypes.DATE()), + new DataField(1, "event_time", DataTypes.TIMESTAMP(6)) + .newGenerationExpression("cast(event_date as timestamp_ntz)")); + GeneratedColumnRow generator = new GeneratedColumnRow(type, type.project("event_date")); + assertThat(generator.replaceRow(GenericRow.of(0)).getTimestamp(1, 6)) + .isEqualTo(Timestamp.fromEpochMillis(0)); + assertThat(generator.replaceRow(GenericRow.of(-1)).getTimestamp(1, 6)) + .isEqualTo(Timestamp.fromEpochMillis(-86400000L)); + assertThat(generator.replaceRow(GenericRow.of((Object) null)).isNullAt(1)).isTrue(); + assertThat( + new GeneratedColumnRow(type, type) + .replaceRow(GenericRow.of(0, Timestamp.fromEpochMillis(0))) + .getTimestamp(1, 6)) + .isEqualTo(Timestamp.fromEpochMillis(0)); + assertThat( + new GeneratedColumnRow(type, type) + .replaceRow(GenericRow.of(0, Timestamp.fromEpochMillis(1))) + .getTimestamp(1, 6)) + .isEqualTo(Timestamp.fromEpochMillis(1)); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index 98acf9fd8de3..8bae14c644a0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java @@ -173,6 +173,9 @@ public static void validateCreateTable(Schema schema, boolean dataTokenEnabled) SchemaValidation.validateQueryAuthTableType( tableType, options.get(CoreOptions.QUERY_AUTH_ENABLED)); for (DataField field : schema.fields()) { + checkArgument( + field.generationExpression() == null || tableType == TableType.TABLE, + "Generated columns are only supported for native Paimon tables."); validateDefaultValue(field.type(), field.defaultValue()); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java index 73b2c9328773..b1256badc476 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/FileSystemSchemaManager.java @@ -271,6 +271,10 @@ public boolean mergeSchema( @VisibleForTesting public boolean commit(TableSchema newSchema) throws Exception { + Optional previous = latest(); + if (previous.isPresent()) { + SchemaValidation.validateGeneratedColumnEvolution(previous.get(), newSchema); + } SchemaValidation.validateTableSchema(newSchema); SchemaValidation.validateHistoricalIcebergTypes( this::listAll, new CoreOptions(newSchema.options())); diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java index 52adcf42b325..712ec70dd0bb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java @@ -282,7 +282,8 @@ protected void updateLastColumn( rename.newName(), field.type(), field.description(), - field.defaultValue()); + field.defaultValue(), + field.generationExpression()); newFields.set(i, newField); return; } @@ -360,7 +361,8 @@ protected void updateLastColumn( field.name(), newFieldType, field.description(), - field.defaultValue()); + field.defaultValue(), + field.generationExpression()); }, lazyIdentifier); } else if (change instanceof UpdateColumnNullability) { @@ -391,7 +393,8 @@ protected void updateLastColumn( depth, update.fieldNames().length), field.description(), - field.defaultValue()); + field.defaultValue(), + field.generationExpression()); }, lazyIdentifier); } else if (change instanceof UpdateColumnComment) { @@ -410,7 +413,8 @@ protected void updateLastColumn( field.name(), field.type(), update.newDescription(), - field.defaultValue()), + field.defaultValue(), + field.generationExpression()), lazyIdentifier); } else if (change instanceof UpdateColumnPosition) { UpdateColumnPosition update = (UpdateColumnPosition) change; @@ -428,7 +432,8 @@ protected void updateLastColumn( field.name(), field.type(), field.description(), - update.newDefaultValue()); + update.newDefaultValue(), + field.generationExpression()); }, lazyIdentifier); } else if (change instanceof SchemaChange.DropPrimaryKey) { @@ -462,6 +467,7 @@ protected void updateLastColumn( newSchema.primaryKeys(), newSchema.options(), newSchema.comment()); + SchemaValidation.validateGeneratedColumnEvolution(oldTableSchema, newTableSchema); SchemaValidation.validateTableSchema(newTableSchema); return newTableSchema; } @@ -888,7 +894,8 @@ private void updateIntermediateColumn( field.name(), wrapNewRowType(field.type(), nestedFields), field.description(), - field.defaultValue())); + field.defaultValue(), + field.generationExpression())); return; } diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java index 88c5eb2f5c4f..c53b10ec116d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java @@ -154,7 +154,8 @@ public static DataType merge( baseField.name(), updatedDataType, baseField.description(), - baseField.defaultValue())); + baseField.defaultValue(), + baseField.generationExpression())); } else { updatedFields.add(baseField); } @@ -327,7 +328,8 @@ private static DataField assignIdForNewField(DataField field, AtomicInteger high field.name(), dataType, field.description(), - field.defaultValue()); + field.defaultValue(), + field.generationExpression()); } /** diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index e279d6479587..45658236f481 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -58,6 +58,7 @@ import org.apache.paimon.types.TimestampType; import org.apache.paimon.types.VariantType; import org.apache.paimon.types.VectorType; +import org.apache.paimon.utils.GeneratedColumn; import org.apache.paimon.utils.Preconditions; import org.apache.paimon.utils.SetUtils; import org.apache.paimon.utils.StringUtils; @@ -172,6 +173,8 @@ public static void validateTableSchema(TableSchema schema) { public static void validateTableSchema(TableSchema schema, Set dynamicOptionKeys) { CoreOptions options = new CoreOptions(schema.options()); + validateGeneratedColumns(schema, options); + validateOnlyContainPrimitiveType(schema.fields(), schema.primaryKeys(), "primary key"); validateOnlyContainPrimitiveType(schema.fields(), schema.partitionKeys(), "partition"); // reject here rather than only at create time, so ALTER cannot turn the option on for a @@ -1619,6 +1622,80 @@ private static void validateMergeFunctionFactory(TableSchema schema) { } } + /** Rejects changes which invalidate stored generated values or their dependencies. */ + public static void validateGeneratedColumnEvolution(TableSchema previous, TableSchema next) { + for (DataField field : previous.fields()) { + if (field.generationExpression() == null) { + continue; + } + DataField target = next.idToFieldMap().get(field.id()); + checkArgument( + target != null + && field.name().equals(target.name()) + && field.type().equals(target.type()) + && field.generationExpression().equals(target.generationExpression()), + "Cannot change generated column '%s'.", + field.name()); + GeneratedColumn expression = GeneratedColumn.create(previous.logicalRowType(), field); + for (String sourceName : expression.sourceNames()) { + DataField source = previous.logicalRowType().getField(sourceName); + DataField newSource = next.idToFieldMap().get(source.id()); + checkArgument( + newSource != null + && source.name().equals(newSource.name()) + && source.type().equals(newSource.type()), + "Cannot rename, drop or change the type of column '%s' referenced by generated column '%s'.", + source.name(), + field.name()); + } + } + for (DataField field : next.fields()) { + if (field.generationExpression() != null) { + DataField old = previous.idToFieldMap().get(field.id()); + checkArgument( + old != null && old.generationExpression() != null, + "Generated columns can only be defined when creating a table."); + } + } + } + + private static void validateGeneratedColumns(TableSchema schema, CoreOptions options) { + RowType rowType = schema.logicalRowType(); + for (DataField field : schema.fields()) { + validateNoNestedGeneratedColumns(field.type()); + if (field.generationExpression() == null) { + continue; + } + checkArgument( + field.defaultValue() == null, + "Generated column '%s' cannot have a default value.", + field.name()); + checkArgument( + !options.rowkindField().filter(field.name()::equals).isPresent(), + "Generated column '%s' cannot be used as rowkind.field.", + field.name()); + GeneratedColumn.create(rowType, field); + } + } + + private static void validateNoNestedGeneratedColumns(DataType type) { + if (type instanceof RowType) { + for (DataField nested : ((RowType) type).getFields()) { + checkArgument( + nested.generationExpression() == null, + "Generated columns must be top-level columns."); + validateNoNestedGeneratedColumns(nested.type()); + } + } else if (type instanceof ArrayType) { + validateNoNestedGeneratedColumns(((ArrayType) type).getElementType()); + } else if (type instanceof MultisetType) { + validateNoNestedGeneratedColumns(((MultisetType) type).getElementType()); + } else if (type instanceof MapType) { + validateNoNestedGeneratedColumns(((MapType) type).getKeyType()); + validateNoNestedGeneratedColumns(((MapType) type).getValueType()); + } + } + private static void validateRowTracking(TableSchema schema, CoreOptions options) { boolean rowTrackingEnabled = options.rowTrackingEnabled(); if (rowTrackingEnabled) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableWrite.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableWrite.java index 538b50b3cd74..f0cc2b0e4f9a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableWrite.java @@ -43,8 +43,8 @@ public interface BatchTableWrite extends TableWrite { List prepareCommit() throws Exception; /** - * Specified the writing rowType, currently only work for table without primary key and row - * tracking enabled. + * Specifies the write row type. Tables with generated columns support omitting those columns. + * Other projections require a table without a primary key and with row tracking enabled. */ BatchTableWrite withWriteType(RowType writeType); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWrite.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWrite.java index 50564d19cfe1..15420308c59a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWrite.java @@ -41,7 +41,13 @@ public interface TableWrite extends AutoCloseable { /** With {@link IOManager}, this is needed if 'write-buffer-spillable' is set to true. */ TableWrite withIOManager(IOManager ioManager); - /** Specified the write rowType. */ + /** + * Specifies the write row type. If all ordinary columns are supplied, omitted generated columns + * are materialized before routing and writing. Otherwise, the underlying writer's + * partial-column write behavior applies, without generating omitted columns. Supplied generated + * values, including nulls, are preserved without evaluating or checking the generation + * expression. + */ TableWrite withWriteType(RowType writeType); /** With a shared {@link MemoryPoolFactory} for the current table write. */ diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java index 7f9add7879a2..af89ed1f20ba 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java @@ -33,14 +33,18 @@ import org.apache.paimon.operation.FileStoreWrite; import org.apache.paimon.operation.FileStoreWrite.State; import org.apache.paimon.operation.WriteRestore; +import org.apache.paimon.table.SpecialFields; import org.apache.paimon.types.DataField; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.GeneratedColumnRow; import org.apache.paimon.utils.Restorable; import org.apache.paimon.utils.RowKindFilter; import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Arrays; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -69,6 +73,7 @@ public class TableWriteImpl implements InnerTableWrite, Restorable deleteNotNullFieldNames; public TableWriteImpl( @@ -107,6 +112,10 @@ public TableWriteImpl( deleteNotNullFieldNames == null ? null : new HashSet<>(deleteNotNullFieldNames); updateNotNullFieldIndexes(); this.defaultValueRow = DefaultValueRow.create(rowType); + this.generatedColumnRow = + GeneratedColumnRow.hasGeneratedColumns(rowType) + ? new GeneratedColumnRow(rowType, rowType) + : null; } public FileStoreWrite fileStoreWrite() { @@ -133,10 +142,40 @@ public TableWriteImpl withIOManager(IOManager ioManager) { @Override public TableWriteImpl withWriteType(RowType writeType) { + if (GeneratedColumnRow.hasGeneratedColumns(rowType) + && rowType.getFields().stream() + .allMatch( + field -> + field.generationExpression() != null + || writeType.containsField(field.name()))) { + List physicalFields = new ArrayList<>(rowType.getFields()); + for (DataField special : + Arrays.asList(SpecialFields.ROW_ID, SpecialFields.SEQUENCE_NUMBER)) { + if (writeType.containsField(special.name())) { + DataField input = writeType.getField(special.name()); + checkState( + input.id() == special.id() + && input.type().equalsIgnoreNullable(special.type()), + "Invalid row-tracking field '%s'.", + special.name()); + physicalFields.add(special.newType(input.type())); + } + } + RowType physicalType = new RowType(rowType.isNullable(), physicalFields); + GeneratedColumnRow generator = new GeneratedColumnRow(physicalType, writeType); + if (!physicalType.equals(this.writeType)) { + write.withWriteType(physicalType); + } + this.writeType = physicalType; + updateNotNullFieldIndexes(); + this.generatedColumnRow = generator; + return this; + } write.withWriteType(writeType); this.writeType = writeType; updateNotNullFieldIndexes(); this.defaultValueRow = DefaultValueRow.create(writeType, rowType); + this.generatedColumnRow = null; return this; } @@ -183,13 +222,13 @@ public TableWriteImpl withCompactExecutor(ExecutorService compactExecutor) { @Override public BinaryRow getPartition(InternalRow row) { - keyAndBucketExtractor.setRecord(row); + keyAndBucketExtractor.setRecord(generatedColumnRow == null ? row : prepareRow(row)); return keyAndBucketExtractor.partition(); } @Override public int getBucket(InternalRow row) { - keyAndBucketExtractor.setRecord(row); + keyAndBucketExtractor.setRecord(generatedColumnRow == null ? row : prepareRow(row)); return keyAndBucketExtractor.bucket(); } @@ -206,7 +245,7 @@ public void write(InternalRow row, int bucket) throws Exception { @Override public void writeBundle(BinaryRow partition, int bucket, BundleRecords bundle) throws Exception { - if (write instanceof BundleFileStoreWriter) { + if (write instanceof BundleFileStoreWriter && generatedColumnRow == null) { ((BundleFileStoreWriter) write).writeBundle(partition, bucket, bundle); } else { for (InternalRow row : bundle) { @@ -237,9 +276,9 @@ public SinkRecord writeAndReturn(InternalRow row, int bucket, int totalBuckets) @Nullable private SinkRecord writeAndReturn(InternalRow row, int bucket, @Nullable Integer totalBuckets) throws Exception { - InternalRow wrappedRow = wrapDefaultValue(row); + InternalRow wrappedRow = prepareRow(row); RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, wrappedRow); - checkNullability(row, rowKind); + checkNullability(generatedColumnRow == null ? row : wrappedRow, rowKind); if (rowKindFilter != null && !rowKindFilter.test(rowKind)) { return null; } @@ -270,6 +309,12 @@ private InternalRow wrapDefaultValue(InternalRow row) { return defaultValueRow == null ? row : defaultValueRow.replaceRow(row); } + private InternalRow prepareRow(InternalRow row) { + return generatedColumnRow == null + ? wrapDefaultValue(row) + : generatedColumnRow.replaceRow(row); + } + private SinkRecord toSinkRecord(InternalRow row) { keyAndBucketExtractor.setRecord(row); return new SinkRecord( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/GeneratedPartitionColumnTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/GeneratedPartitionColumnTest.java new file mode 100644 index 000000000000..0877390c53e1 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/GeneratedPartitionColumnTest.java @@ -0,0 +1,471 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.sink; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Timestamp; +import org.apache.paimon.disk.IOManagerImpl; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.reader.RecordReaderIterator; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaValidation; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowKind; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Storage and schema contracts for generated columns. */ +class GeneratedPartitionColumnTest { + + @TempDir java.nio.file.Path tempDir; + + @ParameterizedTest + @ValueSource(strings = {"append", "primary-key", "data-evolution"}) + void testGenerateRouteCommitAndRead(String mode) throws Exception { + FileStoreTable table = createTable(mode); + RowType input = RowType.of(table.rowType().getField("ts"), table.rowType().getField("id")); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.withWriteType(input); + GenericRow first = GenericRow.of(Timestamp.fromEpochMillis(-1), 1); + assertThat(write.getPartition(first).getInt(0)).isEqualTo(-1); + write.write(first); + write.write(GenericRow.of(Timestamp.fromEpochMillis(0), 2)); + commit.commit(write.prepareCommit()); + } + List splits = table.newScan().plan().splits(); + List partitions = new ArrayList<>(); + for (Split split : splits) { + partitions.add(((DataSplit) split).partition().getInt(0)); + } + assertThat(partitions).containsExactlyInAnyOrder(-1, 0); + List rows = new ArrayList<>(); + try (RecordReaderIterator reader = + new RecordReaderIterator<>(table.newRead().createReader(table.newScan().plan()))) { + while (reader.hasNext()) { + InternalRow row = reader.next(); + rows.add(row.getInt(0) + ":" + row.getInt(1)); + } + } + assertThat(rows).containsExactlyInAnyOrder("1:-1", "2:0"); + // A reader and writer reopened from persistent schema retain the contract. + FileStoreTable reopened = + FileStoreTableFactory.create(LocalFileIO.create(), table.location()); + BatchWriteBuilder explicit = reopened.newBatchWriteBuilder(); + try (BatchTableWrite write = explicit.newWrite().withWriteType(reopened.rowType()); + BatchTableCommit commit = explicit.newCommit()) { + GenericRow supplied = GenericRow.of(3, 1, Timestamp.fromEpochMillis(0)); + assertThat(write.getPartition(supplied).getInt(0)).isEqualTo(1); + write.write(supplied); + assertThatThrownBy( + () -> write.write(GenericRow.of(3, null, Timestamp.fromEpochMillis(0)))) + .hasMessageContaining("Cannot write null to non-null column(dt)"); + commit.commit(write.prepareCommit()); + } + rows.clear(); + try (RecordReaderIterator reader = + new RecordReaderIterator<>( + reopened.newRead().createReader(reopened.newScan().plan()))) { + while (reader.hasNext()) { + InternalRow row = reader.next(); + rows.add(row.getInt(0) + ":" + row.getInt(1)); + } + } + assertThat(rows).containsExactlyInAnyOrder("1:-1", "2:0", "3:1"); + } + + @Test + void testSchemaEvolutionProtectsDependencies() throws Exception { + FileStoreTable table = createTable(false); + FileSystemSchemaManager manager = + new FileSystemSchemaManager(LocalFileIO.create(), table.location()); + TableSchema changed = + manager.commitChanges( + SchemaChange.updateColumnComment(new String[] {"dt"}, "generated date")); + assertThat(changed.logicalRowType().getField("dt").generationExpression()) + .isEqualTo("CAST(ts AS DATE)"); + assertThatThrownBy(() -> manager.commitChanges(SchemaChange.renameColumn("ts", "time"))) + .hasMessageContaining("referenced by generated column"); + assertThatThrownBy(() -> manager.commitChanges(SchemaChange.dropColumn("ts"))) + .hasMessageContaining("referenced by generated column"); + assertThatThrownBy( + () -> + manager.commitChanges( + SchemaChange.updateColumnType( + "ts", DataTypes.TIMESTAMP(6)))) + .hasMessageContaining("referenced by generated column"); + assertThatThrownBy( + () -> + manager.commitChanges( + SchemaChange.updateColumnDefaultValue( + new String[] {"dt"}, "'2026-10-30'"))) + .hasMessageContaining("cannot have a default value"); + } + + @ParameterizedTest + @ValueSource(strings = {"partial-update", "aggregation", "first-row"}) + void testGeneratedValuesFollowMergeEngine(String engine) throws Exception { + Schema.Builder schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("value", DataTypes.INT()) + .generatedColumn("copy", DataTypes.BIGINT(), "CAST(value AS BIGINT)") + .primaryKey("id") + .option("bucket", "1") + .option("merge-engine", engine); + if (engine.equals("aggregation")) { + schema.option("fields.value.aggregate-function", "sum") + .option("fields.copy.aggregate-function", "sum"); + } + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + new FileSystemSchemaManager(LocalFileIO.create(), path).createTable(schema.build()); + FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), path); + for (int step = 0; step < 3; step++) { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (IOManagerImpl ioManager = new IOManagerImpl(tempDir.toString()); + BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + write.withIOManager(ioManager); + if (step == 0) { + write.withWriteType(table.rowType().project("id", "value")); + write.write(GenericRow.of(1, 10)); + } else { + write.write( + step == 1 ? GenericRow.of(1, 20, null) : GenericRow.of(1, null, 99L)); + } + commit.commit(write.prepareCommit()); + } + } + try (RecordReaderIterator rows = + new RecordReaderIterator<>(table.newRead().createReader(table.newScan().plan()))) { + InternalRow row = rows.next(); + assertThat(row.getInt(1)) + .isEqualTo( + engine.equals("aggregation") + ? 30 + : engine.equals("first-row") ? 10 : 20); + assertThat(row.getLong(2)) + .isEqualTo( + engine.equals("aggregation") + ? 109L + : engine.equals("first-row") ? 10L : 99L); + assertThat(rows.hasNext()).isFalse(); + } + } + + @Test + void testDataEvolutionPartialWriteKeepsUnselectedGeneratedValues() throws Exception { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .generatedColumn("label", DataTypes.STRING(), "UPPER(name)") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .build(); + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + new FileSystemSchemaManager(LocalFileIO.create(), path).createTable(schema); + FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), path); + String[] expectedNames = {"old", "new", "new"}; + String[] expectedLabels = {"OLD", "OLD", "external"}; + for (int step = 0; step < 3; step++) { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + // Switching from a partial layout must restore omitted-column generation. + if (step == 0) { + write.withWriteType(table.rowType().project("label")); + } + write.withWriteType( + table.rowType() + .project( + step == 0 + ? new String[] {"id", "name"} + : step == 1 + ? new String[] {"name"} + : new String[] {"label"})); + write.write( + step == 0 + ? GenericRow.of(1, BinaryString.fromString("old")) + : GenericRow.of( + BinaryString.fromString(step == 1 ? "new" : "external"))); + List messages = write.prepareCommit(); + if (step > 0) { + for (CommitMessage message : messages) { + List files = + ((CommitMessageImpl) message).newFilesIncrement().newFiles(); + List originals = new ArrayList<>(files); + files.clear(); + for (DataFileMeta file : originals) { + files.add(file.assignFirstRowId(0)); + } + } + } + commit.commit(messages); + } + try (RecordReaderIterator rows = + new RecordReaderIterator<>( + table.newRead().createReader(table.newScan().plan()))) { + InternalRow row = rows.next(); + assertThat(row.getInt(0)).isEqualTo(1); + assertThat(row.getString(1).toString()).isEqualTo(expectedNames[step]); + assertThat(row.getString(2).toString()).isEqualTo(expectedLabels[step]); + assertThat(rows.hasNext()).isFalse(); + } + } + } + + @Test + void testRejectInvalidDefinitions() { + Schema plain = + Schema.newBuilder() + .column("ts", DataTypes.TIMESTAMP(3)) + .generatedColumn("dt", DataTypes.INT(), "CAST(ts AS DATE)") + .build(); + assertThatThrownBy(() -> validate(plain)).hasMessageContaining("does not match"); + Schema defaults = + Schema.newBuilder() + .column("ts", DataTypes.TIMESTAMP(3)) + .column("dt", DataTypes.DATE(), null, "'2026-10-30'", "CAST(ts AS DATE)") + .partitionKeys("dt") + .build(); + assertThatThrownBy(() -> validate(defaults)) + .hasMessageContaining("cannot have a default value"); + Schema nested = + Schema.newBuilder() + .column( + "nested", + DataTypes.ROW( + new DataField(0, "dt", DataTypes.DATE()) + .newGenerationExpression("CAST(ts AS DATE)"))) + .build(); + assertThatThrownBy(() -> validate(nested)).hasMessageContaining("top-level columns"); + } + + @ParameterizedTest + @ValueSource(strings = {"append", "primary-key", "data-evolution"}) + void testNonPartitionTransformsAndAllDependencies(String mode) throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("ts", DataTypes.TIMESTAMP(3)) + .column("name", DataTypes.STRING()) + .column("days", DataTypes.INT()) + .generatedColumn("dt", DataTypes.DATE(), "DATE_ADD(CAST(ts AS DATE), days)") + .generatedColumn( + "label", DataTypes.STRING(), "CONCAT(UPPER(name), '_', name)") + .generatedColumn("year", DataTypes.INT(), "YEAR(ts)") + .partitionKeys("dt"); + if (mode.equals("primary-key")) { + builder.primaryKey("id", "dt").option("bucket", "1"); + } else if (mode.equals("data-evolution")) { + builder.option("row-tracking.enabled", "true").option("data-evolution.enabled", "true"); + } + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + FileSystemSchemaManager manager = new FileSystemSchemaManager(LocalFileIO.create(), path); + manager.createTable(builder.build()); + FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), path); + BatchWriteBuilder writes = table.newBatchWriteBuilder(); + try (BatchTableWrite write = + writes.newWrite() + .withWriteType( + table.rowType().project("name", "days", "ts", "id")); + BatchTableCommit commit = writes.newCommit()) { + write.write( + GenericRow.of( + BinaryString.fromString("ab"), 2, Timestamp.fromEpochMillis(-1), 1)); + commit.commit(write.prepareCommit()); + } + try (RecordReaderIterator rows = + new RecordReaderIterator<>(table.newRead().createReader(table.newScan().plan()))) { + InternalRow row = rows.next(); + assertThat(row.getInt(4)).isEqualTo(1); + assertThat(row.getString(5).toString()).isEqualTo("AB_ab"); + assertThat(row.getInt(6)).isEqualTo(1969); + assertThat(rows.hasNext()).isFalse(); + } + assertThatThrownBy(() -> manager.commitChanges(SchemaChange.renameColumn("days", "offset"))) + .hasMessageContaining("referenced by generated column"); + } + + @Test + void testRoutingWithoutGeneratedColumnsPreservesRawValues() throws Exception { + List tables = new ArrayList<>(); + for (boolean defaults : new boolean[] {false, true}) { + Schema.Builder builder = + Schema.newBuilder() + .column("pt", DataTypes.INT(), null, defaults ? "7" : null) + .column("id", DataTypes.INT(), null, defaults ? "9" : null) + .partitionKeys("pt") + .primaryKey("pt", "id") + .option("bucket", "1031") + .option("bucket-key", "id"); + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + new FileSystemSchemaManager(LocalFileIO.create(), path).createTable(builder.build()); + tables.add(FileStoreTableFactory.create(LocalFileIO.create(), path)); + } + GenericRow raw = GenericRow.of(null, null); + try (BatchTableWrite plain = tables.get(0).newBatchWriteBuilder().newWrite(); + BatchTableWrite defaults = tables.get(1).newBatchWriteBuilder().newWrite()) { + assertThat(defaults.getPartition(raw).isNullAt(0)).isTrue(); + assertThat(defaults.getBucket(raw)).isEqualTo(plain.getBucket(raw)); + assertThat(defaults.getBucket(raw)) + .isNotEqualTo(defaults.getBucket(GenericRow.of(7, 9))); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testEffectiveRowKindWithSuppliedGeneratedValues(boolean reordered) throws Exception { + Schema schema = + schema(true) + .column("kind", DataTypes.STRING(), null, "'-D'") + .option("rowkind.field", "kind") + .build(); + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + new FileSystemSchemaManager(LocalFileIO.create(), path).createTable(schema); + FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), path); + BatchWriteBuilder inserts = table.newBatchWriteBuilder(); + try (BatchTableWrite write = inserts.newWrite(); + BatchTableCommit commit = inserts.newCommit()) { + for (int id : new int[] {1, 2}) { + write.write( + GenericRow.of( + id, + 0, + Timestamp.fromEpochMillis(0), + BinaryString.fromString("+I"))); + } + commit.commit(write.prepareCommit()); + } + + RowType inputType = + reordered ? table.rowType().project("kind", "ts", "id", "dt") : table.rowType(); + BatchWriteBuilder deletes = table.newBatchWriteBuilder(); + try (BatchTableWrite write = deletes.newWrite().withWriteType(inputType); + BatchTableCommit commit = deletes.newCommit()) { + GenericRow insert = + reordered + ? GenericRow.of(BinaryString.fromString("+I"), null, 3, 0) + : GenericRow.of(3, 0, null, BinaryString.fromString("+I")); + insert.setRowKind(RowKind.DELETE); + assertThat(write.getPartition(insert).getInt(0)).isZero(); + assertThat(write.getBucket(insert)).isZero(); + write.write(insert); + + for (int id : new int[] {1, 2, 3}) { + BinaryString kind = id == 1 ? null : BinaryString.fromString("-D"); + GenericRow delete = + reordered + ? GenericRow.of(kind, null, id, 0) + : GenericRow.of(id, 0, null, kind); + assertThat(delete.getRowKind()).isEqualTo(RowKind.INSERT); + assertThat(write.getPartition(delete).getInt(0)).isZero(); + assertThat(write.getBucket(delete)).isZero(); + write.write(delete); + } + commit.commit(write.prepareCommit()); + } + try (RecordReaderIterator rows = + new RecordReaderIterator<>(table.newRead().createReader(table.newScan().plan()))) { + assertThat(rows.hasNext()).isFalse(); + } + } + + @Test + void testRejectGeneratedRowKindFieldAtCreation() throws Exception { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("operation", DataTypes.STRING()) + .generatedColumn("kind", DataTypes.STRING(), "UPPER(operation)") + .primaryKey("id") + .option("bucket", "1") + .option("rowkind.field", "kind") + .build(); + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + FileSystemSchemaManager manager = new FileSystemSchemaManager(LocalFileIO.create(), path); + assertThatThrownBy(() -> manager.createTable(schema)) + .hasMessageContaining("cannot be used as rowkind.field"); + assertThat(manager.latest()).isEmpty(); + } + + private FileStoreTable createTable(boolean primaryKey) throws Exception { + return createTable(primaryKey ? "primary-key" : "append"); + } + + private FileStoreTable createTable(String mode) throws Exception { + Path path = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + Schema.Builder builder = schema(mode.equals("primary-key")); + if (mode.equals("data-evolution")) { + builder.option("row-tracking.enabled", "true").option("data-evolution.enabled", "true"); + } + new FileSystemSchemaManager(LocalFileIO.create(), path).createTable(builder.build()); + return FileStoreTableFactory.create(LocalFileIO.create(), path); + } + + private Schema.Builder schema(boolean primaryKey) { + Schema.Builder builder = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .generatedColumn("dt", DataTypes.DATE().notNull(), "CAST(ts AS DATE)") + .column("ts", DataTypes.TIMESTAMP(3)) + .partitionKeys("dt"); + if (primaryKey) { + builder.primaryKey("id", "dt").option("bucket", "1"); + } + return builder; + } + + private void validate(Schema schema) { + SchemaValidation.validateTableSchema( + new TableSchema( + 0, + schema.fields(), + 10, + schema.partitionKeys(), + schema.primaryKeys(), + schema.options(), + null)); + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java index fb3cd0bcab8c..d5b676562614 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java @@ -1095,8 +1095,19 @@ private byte[] decodeRefreshHandlerBytes(String refreshHandlerBytes) { public static Schema fromCatalogTable(CatalogBaseTable catalogTable) { @SuppressWarnings("unchecked") - ResolvedSchema schema = - ((ResolvedCatalogBaseTable) catalogTable).getResolvedSchema(); + ResolvedCatalogBaseTable resolvedTable = + (ResolvedCatalogBaseTable) catalogTable; + ResolvedSchema schema = resolvedTable.getResolvedSchema(); + Map generationExpressions = + resolvedTable.getOrigin() instanceof DataCatalogTable + ? ((DataCatalogTable) resolvedTable.getOrigin()) + .table().rowType().getFields().stream() + .filter(f -> f.generationExpression() != null) + .collect( + Collectors.toMap( + f -> f.name(), + f -> f.generationExpression())) + : Collections.emptyMap(); RowType rowType = (RowType) schema.toPhysicalRowDataType().getLogicalType(); Map options = new HashMap<>(catalogTable.getOptions()); @@ -1137,7 +1148,9 @@ public static Schema fromCatalogTable(CatalogBaseTable catalogTable) { options, blobFileFields, blobInlineFields), - columnComments.get(field.getName()))); + columnComments.get(field.getName()), + null, + generationExpressions.get(field.getName()))); return schemaBuilder.build(); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java index 7c1b190e0792..b09fde10d1d2 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/clone/schema/ClonePaimonSchemaFunction.java @@ -102,7 +102,13 @@ public void processElement( .rowType() .getFields() .forEach( - f -> builder.column(f.name(), f.type(), f.description(), f.defaultValue())); + f -> + builder.column( + f.name(), + f.type(), + f.description(), + f.defaultValue(), + f.generationExpression())); builder.partitionKeys(sourceTable.partitionKeys()); builder.primaryKey(sourceTable.primaryKeys()); Map tableOptions = new HashMap<>(sourceTable.options()); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java index a7f5c3e5e2a3..751e6192b1bf 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java @@ -19,12 +19,17 @@ package org.apache.paimon.flink; import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.types.DataTypes; import org.apache.paimon.utils.BlockingIterator; import org.apache.paimon.utils.StringUtils; import org.apache.flink.table.catalog.CatalogPartition; import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogTable; import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; import org.apache.flink.types.Row; @@ -59,6 +64,64 @@ /** ITCase for catalog tables. */ public class CatalogTableITCase extends CatalogITCaseBase { + @Test + void testGeneratedPartitionSchemaRoundTrip() throws Exception { + Schema schema = + Schema.newBuilder() + .column("ts", DataTypes.TIMESTAMP(3)) + .generatedColumn("dt", DataTypes.DATE(), "CAST(ts AS DATE)") + .partitionKeys("dt") + .build(); + flinkCatalog() + .catalog() + .createTable( + Identifier.create(tEnv.getCurrentDatabase(), "generated_events"), + schema, + false); + CatalogTable source = table("generated_events"); + ResolvedCatalogTable resolved = + new ResolvedCatalogTable(source, tEnv.from("generated_events").getResolvedSchema()); + assertThat(FlinkCatalog.fromCatalogTable(resolved).fields().get(1).generationExpression()) + .isEqualTo("CAST(ts AS DATE)"); + sql("ALTER TABLE generated_events MODIFY dt DATE COMMENT 'generated date'"); + assertThat(paimonTable("generated_events").rowType().getField("dt").generationExpression()) + .isEqualTo("CAST(ts AS DATE)"); + } + + @Test + void testDataEvolutionPartialWritePreservesGeneratedColumns() throws Exception { + tEnv.getConfig().set(TABLE_DML_SYNC, true); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .generatedColumn("label", DataTypes.STRING(), "UPPER(name)") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .build(); + flinkCatalog() + .catalog() + .createTable( + Identifier.create(tEnv.getCurrentDatabase(), "generated_evolution"), + schema, + false); + tEnv.executeSql( + "INSERT INTO generated_evolution VALUES (1, 'old', 'stored'), (2, 'other', 'OTHER')") + .await(); + sql("CREATE TABLE generated_updates (id INT, name STRING)"); + tEnv.executeSql("INSERT INTO generated_updates VALUES (1, 'new')").await(); + sql( + "CALL sys.data_evolution_merge_into('default.generated_evolution', '', '', 'generated_updates', " + + "'generated_evolution.id=generated_updates.id', 'name=generated_updates.name', 1)"); + assertThat(sql("SELECT * FROM generated_evolution")) + .containsExactlyInAnyOrder(Row.of(1, "new", "stored"), Row.of(2, "other", "OTHER")); + sql( + "CALL sys.data_evolution_merge_into('default.generated_evolution', '', '', 'generated_updates', " + + "'generated_evolution.id=generated_updates.id', 'label=generated_updates.name', 1)"); + assertThat(sql("SELECT * FROM generated_evolution")) + .containsExactlyInAnyOrder(Row.of(1, "new", "new"), Row.of(2, "other", "OTHER")); + } + @Override protected Map catalogOptions() { Map options = new HashMap<>(); diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java index 03f0a5e8cde2..644afd761403 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java @@ -36,6 +36,7 @@ import org.apache.paimon.spark.catalog.SupportView; import org.apache.paimon.spark.catalog.functions.FunctionIdentifierConverter; import org.apache.paimon.spark.catalog.functions.PaimonFunctions; +import org.apache.paimon.spark.catalyst.analysis.PaimonGeneratedColumn; import org.apache.paimon.spark.utils.CatalogUtils; import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.iceberg.IcebergTable; @@ -46,6 +47,7 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.VectorType; import org.apache.paimon.utils.ExceptionUtils; +import org.apache.paimon.utils.GeneratedColumn; import org.apache.paimon.utils.StringUtils; import org.apache.spark.sql.PaimonSparkSession$; @@ -58,13 +60,17 @@ import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; import org.apache.spark.sql.catalyst.catalog.PaimonV1FunctionRegistry; import org.apache.spark.sql.catalyst.expressions.Expression; +import org.apache.spark.sql.catalyst.parser.ParseException; import org.apache.spark.sql.catalyst.parser.extensions.UnResolvedPaimonV1Function; +import org.apache.spark.sql.connector.catalog.CatalogV2Util; +import org.apache.spark.sql.connector.catalog.Column; import org.apache.spark.sql.connector.catalog.FunctionCatalog; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.NamespaceChange; import org.apache.spark.sql.connector.catalog.StagedTable; import org.apache.spark.sql.connector.catalog.SupportsNamespaces; import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.connector.catalog.TableCatalogCapability; import org.apache.spark.sql.connector.catalog.TableChange; import org.apache.spark.sql.connector.catalog.functions.UnboundFunction; import org.apache.spark.sql.connector.expressions.FieldReference; @@ -77,6 +83,7 @@ import org.apache.spark.sql.internal.SQLConf; import org.apache.spark.sql.types.ArrayType; import org.apache.spark.sql.types.BinaryType; +import org.apache.spark.sql.types.MetadataBuilder; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.util.CaseInsensitiveStringMap; @@ -88,6 +95,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.EnumSet; import java.util.HashMap; import java.util.List; import java.util.Locale; @@ -102,6 +110,7 @@ import static org.apache.paimon.spark.SparkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB; import static org.apache.paimon.spark.SparkCatalogOptions.V1FUNCTION_ENABLED; import static org.apache.paimon.spark.SparkTypeUtils.CURRENT_DEFAULT_COLUMN_METADATA_KEY; +import static org.apache.paimon.spark.SparkTypeUtils.GENERATION_EXPRESSION_METADATA_KEY; import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType; import static org.apache.paimon.spark.util.OptionUtils.checkRequiredConfigurations; import static org.apache.paimon.spark.util.OptionUtils.copyWithSQLConf; @@ -126,6 +135,14 @@ public class SparkCatalog extends SparkBaseCatalog private static final Logger LOG = LoggerFactory.getLogger(SparkCatalog.class); public static final String FUNCTION_DEFINITION_NAME = "spark"; + + @Override + public Set capabilities() { + return EnumSet.of( + TableCatalogCapability.SUPPORT_COLUMN_DEFAULT_VALUE, + TableCatalogCapability.SUPPORTS_CREATE_TABLE_WITH_GENERATED_COLUMNS); + } + private static final String PRIMARY_KEY_IDENTIFIER = "primary-key"; private Catalog catalog; @@ -375,6 +392,37 @@ public org.apache.spark.sql.connector.catalog.Table alterTable( } } + /** Spark separates generation expressions from StructField metadata in its V2 columns. */ + @Override + public org.apache.spark.sql.connector.catalog.Table createTable( + Identifier ident, + Column[] columns, + Transform[] partitions, + Map properties) + throws TableAlreadyExistsException, NoSuchNamespaceException { + return createTable(ident, generatedColumnSchema(columns), partitions, properties); + } + + private static StructType generatedColumnSchema(Column[] columns) { + StructField[] fields = CatalogV2Util.v2ColumnsToStructType(columns).fields(); + for (int i = 0; i < columns.length; i++) { + if (columns[i].generationExpression() != null) { + fields[i] = + new StructField( + fields[i].name(), + fields[i].dataType(), + fields[i].nullable(), + new MetadataBuilder() + .withMetadata(fields[i].metadata()) + .putString( + GENERATION_EXPRESSION_METADATA_KEY, + columns[i].generationExpression()) + .build()); + } + } + return new StructType(fields); + } + @Override public org.apache.spark.sql.connector.catalog.Table createTable( Identifier ident, @@ -431,6 +479,36 @@ && isFormatTable(provider) } } + @Override + public StagedTable stageCreate( + Identifier ident, + Column[] columns, + Transform[] partitions, + Map properties) + throws TableAlreadyExistsException, NoSuchNamespaceException { + return stageCreate(ident, generatedColumnSchema(columns), partitions, properties); + } + + @Override + public StagedTable stageReplace( + Identifier ident, + Column[] columns, + Transform[] partitions, + Map properties) + throws NoSuchNamespaceException, NoSuchTableException { + return stageReplace(ident, generatedColumnSchema(columns), partitions, properties); + } + + @Override + public StagedTable stageCreateOrReplace( + Identifier ident, + Column[] columns, + Transform[] partitions, + Map properties) + throws NoSuchNamespaceException { + return stageCreateOrReplace(ident, generatedColumnSchema(columns), partitions, properties); + } + @Override public StagedTable stageCreate( Identifier ident, @@ -674,13 +752,40 @@ private Schema toInitialSchema( type = toPaimonType(field.dataType()).copy(field.nullable()); } String comment = field.getComment().getOrElse(() -> null); - if (field.metadata().contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) { - String defaultValue = - field.metadata().getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY); - schemaBuilder.column(name, type, comment, defaultValue); - } else { - schemaBuilder.column(name, type, comment); + String generationExpression = + field.metadata().contains(GENERATION_EXPRESSION_METADATA_KEY) + ? field.metadata().getString(GENERATION_EXPRESSION_METADATA_KEY) + : null; + if (generationExpression != null) { + try { + generationExpression = + PaimonGeneratedColumn.expressionSql( + SparkSession.active() + .sessionState() + .sqlParser() + .parseExpression(generationExpression)); + } catch (ParseException e) { + throw new IllegalArgumentException( + "Invalid generation expression: " + generationExpression, e); + } + generationExpression = + GeneratedColumn.normalizeExpression( + generationExpression, + SparkTypeUtils.toPaimonRowType(schema), + source -> + resolveColumnNames( + Collections.singletonList(source), + fieldNames) + .get(0)); } + schemaBuilder.column( + name, + type, + comment, + field.metadata().contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY) + ? field.metadata().getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY) + : null, + generationExpression); } return schemaBuilder.build(); } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java index 7ced9f29ffee..2cdbca6d787d 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkTypeUtils.java @@ -67,11 +67,11 @@ public class SparkTypeUtils { private SparkTypeUtils() {} - /** - * Copy here from Spark ResolveDefaultColumnsUtils for old Spark versions. - */ + /** Copy here from Spark ResolveDefaultColumnsUtils for old Spark versions. */ public static final String CURRENT_DEFAULT_COLUMN_METADATA_KEY = "CURRENT_DEFAULT"; + public static final String GENERATION_EXPRESSION_METADATA_KEY = "GENERATION_EXPRESSION"; + // Spark SQL functions store DEFAULT in a different metadata key than table columns. public static final String SQL_FUNCTION_DEFAULT_METADATA_KEY = "default"; @@ -108,13 +108,14 @@ public static boolean containsCharType(org.apache.paimon.types.DataType type) { if (type instanceof CharType) { return true; } else if (type instanceof RowType) { - return ((RowType) type).getFields().stream() - .anyMatch(field -> containsCharType(field.type())); + return ((RowType) type) + .getFields().stream().anyMatch(field -> containsCharType(field.type())); } else if (type instanceof ArrayType) { return containsCharType(((ArrayType) type).getElementType()); } else if (type instanceof MapType) { MapType mapType = (MapType) type; - return containsCharType(mapType.getKeyType()) || containsCharType(mapType.getValueType()); + return containsCharType(mapType.getKeyType()) + || containsCharType(mapType.getValueType()); } else if (type instanceof MultisetType) { return containsCharType(((MultisetType) type).getElementType()); } @@ -309,17 +310,22 @@ public DataType visit(RowType rowType) { List fields = new ArrayList<>(rowType.getFieldCount()); for (DataField field : rowType.getFields()) { MetadataBuilder metadataBuilder = new MetadataBuilder(); + if (field.generationExpression() != null) { + metadataBuilder.putString( + GENERATION_EXPRESSION_METADATA_KEY, field.generationExpression()); + } if (field.defaultValue() != null) { - metadataBuilder.putString(CURRENT_DEFAULT_COLUMN_METADATA_KEY, field.defaultValue()); - metadataBuilder.putString(SQL_FUNCTION_DEFAULT_METADATA_KEY, field.defaultValue()); + metadataBuilder.putString( + CURRENT_DEFAULT_COLUMN_METADATA_KEY, field.defaultValue()); + metadataBuilder.putString( + SQL_FUNCTION_DEFAULT_METADATA_KEY, field.defaultValue()); } StructField structField = DataTypes.createStructField( field.name(), field.type().accept(this), field.type().isNullable(), - metadataBuilder.build() - ); + metadataBuilder.build()); structField = Optional.ofNullable(field.description()) .map(structField::withComment) @@ -396,15 +402,21 @@ public org.apache.paimon.types.DataType struct( String comment = field.getComment().getOrElse(() -> null); String defaultValue = null; if (field.metadata().contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) { - defaultValue = - field.metadata().getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY); + defaultValue = field.metadata().getString(CURRENT_DEFAULT_COLUMN_METADATA_KEY); } else if (field.metadata().contains(SQL_FUNCTION_DEFAULT_METADATA_KEY)) { - defaultValue = - field.metadata().getString(SQL_FUNCTION_DEFAULT_METADATA_KEY); + defaultValue = field.metadata().getString(SQL_FUNCTION_DEFAULT_METADATA_KEY); } newFields.add( new DataField( - atomicInteger.incrementAndGet(), field.name(), fieldType, comment, defaultValue)); + atomicInteger.incrementAndGet(), + field.name(), + fieldType, + comment, + defaultValue, + field.metadata().contains(GENERATION_EXPRESSION_METADATA_KEY) + ? field.metadata() + .getString(GENERATION_EXPRESSION_METADATA_KEY) + : null)); } return new RowType(newFields); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala index 527ca1faad64..26f9c3209c6a 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala @@ -27,10 +27,11 @@ import org.apache.paimon.spark.commands.{PaimonAnalyzeFormatTablePartitionsComma import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.spark.util.OptionUtils import org.apache.paimon.table.FileStoreTable +import org.apache.paimon.utils.GeneratedColumn import org.apache.spark.sql.{PaimonUtils, SparkSession} import org.apache.spark.sql.catalyst.analysis.ResolvedTable -import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute} import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper, _} import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.trees.TreeNodeTag @@ -119,7 +120,8 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { table: DataSourceV2Relation, options: Options, mergeSchemaEnabled: Boolean): LogicalPlan = { - val query = stripHiveDynamicPartitionMarker(v2WriteCommand.query) + val input = stripHiveDynamicPartitionMarker(v2WriteCommand.query) + val query = generateColumns(input, table, v2WriteCommand.isByName, mergeSchemaEnabled) val hiveStyleDynamicPartitionEnabled = OptionUtils.hiveStyleDynamicPartitionEnabled() hiveDynamicPartitionColumns(v2WriteCommand.query) match { case Some(dynamicPartitionColumns) @@ -152,6 +154,57 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { } } + private def generateColumns( + query: LogicalPlan, + table: DataSourceV2Relation, + byName: Boolean, + mergeSchemaEnabled: Boolean): LogicalPlan = { + val rowType = table.table.asInstanceOf[SparkTable].getTable.rowType() + val generated = rowType.getFields.asScala.filter(_.generationExpression() != null) + if (generated.isEmpty) { + return query + } + val omitted = if (byName) { + generated.filterNot(f => query.output.exists(a => conf.resolver(a.name, f.name()))) + } else if (query.output.size == table.output.size - generated.size) { + generated + } else { + Seq.empty + } + if (omitted.isEmpty) { + query + } else { + omitted.foreach(PaimonGeneratedColumn.validateResultType) + val names = omitted.map(_.name()).toSet + val aligned = resolveWriteOutput( + query, + table.name, + table.output.filterNot(a => names.contains(a.name)), + byName, + mergeSchemaEnabled) + val projected = table.output.map { + attr => + omitted.find(f => conf.resolver(f.name(), attr.name)) match { + case Some(field) => + val sources = GeneratedColumn + .create(rowType, field) + .sourceNames() + .asScala + .map(rowType.getField) + val inputs = sources.map { + source => aligned.output.find(a => conf.resolver(a.name, source.name())).get + } + val expression = PaimonGeneratedColumn(inputs.toSeq, sources.toSeq, field) + Alias(expression, attr.name)(explicitMetadata = Some(attr.metadata)) + case None => aligned.output.find(a => conf.resolver(a.name, attr.name)).get + } + } + val extras = + aligned.output.filterNot(a => table.output.exists(t => conf.resolver(t.name, a.name))) + Project(projected ++ extras, aligned) + } + } + private def hiveDynamicPartitionColumns(query: LogicalPlan): Option[Seq[String]] = { query.collectFirst { case PaimonHiveDynamicPartitionQuery(dynamicPartitionColumns, _) => diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonGeneratedColumn.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonGeneratedColumn.scala new file mode 100644 index 000000000000..dc1148c2e8fc --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonGeneratedColumn.scala @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.catalyst.analysis + +import org.apache.paimon.data.GenericRow +import org.apache.paimon.spark.{SparkInternalRowWrapper, SparkTypeUtils} +import org.apache.paimon.spark.data.SparkInternalRow +import org.apache.paimon.types.{ArrayType, BlobType, DataField, DataType => PaimonType, MapType, MultisetType, RowType, TimestampType} +import org.apache.paimon.utils.{GeneratedColumnRow, InternalRowUtils} + +import org.apache.spark.sql.catalyst.{InternalRow => CatalystRow} +import org.apache.spark.sql.catalyst.expressions.{Expression, GenericInternalRow, LeafExpression, Literal, Unevaluable} +import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback +import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType} + +import scala.collection.JavaConverters._ + +/** Shares default normalization and expression evaluation with the storage writer. */ +case class PaimonGeneratedColumn( + override val children: Seq[Expression], + sources: Seq[DataField], + field: DataField) + extends Expression + with CodegenFallback { + + @transient private lazy val sourceType = new RowType(sources.asJava) + @transient private lazy val generator = + new GeneratedColumnRow(new RowType((sources :+ field).asJava), sourceType) + @transient private lazy val input = new GenericInternalRow(sources.size) + @transient private lazy val wrapper = new SparkInternalRowWrapper( + StructType(children.zipWithIndex.map { + case (child, index) => StructField(sources(index).name(), child.dataType, child.nullable) + }), + sources.size) + @transient private lazy val result = new GenericRow(1) + @transient private lazy val sparkResult = SparkInternalRow.create(RowType.of(field)) + + override def dataType: DataType = SparkTypeUtils.fromPaimonType(field.`type`()) + + override def nullable: Boolean = true + + override def eval(row: CatalystRow): Any = { + children.indices.foreach(i => input.update(i, children(i).eval(row))) + val generated = generator.replaceRow(wrapper.replace(input)) + result.setField(0, InternalRowUtils.get(generated, sources.size, field.`type`())) + sparkResult.replace(result).get(0, dataType) + } + + override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = + copy(children = newChildren) +} + +object PaimonGeneratedColumn { + + /** Store SQL-standard string literals independently of Spark's version-specific SQL printer. */ + def expressionSql(expression: Expression): String = { + expression.transformUp { + case literal: Literal if literal.dataType == StringType => SqlStringLiteral(literal) + }.sql + } + + private case class SqlStringLiteral(literal: Literal) extends LeafExpression with Unevaluable { + override def dataType: DataType = literal.dataType + override def nullable: Boolean = literal.nullable + override def sql: String = + if (literal.value == null) "NULL" else "'" + literal.value.toString.replace("'", "''") + "'" + } + + def validateResultType(field: DataField): Unit = { + def representable(dataType: PaimonType): Boolean = dataType match { + case timestamp: TimestampType => timestamp.getPrecision <= 6 + case _: BlobType => false + case array: ArrayType => representable(array.getElementType) + case map: MapType => representable(map.getKeyType) && representable(map.getValueType) + case multiset: MultisetType => representable(multiset.getElementType) + case row: RowType => row.getFields.asScala.forall(f => representable(f.`type`())) + case _ => true + } + require( + representable(field.`type`()), + s"Spark cannot losslessly write generated column '${field.name()}' with BLOB results or TIMESTAMP precision greater than 6." + ) + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/GeneratedPartitionColumnTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/GeneratedPartitionColumnTest.scala new file mode 100644 index 000000000000..e8a2944b4922 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/GeneratedPartitionColumnTest.scala @@ -0,0 +1,344 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.sql + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.schema.Schema +import org.apache.paimon.spark.PaimonSparkTestBase +import org.apache.paimon.types.{DataTypes => PaimonTypes} + +import org.apache.spark.sql.Row + +import java.sql.Date + +/** SQL and catalog contracts for physical generated columns. */ +class GeneratedPartitionColumnTest extends PaimonSparkTestBase { + + test("resolve quoted and case-insensitive source columns") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events ( + |id INT, `event time` TIMESTAMP_NTZ, + |dt DATE GENERATED ALWAYS AS (CAST(`EVENT TIME` AS DATE)) + |) USING paimon PARTITIONED BY (dt)""".stripMargin) + sql("""INSERT INTO generated_events (`event time`, id) + |VALUES (CAST('2026-10-30 12:00:00' AS TIMESTAMP_NTZ), 1)""".stripMargin) + checkAnswer( + sql("SELECT id, dt FROM generated_events"), + Seq(Row(1, Date.valueOf("2026-10-30")))) + } + } + + for (mode <- Seq("primary-key", "data-evolution")) { + test(s"generated partition insert for $mode tables") { + assume(gteqSpark3_5) + withTable("generated_events") { + val properties = if (mode == "primary-key") { + "'primary-key'='id,dt', 'bucket'='1'" + } else { + "'row-tracking.enabled'='true', 'data-evolution.enabled'='true'" + } + sql(s"""CREATE TABLE generated_events ( + |id INT, event_time TIMESTAMP_NTZ, + |dt DATE GENERATED ALWAYS AS (CAST(event_time AS DATE)) + |) USING paimon PARTITIONED BY (dt) TBLPROPERTIES ($properties)""".stripMargin) + sql("""INSERT INTO generated_events (id, event_time) + |VALUES (1, CAST('2026-10-30 12:00:00' AS TIMESTAMP_NTZ))""".stripMargin) + checkAnswer( + sql("SELECT id, dt FROM generated_events"), + Seq(Row(1, Date.valueOf("2026-10-30")))) + } + } + } + + test("generate omitted partition columns before routing") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql(""" + |CREATE TABLE generated_events ( + | id INT, + | event_time TIMESTAMP_NTZ, + | dt DATE GENERATED ALWAYS AS (CAST(event_time AS DATE)) + |) USING paimon PARTITIONED BY (dt) + |""".stripMargin) + assert( + loadTable("generated_events").rowType().getField("dt").generationExpression() == + "CAST(event_time AS DATE)") + + sql("""INSERT INTO generated_events + |SELECT 1, CAST('1969-12-31 23:59:59.999999' AS TIMESTAMP_NTZ) + |""".stripMargin) + sql("""INSERT INTO generated_events (id, event_time) + |VALUES (2, CAST('2026-10-30 00:00:00' AS TIMESTAMP_NTZ)) + |""".stripMargin) + sql("""INSERT INTO generated_events + |VALUES (3, CAST('2026-10-31 12:00:00' AS TIMESTAMP_NTZ), DATE '2026-10-31') + |""".stripMargin) + checkAnswer( + sql("SELECT id, dt FROM generated_events"), + Seq( + Row(1, Date.valueOf("1969-12-31")), + Row(2, Date.valueOf("2026-10-30")), + Row(3, Date.valueOf("2026-10-31"))) + ) + + sql("ALTER TABLE generated_events ALTER COLUMN dt COMMENT 'generated partition'") + assert( + loadTable("generated_events").rowType().getField("dt").generationExpression() == + "CAST(event_time AS DATE)") + } + } + + test("preserve supplied generated partition values and explicit nulls") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events ( + |id INT, event_time TIMESTAMP_NTZ, + |dt DATE GENERATED ALWAYS AS (CAST(event_time AS DATE)) + |) USING paimon PARTITIONED BY (dt)""".stripMargin) + for ((partition, id) <- Seq("DATE '2026-10-31'", "CAST(NULL AS DATE)").zipWithIndex) { + sql(s"""INSERT INTO generated_events + |VALUES ($id, CAST('2026-10-30 12:00:00' AS TIMESTAMP_NTZ), $partition) + |""".stripMargin) + } + sql("INSERT INTO generated_events (id, event_time) VALUES (2, NULL)") + checkAnswer( + sql("SELECT id, dt FROM generated_events"), + Seq(Row(0, Date.valueOf("2026-10-31")), Row(1, null), Row(2, null))) + } + } + + test("generate non-partition columns with transforms and nested scalar calls") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events ( + |id INT, name STRING, event_time TIMESTAMP_NTZ, days INT, + |label STRING GENERATED ALWAYS AS (CONCAT(UPPER(`NAME`), '-', SUBSTRING(name, 2, 2))), + |event_year INT GENERATED ALWAYS AS (YEAR(event_time)), + |dt DATE GENERATED ALWAYS AS (DATE_ADD(CAST(event_time AS DATE), days)) + |) USING paimon PARTITIONED BY (dt)""".stripMargin) + sql("""INSERT INTO generated_events (days, event_time, name, id) + |VALUES (2, CAST('1969-12-31 12:00:00' AS TIMESTAMP_NTZ), 'abc', 1)""".stripMargin) + checkAnswer( + sql("SELECT id, label, event_year, dt FROM generated_events"), + Seq(Row(1, "ABC-bc", 1969, Date.valueOf("1970-01-02")))) + sql("""INSERT INTO generated_events + |VALUES (2, 'abc', CAST('1969-12-31 12:00:00' AS TIMESTAMP_NTZ), 2, + |'wrong', 1969, DATE '1970-01-02')""".stripMargin) + checkAnswer( + sql("SELECT id, label FROM generated_events"), + Seq(Row(1, "ABC-bc"), Row(2, "wrong"))) + } + withTable("generated_events") { + sql("""CREATE TABLE generated_events (name STRING, + |label STRING GENERATED ALWAYS AS (LOWER(name))) USING paimon""".stripMargin) + sql("INSERT INTO generated_events VALUES ('AbC')") + checkAnswer(sql("SELECT label FROM generated_events"), Seq(Row("abc"))) + } + } + + test("reject nested references and unsupported generation functions") { + assume(gteqSpark3_5) + for ( + (expression, dataType, expected) <- Seq( + ("ABS(id)", "INT", "Unsupported generation"), + ("RAND()", "DOUBLE", "not deterministic"), + ("UPPER(nested.name)", "STRING", "top-level field")) + ) { + withTable("generated_events") { + val error = intercept[Throwable] { + sql(s"""CREATE TABLE generated_events ( + |id INT, nested STRUCT, + |value $dataType GENERATED ALWAYS AS ($expression) + |) USING paimon""".stripMargin) + } + assert(error.getMessage.contains(expected)) + } + } + } + + test("canonicalize quoted and escaped string literals in generation expressions") { + assume(gteqSpark3_5) + for (literal <- Seq("'NAME''s'", """'C:\\path'""", """'line\nend'""", """'quote\'end'""")) { + withTable("generated_events") { + sql( + s"""CREATE TABLE generated_events (name STRING, + |label STRING GENERATED ALWAYS AS (CONCAT(name, $literal))) USING paimon""".stripMargin) + sql("INSERT INTO generated_events (name) VALUES ('ab')") + checkAnswer( + sql("SELECT label FROM generated_events"), + Seq(sql(s"SELECT CONCAT('ab', $literal)").head())) + } + } + } + + test("use storage transforms and defaults for omitted generated values") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events (id INT, ts TIMESTAMP_NTZ, name STRING, + |text STRING GENERATED ALWAYS AS (CAST(ts AS STRING)), + |trimmed STRING GENERATED ALWAYS AS (TRIM('x', name)), + |literal STRING GENERATED ALWAYS AS (CONCAT(name, 'NAME''s')) + |) USING paimon""".stripMargin) + sql("""INSERT INTO generated_events (id, ts, name) + |VALUES (1, CAST('1970-01-01 00:00:00' AS TIMESTAMP_NTZ), 'xabcx')""".stripMargin) + sql("""INSERT INTO generated_events VALUES + |(2, CAST('1970-01-01 00:00:00' AS TIMESTAMP_NTZ), 'xabcx', + |'1970-01-01 00:00:00.000000', 'abc', CONCAT('xabcx', 'NAME''s'))""".stripMargin) + val suffix = sql("SELECT 'NAME''s'").head().getString(0) + checkAnswer( + sql("SELECT text, trimmed, literal FROM generated_events"), + Seq.fill(2)(Row("1970-01-01 00:00:00.000000", "abc", "xabcx" + suffix))) + } + withTable("generated_events") { + sql("""CREATE TABLE generated_events (id INT, name STRING DEFAULT 'abc', + |label STRING GENERATED ALWAYS AS (UPPER(name))) USING paimon""".stripMargin) + sql("INSERT INTO generated_events (id, name) VALUES (1, NULL)") + sql("INSERT INTO generated_events VALUES (2, NULL, 'ABC')") + checkAnswer( + sql("SELECT id, name, label FROM generated_events"), + Seq(Row(1, "abc", "ABC"), Row(2, "abc", "ABC"))) + } + } + + test("skip supplied expressions when generating other omitted columns") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events (id INT, text STRING, + |number INT GENERATED ALWAYS AS (CAST(text AS INT)), + |label STRING GENERATED ALWAYS AS (UPPER(text))) USING paimon""".stripMargin) + sql("INSERT INTO generated_events (id, text, number) VALUES (1, 'invalid', 42)") + sql("INSERT INTO generated_events VALUES (2, 'invalid', NULL, NULL)") + checkAnswer( + sql("SELECT id, number, label FROM generated_events"), + Seq(Row(1, 42, "INVALID"), Row(2, null, null))) + } + } + + test("preserve extra columns and generation metadata when merging schemas") { + assume(gteqSpark3_5) + withTable("generated_events") { + sql("""CREATE TABLE generated_events (id INT, name STRING, + |label STRING GENERATED ALWAYS AS (UPPER(name))) USING paimon""".stripMargin) + sql("SELECT 1 AS id, 'ab' AS name, 99 AS extra") + .writeTo("generated_events") + .option("write.merge-schema", "true") + .append() + sql("REFRESH TABLE generated_events") + sql("SELECT 2 AS id, 'cd' AS name, 'CD' AS label, 100 AS extra, 7 AS other") + .writeTo("generated_events") + .option("write.merge-schema", "true") + .append() + sql("REFRESH TABLE generated_events") + assert( + loadTable("generated_events").rowType().getField("label").generationExpression() != null) + checkAnswer( + sql("SELECT id, label, extra, other FROM generated_events"), + Seq(Row(1, "AB", 99, null), Row(2, "CD", 100, 7))) + } + } + + test("reject generated results Spark cannot represent without losing nanoseconds") { + assume(gteqSpark3_5) + val timestamp = PaimonTypes.TIMESTAMP(9) + for ( + (sourceType, targetType, expression, value) <- Seq( + ( + PaimonTypes.STRING(), + timestamp, + "CAST(source AS TIMESTAMP(9))", + "'2026-01-01 00:00:00.123456789'"), + ( + PaimonTypes.ARRAY(timestamp), + PaimonTypes.ARRAY(timestamp), + "source", + "array(CAST('2026-01-01 00:00:00' AS TIMESTAMP_NTZ))") + ) + ) { + withTable("generated_events") { + paimonCatalog.createTable( + Identifier.create(dbName0, "generated_events"), + Schema + .newBuilder() + .column("source", sourceType) + .generatedColumn("generated", targetType, expression) + .build(), + false + ) + val error = intercept[Throwable] { + sql(s"INSERT INTO generated_events (source) SELECT $value") + } + assert(error.getMessage.contains("TIMESTAMP precision greater than 6")) + assert(loadTable("generated_events").snapshotManager().latestSnapshotId() == null) + sql(s"INSERT INTO generated_events (source, generated) SELECT $value, $value") + checkAnswer(sql("SELECT generated = source FROM generated_events"), Seq(Row(true))) + } + } + } + + test("reject generated BLOB results before writing through a lossy Spark representation") { + assume(gteqSpark3_5) + withTable("generated_events") { + paimonCatalog.createTable( + Identifier.create(dbName0, "generated_events"), + Schema + .newBuilder() + .column("id", PaimonTypes.INT()) + .column("source", PaimonTypes.BLOB()) + .generatedColumn("generated", PaimonTypes.BLOB(), "source") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .build(), + false + ) + val error = intercept[Throwable] { + sql("INSERT INTO generated_events (id, source) SELECT 1, unhex('010203')") + } + assert(error.getMessage.contains("BLOB results")) + assert(loadTable("generated_events").snapshotManager().latestSnapshotId() == null) + sql("INSERT INTO generated_events VALUES (1, unhex('010203'), unhex('aabb'))") + checkAnswer(sql("SELECT hex(generated) FROM generated_events"), Seq(Row("AABB"))) + } + } + + test("preserve supplied and untouched generated values in row-tracking and DE updates") { + assume(gteqSpark3_5) + for (dataEvolution <- Seq(false, true)) { + withTable("generated_events") { + sql(s"""CREATE TABLE generated_events (id INT, name STRING, + |label STRING GENERATED ALWAYS AS (UPPER(name))) USING paimon + |TBLPROPERTIES ('row-tracking.enabled'='true', + |'data-evolution.enabled'='$dataEvolution')""".stripMargin) + sql("INSERT INTO generated_events (id, name) VALUES (1, 'old'), (2, 'other')") + sql("UPDATE generated_events SET name='new' WHERE id=1") + checkAnswer( + sql("SELECT id, name, label FROM generated_events"), + Seq(Row(1, "new", "OLD"), Row(2, "other", "OTHER"))) + sql("UPDATE generated_events SET label='external' WHERE id=1") + checkAnswer( + sql("SELECT id, name, label FROM generated_events"), + Seq(Row(1, "new", "external"), Row(2, "other", "OTHER"))) + sql("UPDATE generated_events SET name='changed', label=NULL WHERE id=1") + checkAnswer( + sql("SELECT id, name, label FROM generated_events"), + Seq(Row(1, "changed", null), Row(2, "other", "OTHER"))) + } + } + } +}