You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/06/01 03:47:04 UTC
(paimon) branch master updated: [spark] Decouple type-widening from merge-schema with an explicit switch (#8042)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c73e287aaa [spark] Decouple type-widening from merge-schema with an explicit switch (#8042)
c73e287aaa is described below
commit c73e287aaaeaea926a260ebb0dc6e8643e0b39bc
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon Jun 1 11:46:58 2026 +0800
[spark] Decouple type-widening from merge-schema with an explicit switch (#8042)
Today `write.merge-schema` couples column-addition with *unconditional*
type widening. That has two problems: it can attempt unsupported
widenings (e.g. `ARRAY<INT>` -> `ARRAY<BIGINT>`) on a plain
column-addition write, and the widening behavior is inconsistent between
catalog writes and `MERGE INTO`.
This PR decouples the two with an explicit, opt-in switch:
| Option | Default | Effect |
|--------|---------|--------|
| `write.merge-schema` | false | Evolve schema for **new columns only**;
existing column types are kept and incoming values are cast to them. |
| `write.merge-schema.type-widening` | false | Additionally widen an
existing column type to a wider compatible type (e.g. `INT -> BIGINT`,
`DECIMAL` precision increase). |
| `write.merge-schema.explicit-cast` | false | Additionally allow lossy
casts (e.g. `BIGINT -> INT`, `STRING -> DATE`). |
Two things worth calling out:
1. **The write flow is `compute -> cast -> commit`.** For catalog writes
(`saveAsTable` / `INSERT` / `writeTo`) and `MERGE INTO`, analysis first
computes the merged (evolved) schema **in memory** (nothing is persisted
yet) and casts the incoming data to it; the evolved schema is persisted
only by the final commit, which is **deferred to execution**.
2. **Schema evolution on path-based DataFrame write is intentional.**
`df.write.format("paimon").save(path)` and the streaming sink also
evolve the schema. They have no analysis-time cast hook, so they commit
the evolved schema and write the data as-is (works for column additions
and widening, where the incoming data already matches the evolved
types). This is by design, and is covered by the added
`DataFrameWriteTest` / `PaimonSinkTest` cases.
---
docs/docs/spark/sql-write.md | 189 ++++++-----
docs/generated/spark_connector_configuration.html | 10 +-
.../java/org/apache/paimon/AbstractFileStore.java | 12 +-
.../src/main/java/org/apache/paimon/FileStore.java | 6 +-
.../paimon/privilege/PrivilegedFileStore.java | 8 +-
.../org/apache/paimon/schema/SchemaManager.java | 10 +-
.../apache/paimon/schema/SchemaMergingUtils.java | 49 ++-
.../paimon/operation/FileStoreCommitTest.java | 2 +-
.../paimon/schema/SchemaMergingUtilsTest.java | 365 ++++++++++++++-------
.../apache/paimon/spark/DataFrameWriteTest.scala | 2 -
.../org/apache/paimon/spark/PaimonSinkTest.scala | 1 +
.../MergeIntoPaimonDataEvolutionTable.scala | 2 +
.../spark/commands/MergeIntoPaimonTable.scala | 2 +
.../apache/paimon/spark/SparkConnectorOptions.java | 18 +-
.../analysis/MergeSchemaEvolutionHelper.scala | 39 +--
.../spark/catalyst/analysis/PaimonAnalysis.scala | 21 +-
.../catalyst/analysis/PaimonAssignmentUtils.scala | 10 +-
.../spark/catalyst/analysis/PaimonMergeInto.scala | 3 +-
.../catalyst/analysis/PaimonOutputResolver.scala | 12 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 2 +
.../spark/commands/MergeIntoPaimonTable.scala | 2 +
.../spark/commands/SchemaEvolutionHelper.scala | 200 +++++++++++
.../paimon/spark/commands/SchemaHelper.scala | 168 ----------
.../spark/commands/WriteIntoPaimonTable.scala | 6 +-
.../apache/paimon/spark/sources/PaimonSink.scala | 4 +-
.../org/apache/paimon/spark/util/OptionUtils.scala | 4 +
.../apache/paimon/spark/write/PaimonV2Write.scala | 7 +-
.../scala/org/apache/spark/sql/PaimonUtils.scala | 6 +-
.../org/apache/paimon/spark/PaimonSinkTest.scala | 1 +
.../paimon/spark/sql/DataFrameWriteTestBase.scala | 96 +++---
.../paimon/spark/sql/V2WriteMergeSchemaTest.scala | 1 +
.../paimon/spark/sql/WriteMergeSchemaTest.scala | 137 +++++++-
32 files changed, 884 insertions(+), 511 deletions(-)
diff --git a/docs/docs/spark/sql-write.md b/docs/docs/spark/sql-write.md
index 56899796ea..83d26da622 100644
--- a/docs/docs/spark/sql-write.md
+++ b/docs/docs/spark/sql-write.md
@@ -175,132 +175,108 @@ DELETE FROM my_table WHERE id = 1;
Merges a set of updates, insertions and deletions based on a source table into a target table.
-Note:
-
:::info
-In update clause, to update primary key columns is not supported when the target table is a primary key table.
+Updating primary key columns is not supported when the target table is a primary key table.
:::
-**Example: One**
-
-This is a simple demo that, if a row exists in the target table update it, else insert it.
+### Syntax
```sql
--- Here both source and target tables have the same schema: (a INT, b INT, c STRING), and a is a primary key.
-
MERGE INTO target
USING source
-ON target.a = source.a
-WHEN MATCHED THEN
-UPDATE SET *
-WHEN NOT MATCHED
-THEN INSERT *
+ON <merge condition>
+WHEN MATCHED [AND <condition>] THEN { UPDATE SET ... | DELETE }
+WHEN NOT MATCHED [AND <condition>] THEN INSERT ...
```
-**Example: Two**
+Each `WHEN` clause can be repeated; clauses are evaluated in order, and the first matching one wins for a given row.
-This is a demo with multiple, conditional clauses.
+### Examples
+
+The examples below assume both source and target have schema `(a INT, b INT, c STRING)`, with `a` as the primary key.
+
+Simple upsert — update existing rows, insert new ones:
```sql
--- Here both source and target tables have the same schema: (a INT, b INT, c STRING), and a is a primary key.
+MERGE INTO target
+USING source
+ON target.a = source.a
+WHEN MATCHED THEN UPDATE SET *
+WHEN NOT MATCHED THEN INSERT *
+```
+Multiple conditional clauses:
+
+```sql
MERGE INTO target
USING source
ON target.a = source.a
-WHEN MATCHED AND target.a = 5 THEN
- UPDATE SET b = source.b + target.b -- when matched and meet the condition 1, then update b;
-WHEN MATCHED AND source.c > 'c2' THEN
- UPDATE SET * -- when matched and meet the condition 2, then update all the columns;
-WHEN MATCHED THEN
- DELETE -- when matched, delete this row in target table;
-WHEN NOT MATCHED AND c > 'c9' THEN
- INSERT (a, b, c) VALUES (a, b * 1.1, c) -- when not matched but meet the condition 3, then transform and insert this row;
-WHEN NOT MATCHED THEN
-INSERT * -- when not matched, insert this row without any transformation;
+WHEN MATCHED AND target.a = 5 THEN UPDATE SET b = source.b + target.b
+WHEN MATCHED AND source.c > 'c2' THEN UPDATE SET *
+WHEN MATCHED THEN DELETE
+WHEN NOT MATCHED AND c > 'c9' THEN INSERT (a, b, c) VALUES (a, b * 1.1, c)
+WHEN NOT MATCHED THEN INSERT *
```
### Column Alignment
Assignments are aligned to the target table by **column name**.
-For explicit clauses (`UPDATE SET col = expr` / `INSERT (col list) VALUES ...`), only the mentioned columns are written. Unmentioned target columns preserve their current value for `UPDATE`, or are filled with NULL / `CURRENT_DEFAULT` for `INSERT`.
-
-For star clauses (`UPDATE SET *` / `INSERT *`), `*` expands against the **target** columns. The behavior when source and target columns don't match exactly depends on `spark.paimon.write.merge-schema` (see [Write Merge Schema](#write-merge-schema)):
-
-| Scenario | `merge-schema=false` (default) | `merge-schema=true` |
-|----------|-------------------------------|---------------------|
-| Top-level source-extra columns | Silently dropped (`*` only covers target columns) | Evolved into the target schema |
-| Top-level target columns missing from source | Throws | `UPDATE *` preserves current value; `INSERT *` fills NULL |
-| Nested struct source-extra fields | Throws | Evolved into the target schema |
-| Nested struct target-missing fields | Throws | `UPDATE *` preserves current value; `INSERT *` fills NULL |
-
-The key difference between top-level and nested: under strict mode (`merge-schema=false`), top-level source-extras are silently dropped because `*` never references them, while nested source-extras inside a struct value throw an error to avoid silent data loss.
+- **Explicit clauses** (`UPDATE SET col = expr` / `INSERT (col list) VALUES ...`) — only the mentioned columns are written. Unmentioned target columns preserve their current value for `UPDATE`, or get NULL / `CURRENT_DEFAULT` for `INSERT`.
+- **Star clauses** (`UPDATE SET *` / `INSERT *`) — `*` expands against the **target** columns. When source and target columns don't match exactly, the behavior depends on `spark.paimon.write.merge-schema`; see [Column Alignment by Write Path](#column-alignment-by-write-path) under Write Merge Schema for the full table covering both `MERGE INTO *` and byName `INSERT` paths.
## Write Merge Schema
+When `write.merge-schema` is enabled, Paimon automatically evolves the table schema during write to accommodate new columns in the incoming data, while preserving data integrity.
+
:::info
Since the table schema may be updated during writing, catalog caching needs to be disabled to use this feature. Configure `spark.sql.catalog.<catalogName>.cache-enabled` to `false`.
:::
-Write merge schema is a feature that allows users to easily modify the current schema of a table to adapt to existing data, or new data that changes over time, while maintaining data integrity and consistency.
-
-Paimon supports automatic schema merging of source data and current table data while data is being written, and uses the merged schema as the latest schema of the table, and it only requires configuring `write.merge-schema`.
-
-```scala
-data.write
- .format("paimon")
- .mode("append")
- .option("write.merge-schema", "true")
- .save(location)
-```
-
-When enable `write.merge-schema`, Paimon can allow users to perform the following actions on table schema by default:
-- Adding columns
-- Up-casting the type of column(e.g. Int -> Long)
-
-Paimon also supports explicit type conversions between certain types (e.g. String -> Date, Long -> Int), it requires an explicit configuration `write.merge-schema.explicit-cast`.
-
-Write merge schema can be used in streaming mode at the same time.
-
-```scala
-val inputData = MemoryStream[(Int, String)]
-inputData
- .toDS()
- .toDF("col1", "col2")
- .writeStream
- .format("paimon")
- .option("checkpointLocation", "/path/to/checkpoint")
- .option("write.merge-schema", "true")
- .option("write.merge-schema.explicit-cast", "true")
- .start(location)
-```
+### How It Evolves the Schema
-Here list the configurations.
+Three options control how aggressively the schema evolves; each only takes effect when the previous one is enabled:
<table class="configuration table table-bordered">
<thead>
<tr>
- <th class="text-left" style="width: 20%">Scan Mode</th>
- <th class="text-left" style="width: 60%">Description</th>
+ <th class="text-left" style="width: 30%">Option</th>
+ <th class="text-left" style="width: 70%">Description</th>
</tr>
</thead>
<tbody>
<tr>
<td><h5>write.merge-schema</h5></td>
- <td>If true, merge the data schema and the table schema automatically before write data.</td>
+ <td>If true, evolve the table schema to accept new columns from the incoming data. Existing column types are preserved and incoming values are cast to them; to also widen existing types, enable <code>write.merge-schema.type-widening</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>write.merge-schema.type-widening</h5></td>
+ <td>Only effective when <code>write.merge-schema</code> is true. If true, widen an existing column type when the incoming data has a wider compatible type (e.g. INT -> BIGINT, DECIMAL precision increase). Lossy changes are still rejected unless <code>write.merge-schema.explicit-cast</code> is also true.</td>
</tr>
<tr>
<td><h5>write.merge-schema.explicit-cast</h5></td>
- <td>If true, allow to merge data types if the two types meet the rules for explicit casting.</td>
+ <td>Only effective when <code>write.merge-schema.type-widening</code> is true. If true, also allow lossy type changes between compatible types (e.g. BIGINT -> INT, STRING -> DATE).</td>
</tr>
</tbody>
</table>
-This mode also supports Spark SQL. Here is an example:
+### Examples
+
+DataFrame batch write:
+
+```scala
+data.write
+ .format("paimon")
+ .mode("append")
+ .option("write.merge-schema", "true")
+ .saveAsTable("t")
+```
+
+Spark SQL (requires Spark 3.5+ for `BY NAME`):
```sql
SET `spark.paimon.write.merge-schema` = true;
@@ -308,15 +284,48 @@ SET `spark.paimon.write.merge-schema` = true;
CREATE TABLE t (a INT, b STRING);
INSERT INTO t VALUES (1, '1'), (2, '2');
--- Need using `BY NAME` statement (requires Spark 3.5+)
INSERT INTO t BY NAME SELECT 3 AS a, '3' AS b, 3 AS c;
```
+Streaming write:
+
+```scala
+val inputData = MemoryStream[(Int, String)]
+inputData
+ .toDS()
+ .toDF("col1", "col2")
+ .writeStream
+ .format("paimon")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .option("write.merge-schema", "true")
+ .toTable("t")
+```
+
+### Column Alignment by Write Path
+
+When the source schema doesn't match the target schema exactly, the behavior depends on both `write.merge-schema` and the write path. For nested struct fields, all byName paths behave the same; at the top level, `MERGE INTO *` differs from regular byName `INSERT` because `*` expansion only references target columns.
+
+| Write path | Scenario | `merge-schema=false` (default) | `merge-schema=true` |
+|------------|----------|-------------------------------|---------------------|
+| **byName `INSERT`** (`INSERT INTO ... BY NAME` / `saveAsTable` / `writeTo`) | Top-level source-extra columns | Throws | Evolved into the target schema |
+| | Top-level target columns missing from source | NULL-filled | NULL-filled |
+| | Nested struct source-extra fields | Throws | Evolved into the target schema |
+| | Nested struct target-missing fields | Throws | NULL-filled |
+| **`MERGE INTO *`** (`UPDATE *` / `INSERT *`) | Top-level source-extra columns | Silently dropped (`*` only covers target columns) | Evolved into the target schema |
+| | Top-level target columns missing from source | Throws | `UPDATE *` preserves current value; `INSERT *` fills `CURRENT_DEFAULT` (or NULL when no default) |
+| | Nested struct source-extra fields | Throws | Evolved into the target schema |
+| | Nested struct target-missing fields | Throws | `UPDATE *` preserves current value; `INSERT *` fills `CURRENT_DEFAULT` (or NULL when no default) |
+
+Notes:
+- Position-based writes (e.g. `INSERT INTO t VALUES (...)` without `BY NAME`) require an exact column count match and don't engage schema evolution; only byName writes are covered above.
+- Top-level target-missing under `merge-schema=false` for byName `INSERT` mirrors Spark's `INSERT FILL` semantics — only nested missing fields throw.
+- Under strict mode (`merge-schema=false`), nested source-extra fields throw to avoid silent data loss; for `MERGE INTO *` at the top level, source-extras are silently dropped because `*` never references them.
+
## COPY INTO
`COPY INTO` provides a SQL command for bulk loading data files into Paimon tables and exporting table data to files. Supported formats: **CSV**, **JSON**, and **Parquet**.
-### CSV Import
+#### CSV Import
```sql
COPY INTO table_name [(col1, col2, ...)]
@@ -354,7 +363,7 @@ PATTERN = '.*\.csv'
FORCE = FALSE;
```
-### JSON Import
+#### JSON Import
```sql
COPY INTO table_name [(col1, col2, ...)]
@@ -383,7 +392,7 @@ FILE_FORMAT = (TYPE = JSON, MULTI_LINE = TRUE);
JSON columns are matched **by column name** (not by position), so source field order does not matter.
-### Parquet Import
+#### Parquet Import
```sql
COPY INTO table_name [(col1, col2, ...)]
@@ -414,7 +423,7 @@ FORCE = FALSE;
Parquet columns are matched **by column name** (not by position). Extra columns in the source files are ignored; missing columns become NULL.
-### Write CSV Files
+#### Write CSV Files
```sql
COPY INTO 'target_path'
@@ -432,7 +441,7 @@ FILE_FORMAT = (TYPE = CSV, HEADER = TRUE, FIELD_DELIMITER = ',')
OVERWRITE = TRUE;
```
-### Write JSON Files
+#### Write JSON Files
```sql
COPY INTO 'target_path'
@@ -450,7 +459,7 @@ FILE_FORMAT = (TYPE = JSON)
OVERWRITE = TRUE;
```
-### Write Parquet Files
+#### Write Parquet Files
```sql
COPY INTO 'target_path'
@@ -477,7 +486,7 @@ FILE_FORMAT = (TYPE = PARQUET, COMPRESSION = GZIP)
OVERWRITE = TRUE;
```
-### FILE_FORMAT Options
+#### FILE_FORMAT Options
`FILE_FORMAT` is required and must include `TYPE = CSV`, `TYPE = JSON`, or `TYPE = PARQUET`.
@@ -538,7 +547,7 @@ OVERWRITE = TRUE;
| TYPE | File format type. `CSV`, `JSON`, or `PARQUET`. | (required) |
| COMPRESSION | Compression codec (`SNAPPY`, `GZIP`, `NONE`, etc.). | `SNAPPY` |
-### Import Options
+#### Import Options
| Option | Description | Default |
|--------|-------------|---------|
@@ -546,13 +555,13 @@ OVERWRITE = TRUE;
| FORCE | `FALSE`: skip files already loaded (idempotent). `TRUE`: reload all files. | `FALSE` |
| ON_ERROR | Error handling strategy. Only `ABORT_STATEMENT` is supported. | `ABORT_STATEMENT` |
-### File Write Options
+#### File Write Options
| Option | Description | Default |
|--------|-------------|---------|
| OVERWRITE | `FALSE`: fail if target path exists. `TRUE`: overwrite existing files. | `FALSE` |
-### Column Mapping
+#### Column Mapping
When an explicit column list is provided (e.g., `COPY INTO t (col1, col2) FROM ...`):
@@ -568,7 +577,7 @@ When no column list is provided:
- **CSV**: Columns are mapped positionally to all writable columns in the target table. The number of CSV columns must match the number of writable columns.
- **JSON**: Columns are matched by name to the writable columns. Missing fields in JSON become NULL.
-### Repeated Imports
+#### Repeated Imports
By default (`FORCE = FALSE`), COPY INTO tracks which files have been successfully loaded. A file is identified by its path, size, and last-modified timestamp.
@@ -576,7 +585,7 @@ By default (`FORCE = FALSE`), COPY INTO tracks which files have been successfull
- If a source file is modified (size or timestamp changes), it becomes eligible for re-loading.
- `FORCE = TRUE` bypasses load history and always re-imports all matching files.
-### Result Output
+#### Result Output
**Import** returns one row per source file:
@@ -595,7 +604,7 @@ By default (`FORCE = FALSE`), COPY INTO tracks which files have been successfull
| file_count | INT | Number of files written |
| rows_written | BIGINT | Total rows written |
-### Limitations
+#### Limitations
- Only **CSV**, **JSON**, and **Parquet** formats are supported.
- Writing files only supports `FROM table_name`; `FROM (SELECT ...)` is not supported.
diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html
index a6c8278f4c..0dbc8b830a 100644
--- a/docs/generated/spark_connector_configuration.html
+++ b/docs/generated/spark_connector_configuration.html
@@ -84,13 +84,19 @@ under the License.
<td><h5>write.merge-schema</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>If true, merge the data schema and the table schema automatically before write data.</td>
+ <td>If true, evolve the table schema to accept new columns from the incoming data. Existing column types are preserved and incoming values are cast to them; to also widen existing types, enable 'write.merge-schema.type-widening'.</td>
</tr>
<tr>
<td><h5>write.merge-schema.explicit-cast</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>If true, allow to merge data types if the two types meet the rules for explicit casting.</td>
+ <td>Only effective when 'write.merge-schema.type-widening' is true. If true, also allow lossy type changes between compatible types (e.g. BIGINT -> INT, STRING -> DATE).</td>
+ </tr>
+ <tr>
+ <td><h5>write.merge-schema.type-widening</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Only effective when 'write.merge-schema' is true. If true, widen an existing column type when the incoming data has a wider compatible type (e.g. INT -> BIGINT, DECIMAL precision increase). Lossy changes are still rejected unless 'write.merge-schema.explicit-cast' is also true.</td>
</tr>
<tr>
<td><h5>write.use-v2-write</h5></td>
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index e8153aee52..3aa399186e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -271,9 +271,17 @@ abstract class AbstractFileStore<T> implements FileStore<T> {
}
@Override
- public boolean mergeSchema(RowType rowType, boolean allowExplicitCast, boolean caseSensitive) {
+ public boolean mergeSchema(
+ RowType rowType,
+ boolean typeWidening,
+ boolean allowExplicitCast,
+ boolean caseSensitive) {
return schemaManager.mergeSchema(
- rowType, allowExplicitCast, caseSensitive, catalogEnvironment.schemaModification());
+ rowType,
+ typeWidening,
+ allowExplicitCast,
+ caseSensitive,
+ catalogEnvironment.schemaModification());
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/FileStore.java b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
index 7a917adde8..1714eec2a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/FileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/FileStore.java
@@ -117,7 +117,11 @@ public interface FileStore<T> {
ServiceManager newServiceManager();
- boolean mergeSchema(RowType rowType, boolean allowExplicitCast, boolean caseSensitive);
+ boolean mergeSchema(
+ RowType rowType,
+ boolean typeWidening,
+ boolean allowExplicitCast,
+ boolean caseSensitive);
List<TagCallback> createTagCallbacks(FileStoreTable table);
diff --git a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java
index f4e6922b98..a1f6d5fbb2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStore.java
@@ -219,9 +219,13 @@ public class PrivilegedFileStore<T> implements FileStore<T> {
}
@Override
- public boolean mergeSchema(RowType rowType, boolean allowExplicitCast, boolean caseSensitive) {
+ public boolean mergeSchema(
+ RowType rowType,
+ boolean typeWidening,
+ boolean allowExplicitCast,
+ boolean caseSensitive) {
privilegeChecker.assertCanInsert(identifier);
- return wrapped.mergeSchema(rowType, allowExplicitCast, caseSensitive);
+ return wrapped.mergeSchema(rowType, typeWidening, allowExplicitCast, caseSensitive);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 8a4128e58d..95c6f04514 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -750,8 +750,15 @@ public class SchemaManager implements Serializable {
}
}
+ /**
+ * Merge {@code rowType} into the current schema (via {@link SchemaMergingUtils#mergeSchemas})
+ * and persist the result. Returns {@code true} if the schema changed and was committed, {@code
+ * false} if the merge was a no-op. See {@code SchemaMergingUtils} for how {@code typeWidening}
+ * / {@code allowExplicitCast} drive existing-column type evolution.
+ */
public boolean mergeSchema(
RowType rowType,
+ boolean typeWidening,
boolean allowExplicitCast,
boolean caseSensitive,
@Nullable SchemaModification schemaModification) {
@@ -761,7 +768,8 @@ public class SchemaManager implements Serializable {
new RuntimeException(
"It requires that the current schema to exist when calling 'mergeSchema'"));
TableSchema update =
- SchemaMergingUtils.mergeSchemas(current, rowType, allowExplicitCast, caseSensitive);
+ SchemaMergingUtils.mergeSchemas(
+ current, rowType, typeWidening, allowExplicitCast, caseSensitive);
if (current.equals(update)) {
return false;
}
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 d007b96db0..d444d89ca4 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
@@ -45,6 +45,7 @@ public class SchemaMergingUtils {
public static TableSchema mergeSchemas(
TableSchema currentTableSchema,
RowType targetType,
+ boolean typeWidening,
boolean allowExplicitCast,
boolean caseSensitive) {
RowType currentType = currentTableSchema.logicalRowType();
@@ -55,7 +56,12 @@ public class SchemaMergingUtils {
AtomicInteger highestFieldId = new AtomicInteger(currentTableSchema.highestFieldId());
RowType newRowType =
mergeSchemas(
- currentType, targetType, highestFieldId, allowExplicitCast, caseSensitive);
+ currentType,
+ targetType,
+ highestFieldId,
+ typeWidening,
+ allowExplicitCast,
+ caseSensitive);
if (newRowType.equals(currentType)) {
// It happens if the `targetType` only changes `nullability` but we always respect the
// current's.
@@ -76,32 +82,40 @@ public class SchemaMergingUtils {
RowType tableSchema,
RowType dataSchema,
AtomicInteger highestFieldId,
+ boolean typeWidening,
boolean allowExplicitCast,
boolean caseSensitive) {
return (RowType)
- merge(tableSchema, dataSchema, highestFieldId, allowExplicitCast, caseSensitive);
+ merge(
+ tableSchema,
+ dataSchema,
+ highestFieldId,
+ typeWidening,
+ allowExplicitCast,
+ caseSensitive);
}
/**
- * Merge the base data type and the update data type if possible.
+ * Merge the base (target) data type with the update (incoming) data type.
*
- * <p>For RowType, find the fields which exists in both the base schema and the update schema,
- * and try to merge them by calling the method iteratively; remain those fields that are only in
- * the base schema and append those fields that are only in the update schema.
- *
- * <p>For other complex type, try to merge the element types.
- *
- * <p>For primitive data type, we treat that's compatible if the original type can be safely
- * cast to the new type.
+ * <ul>
+ * <li>RowType: merge existing fields recursively, keep base-only fields, append update-only
+ * fields as new columns.
+ * <li>Complex types (Array/Map/Multiset): recursively merge element/value types.
+ * <li>Leaf types when {@code typeWidening=false} (default): keep the base type unchanged —
+ * incoming data is cast to it by the alignment layer.
+ * <li>Leaf types when {@code typeWidening=true}: widen the base type to the update type if
+ * the cast is safe (or explicit when {@code allowExplicitCast=true}).
+ * </ul>
*/
public static DataType merge(
DataType base0,
DataType update0,
AtomicInteger highestFieldId,
+ boolean typeWidening,
boolean allowExplicitCast,
boolean caseSensitive) {
- // Here we try to merge the base0 and update0 without regard to the nullability,
- // and set the base0's nullability to the return's.
+ // Compare ignoring nullability; the base's nullability flows to the result.
DataType base = base0.copy(true);
DataType update = update0.copy(true);
@@ -120,6 +134,7 @@ public class SchemaMergingUtils {
baseField.type(),
updateField.type(),
highestFieldId,
+ typeWidening,
allowExplicitCast,
caseSensitive);
updatedFields.add(
@@ -149,12 +164,14 @@ public class SchemaMergingUtils {
((MapType) base).getKeyType(),
((MapType) update).getKeyType(),
highestFieldId,
+ typeWidening,
allowExplicitCast,
caseSensitive),
merge(
((MapType) base).getValueType(),
((MapType) update).getValueType(),
highestFieldId,
+ typeWidening,
allowExplicitCast,
caseSensitive));
} else if (base instanceof ArrayType && update instanceof ArrayType) {
@@ -164,6 +181,7 @@ public class SchemaMergingUtils {
((ArrayType) base).getElementType(),
((ArrayType) update).getElementType(),
highestFieldId,
+ typeWidening,
allowExplicitCast,
caseSensitive));
} else if (base instanceof MultisetType && update instanceof MultisetType) {
@@ -173,8 +191,13 @@ public class SchemaMergingUtils {
((MultisetType) base).getElementType(),
((MultisetType) update).getElementType(),
highestFieldId,
+ typeWidening,
allowExplicitCast,
caseSensitive));
+ } else if (!typeWidening) {
+ // Default: keep the existing leaf type — only column additions evolve the schema.
+ // Incoming values are cast to this type by the alignment layer.
+ return base0;
} else if (base instanceof DecimalType && update instanceof DecimalType) {
if (((DecimalType) base).getScale() == ((DecimalType) update).getScale()) {
return new DecimalType(
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 496242db70..e13b11d474 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -807,7 +807,7 @@ public class FileStoreCommitTest {
ArrayList<DataField> newFields =
new ArrayList<>(TestKeyValueGenerator.DEFAULT_ROW_TYPE.getFields());
newFields.add(new DataField(-1, "newField", DataTypes.INT()));
- store.mergeSchema(new RowType(false, newFields), true, true);
+ store.mergeSchema(new RowType(false, newFields), true, true, true);
store.commitData(generateDataList(10), gen::getPartition, kv -> 0);
readStats = statsFileHandler.readStats();
assertThat(readStats).isEmpty();
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
index 050a4f12d7..47fc100032 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaMergingUtilsTest.java
@@ -83,7 +83,7 @@ public class SchemaMergingUtilsTest {
DataField f = new DataField(-1, "f", fDataType);
RowType t = new RowType(Lists.newArrayList(a, b, d, f));
- TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, false, true);
+ TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, true, false, true);
assertThat(merged.id()).isEqualTo(1);
assertThat(merged.highestFieldId()).isEqualTo(6);
assertThat(merged.primaryKeys()).containsExactlyInAnyOrder("a", "d");
@@ -110,10 +110,40 @@ public class SchemaMergingUtilsTest {
// fake the RowType of data with different field sequences
RowType t = new RowType(Lists.newArrayList(b, a));
- TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, false, true);
+ TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, true, false, true);
assertThat(merged.id()).isEqualTo(0);
}
+ @Test
+ public void testMergeSchemasWithoutTypeWidening() {
+ // typeWidening=false (default): existing column types are kept; only new columns are added.
+ DataField a = new DataField(0, "a", new IntType());
+ DataField b = new DataField(1, "b", new VarCharType(VarCharType.MAX_LENGTH));
+ TableSchema current =
+ new TableSchema(
+ 0,
+ Lists.newArrayList(a, b),
+ 1,
+ new ArrayList<>(),
+ Lists.newArrayList("a"),
+ new HashMap<>(),
+ "");
+
+ // Incoming data widens `a` (INT -> BIGINT) and adds a new column `c`.
+ DataField aWidened = new DataField(-1, "a", new BigIntType());
+ DataField c = new DataField(-1, "c", new IntType());
+ RowType t = new RowType(Lists.newArrayList(aWidened, b, c));
+
+ TableSchema merged = SchemaMergingUtils.mergeSchemas(current, t, false, false, true);
+ List<DataField> fields = merged.fields();
+ assertThat(fields.size()).isEqualTo(3);
+ // `a` keeps its existing INT type instead of widening to BIGINT.
+ assertThat(fields.get(0).type()).isEqualTo(new IntType());
+ // `c` is appended as a new column.
+ assertThat(fields.get(2).name()).isEqualTo("c");
+ assertThat(fields.get(2).type()).isEqualTo(new IntType());
+ }
+
@Test
public void testMergeSchemas() {
// This will test both `mergeSchemas` and `merge` methods.
@@ -132,7 +162,8 @@ public class SchemaMergingUtilsTest {
// Case 1: an additional field.
DataField e = new DataField(-1, "e", new DateType());
RowType t1 = new RowType(Lists.newArrayList(a, b, c, d, e));
- RowType r1 = (RowType) SchemaMergingUtils.merge(source, t1, highestFieldId, false, true);
+ RowType r1 =
+ (RowType) SchemaMergingUtils.merge(source, t1, highestFieldId, true, false, true);
assertThat(highestFieldId.get()).isEqualTo(4);
assertThat(r1.isNullable()).isTrue();
assertThat(r1.getFieldCount()).isEqualTo(5);
@@ -143,7 +174,7 @@ public class SchemaMergingUtilsTest {
// Case 2: two missing fields.
RowType t2 = new RowType(Lists.newArrayList(a, c, e));
- RowType r2 = SchemaMergingUtils.mergeSchemas(r1, t2, highestFieldId, false, true);
+ RowType r2 = SchemaMergingUtils.mergeSchemas(r1, t2, highestFieldId, true, false, true);
assertThat(highestFieldId.get()).isEqualTo(4);
assertThat(r2.getFieldCount()).isEqualTo(5);
assertThat(r2.getTypeAt(3)).isEqualTo(d.type());
@@ -155,7 +186,7 @@ public class SchemaMergingUtilsTest {
RowType fDataType = new RowType(Lists.newArrayList(f1, f2));
DataField f = new DataField(-1, "f", fDataType);
RowType t3 = new RowType(Lists.newArrayList(a, b, c, d, f));
- RowType r3 = (RowType) SchemaMergingUtils.merge(r2, t3, highestFieldId, false, true);
+ RowType r3 = (RowType) SchemaMergingUtils.merge(r2, t3, highestFieldId, true, false, true);
assertThat(highestFieldId.get()).isEqualTo(7);
assertThat(r3.getFieldCount()).isEqualTo(6);
RowType expectedFDataType = new RowType(Lists.newArrayList(f1.newId(5), f2.newId(6)));
@@ -168,7 +199,7 @@ public class SchemaMergingUtilsTest {
RowType newFDataType = new RowType(Lists.newArrayList(f1, f2, f3));
DataField newF = new DataField(-1, "f", newFDataType);
RowType t4 = new RowType(Lists.newArrayList(a, b, c, d, e, newF));
- RowType r4 = SchemaMergingUtils.mergeSchemas(r3, t4, highestFieldId, false, true);
+ RowType r4 = SchemaMergingUtils.mergeSchemas(r3, t4, highestFieldId, true, false, true);
assertThat(highestFieldId.get()).isEqualTo(8);
assertThat(r4.getFieldCount()).isEqualTo(6);
RowType newExpectedFDataType =
@@ -178,14 +209,15 @@ public class SchemaMergingUtilsTest {
// Case 5: a field that isn't compatible with the existing one.
DataField newA = new DataField(-1, "a", new SmallIntType());
RowType t5 = new RowType(Lists.newArrayList(newA, b, c, d, e, newF));
- assertThatThrownBy(() -> SchemaMergingUtils.merge(r4, t5, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(r4, t5, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// Case 6: all new-coming fields
DataField g = new DataField(-1, "g", new TimeType());
DataField h = new DataField(-1, "h", new TimeType());
RowType t6 = new RowType(Lists.newArrayList(g, h));
- RowType r6 = SchemaMergingUtils.mergeSchemas(r4, t6, highestFieldId, false, true);
+ RowType r6 = SchemaMergingUtils.mergeSchemas(r4, t6, highestFieldId, true, false, true);
assertThat(highestFieldId.get()).isEqualTo(10);
assertThat(r6.getFieldCount()).isEqualTo(8);
}
@@ -381,24 +413,28 @@ public class SchemaMergingUtilsTest {
// the element types are same.
DataType t1 = new ArrayType(true, new IntType());
ArrayType r1 =
- (ArrayType) SchemaMergingUtils.merge(source, t1, highestFieldId, false, true);
+ (ArrayType) SchemaMergingUtils.merge(source, t1, highestFieldId, true, false, true);
assertThat(r1.isNullable()).isFalse();
assertThat(r1.getElementType() instanceof IntType).isTrue();
// the element types aren't same, but can be evolved safety.
DataType t2 = new ArrayType(true, new BigIntType());
ArrayType r2 =
- (ArrayType) SchemaMergingUtils.merge(source, t2, highestFieldId, false, true);
+ (ArrayType) SchemaMergingUtils.merge(source, t2, highestFieldId, true, false, true);
assertThat(r2.isNullable()).isFalse();
assertThat(r2.getElementType() instanceof BigIntType).isTrue();
// the element types aren't same, and can't be evolved safety.
DataType t3 = new ArrayType(true, new SmallIntType());
- assertThatThrownBy(() -> SchemaMergingUtils.merge(source, t3, highestFieldId, false, true))
+ assertThatThrownBy(
+ () ->
+ SchemaMergingUtils.merge(
+ source, t3, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// the value type of target's isn't same to the source's, but the source type can be cast to
// the target type explicitly.
- ArrayType r3 = (ArrayType) SchemaMergingUtils.merge(source, t3, highestFieldId, true, true);
+ ArrayType r3 =
+ (ArrayType) SchemaMergingUtils.merge(source, t3, highestFieldId, true, true, true);
assertThat(r3.isNullable()).isFalse();
assertThat(r3.getElementType() instanceof SmallIntType).isTrue();
}
@@ -411,25 +447,31 @@ public class SchemaMergingUtilsTest {
// both the key and value types are same to the source's.
DataType t1 = new MapType(new VarCharType(VarCharType.MAX_LENGTH), new IntType());
- MapType r1 = (MapType) SchemaMergingUtils.merge(source, t1, highestFieldId, false, true);
+ MapType r1 =
+ (MapType) SchemaMergingUtils.merge(source, t1, highestFieldId, true, false, true);
assertThat(r1.isNullable()).isTrue();
assertThat(r1.getKeyType() instanceof VarCharType).isTrue();
assertThat(r1.getValueType() instanceof IntType).isTrue();
// the value type of target's isn't same to the source's, but can be evolved safety.
DataType t2 = new MapType(new VarCharType(VarCharType.MAX_LENGTH), new DoubleType());
- MapType r2 = (MapType) SchemaMergingUtils.merge(source, t2, highestFieldId, false, true);
+ MapType r2 =
+ (MapType) SchemaMergingUtils.merge(source, t2, highestFieldId, true, false, true);
assertThat(r2.isNullable()).isTrue();
assertThat(r2.getKeyType() instanceof VarCharType).isTrue();
assertThat(r2.getValueType() instanceof DoubleType).isTrue();
// the value type of target's isn't same to the source's, and can't be evolved safety.
DataType t3 = new MapType(new VarCharType(VarCharType.MAX_LENGTH), new SmallIntType());
- assertThatThrownBy(() -> SchemaMergingUtils.merge(source, t3, highestFieldId, false, true))
+ assertThatThrownBy(
+ () ->
+ SchemaMergingUtils.merge(
+ source, t3, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// the value type of target's isn't same to the source's, but the source type can be cast to
// the target type explicitly.
- MapType r3 = (MapType) SchemaMergingUtils.merge(source, t3, highestFieldId, true, true);
+ MapType r3 =
+ (MapType) SchemaMergingUtils.merge(source, t3, highestFieldId, true, true, true);
assertThat(r3.isNullable()).isTrue();
assertThat(r3.getKeyType() instanceof VarCharType).isTrue();
assertThat(r3.getValueType() instanceof SmallIntType).isTrue();
@@ -444,25 +486,31 @@ public class SchemaMergingUtilsTest {
// the element types are same.
DataType t1 = new MultisetType(true, new IntType());
MultisetType r1 =
- (MultisetType) SchemaMergingUtils.merge(source, t1, highestFieldId, false, true);
+ (MultisetType)
+ SchemaMergingUtils.merge(source, t1, highestFieldId, true, false, true);
assertThat(r1.isNullable()).isFalse();
assertThat(r1.getElementType() instanceof IntType).isTrue();
// the element types aren't same, but can be evolved safety.
DataType t2 = new MultisetType(true, new BigIntType());
MultisetType r2 =
- (MultisetType) SchemaMergingUtils.merge(source, t2, highestFieldId, false, true);
+ (MultisetType)
+ SchemaMergingUtils.merge(source, t2, highestFieldId, true, false, true);
assertThat(r2.isNullable()).isFalse();
assertThat(r2.getElementType() instanceof BigIntType).isTrue();
// the element types aren't same, and can't be evolved safety.
DataType t3 = new MultisetType(true, new SmallIntType());
- assertThatThrownBy(() -> SchemaMergingUtils.merge(source, t3, highestFieldId, false, true))
+ assertThatThrownBy(
+ () ->
+ SchemaMergingUtils.merge(
+ source, t3, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// the value type of target's isn't same to the source's, but the source type can be cast to
// the target type explicitly.
MultisetType r3 =
- (MultisetType) SchemaMergingUtils.merge(source, t3, highestFieldId, true, true);
+ (MultisetType)
+ SchemaMergingUtils.merge(source, t3, highestFieldId, true, true, true);
assertThat(r3.isNullable()).isFalse();
assertThat(r3.getElementType() instanceof SmallIntType).isTrue();
}
@@ -474,20 +522,21 @@ public class SchemaMergingUtilsTest {
DataType s1 = new DecimalType();
DataType t1 = new DecimalType(10, 0);
DecimalType r1 =
- (DecimalType) SchemaMergingUtils.merge(s1, t1, highestFieldId, false, true);
+ (DecimalType) SchemaMergingUtils.merge(s1, t1, highestFieldId, true, false, true);
assertThat(r1.isNullable()).isTrue();
assertThat(r1.getPrecision()).isEqualTo(DecimalType.DEFAULT_PRECISION);
assertThat(r1.getScale()).isEqualTo(DecimalType.DEFAULT_SCALE);
DataType s2 = new DecimalType(5, 2);
DataType t2 = new DecimalType(7, 3);
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s2, t2, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s2, t2, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
DataType s3 = new DecimalType(false, 5, 2);
DataType t3 = new DecimalType(7, 2);
DecimalType r3 =
- (DecimalType) SchemaMergingUtils.merge(s3, t3, highestFieldId, false, true);
+ (DecimalType) SchemaMergingUtils.merge(s3, t3, highestFieldId, true, false, true);
assertThat(r3.isNullable()).isFalse();
assertThat(r3.getPrecision()).isEqualTo(7);
assertThat(r3.getScale()).isEqualTo(2);
@@ -495,7 +544,7 @@ public class SchemaMergingUtilsTest {
DataType s4 = new DecimalType(7, 2);
DataType t4 = new DecimalType(5, 2);
DecimalType r4 =
- (DecimalType) SchemaMergingUtils.merge(s4, t4, highestFieldId, false, true);
+ (DecimalType) SchemaMergingUtils.merge(s4, t4, highestFieldId, true, false, true);
assertThat(r4.isNullable()).isTrue();
assertThat(r4.getPrecision()).isEqualTo(7);
assertThat(r4.getScale()).isEqualTo(2);
@@ -506,10 +555,11 @@ public class SchemaMergingUtilsTest {
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dcmSource, iTarget, highestFieldId, false, true))
+ dcmSource, iTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DecimalType -> Other Numeric Type with allowExplicitCast = true
- DataType res = SchemaMergingUtils.merge(dcmSource, iTarget, highestFieldId, true, true);
+ DataType res =
+ SchemaMergingUtils.merge(dcmSource, iTarget, highestFieldId, true, true, true);
assertThat(res instanceof IntType).isTrue();
}
@@ -520,122 +570,141 @@ public class SchemaMergingUtilsTest {
// BinaryType
DataType s1 = new BinaryType(10);
DataType t1 = new BinaryType(10);
- BinaryType r1 = (BinaryType) SchemaMergingUtils.merge(s1, t1, highestFieldId, false, true);
+ BinaryType r1 =
+ (BinaryType) SchemaMergingUtils.merge(s1, t1, highestFieldId, true, false, true);
assertThat(r1.getLength()).isEqualTo(10);
DataType s2 = new BinaryType(2);
DataType t2 = new BinaryType();
// smaller length
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s2, t2, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s2, t2, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// smaller length with allowExplicitCast = true
- BinaryType r2 = (BinaryType) SchemaMergingUtils.merge(s2, t2, highestFieldId, true, true);
+ BinaryType r2 =
+ (BinaryType) SchemaMergingUtils.merge(s2, t2, highestFieldId, true, true, true);
assertThat(r2.getLength()).isEqualTo(BinaryType.DEFAULT_LENGTH);
// bigger length
DataType t3 = new BinaryType(5);
- BinaryType r3 = (BinaryType) SchemaMergingUtils.merge(s2, t3, highestFieldId, false, true);
+ BinaryType r3 =
+ (BinaryType) SchemaMergingUtils.merge(s2, t3, highestFieldId, true, false, true);
assertThat(r3.getLength()).isEqualTo(5);
// VarCharType
DataType s4 = new VarCharType();
DataType t4 = new VarCharType(1);
VarCharType r4 =
- (VarCharType) SchemaMergingUtils.merge(s4, t4, highestFieldId, false, true);
+ (VarCharType) SchemaMergingUtils.merge(s4, t4, highestFieldId, true, false, true);
assertThat(r4.getLength()).isEqualTo(VarCharType.DEFAULT_LENGTH);
DataType s5 = new VarCharType(2);
DataType t5 = new VarCharType();
// smaller length
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s5, t5, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s5, t5, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// smaller length with allowExplicitCast = true
- VarCharType r5 = (VarCharType) SchemaMergingUtils.merge(s5, t5, highestFieldId, true, true);
+ VarCharType r5 =
+ (VarCharType) SchemaMergingUtils.merge(s5, t5, highestFieldId, true, true, true);
assertThat(r5.getLength()).isEqualTo(VarCharType.DEFAULT_LENGTH);
// bigger length
DataType t6 = new VarCharType(5);
VarCharType r6 =
- (VarCharType) SchemaMergingUtils.merge(s5, t6, highestFieldId, false, true);
+ (VarCharType) SchemaMergingUtils.merge(s5, t6, highestFieldId, true, false, true);
assertThat(r6.getLength()).isEqualTo(5);
// CharType
DataType s7 = new CharType();
DataType t7 = new CharType(1);
- CharType r7 = (CharType) SchemaMergingUtils.merge(s7, t7, highestFieldId, false, true);
+ CharType r7 =
+ (CharType) SchemaMergingUtils.merge(s7, t7, highestFieldId, true, false, true);
assertThat(r7.getLength()).isEqualTo(CharType.DEFAULT_LENGTH);
DataType s8 = new CharType(2);
DataType t8 = new CharType();
// smaller length
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s8, t8, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s8, t8, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// smaller length with allowExplicitCast = true
- CharType r8 = (CharType) SchemaMergingUtils.merge(s8, t8, highestFieldId, true, true);
+ CharType r8 = (CharType) SchemaMergingUtils.merge(s8, t8, highestFieldId, true, true, true);
assertThat(r8.getLength()).isEqualTo(CharType.DEFAULT_LENGTH);
// bigger length
DataType t9 = new CharType(5);
- CharType r9 = (CharType) SchemaMergingUtils.merge(s8, t9, highestFieldId, false, true);
+ CharType r9 =
+ (CharType) SchemaMergingUtils.merge(s8, t9, highestFieldId, true, false, true);
assertThat(r9.getLength()).isEqualTo(5);
// VarBinaryType
DataType s10 = new VarBinaryType();
DataType t10 = new VarBinaryType(1);
VarBinaryType r10 =
- (VarBinaryType) SchemaMergingUtils.merge(s10, t10, highestFieldId, false, true);
+ (VarBinaryType)
+ SchemaMergingUtils.merge(s10, t10, highestFieldId, true, false, true);
assertThat(r10.getLength()).isEqualTo(VarBinaryType.DEFAULT_LENGTH);
DataType s11 = new VarBinaryType(2);
DataType t11 = new VarBinaryType();
// smaller length
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s11, t11, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s11, t11, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// smaller length with allowExplicitCast = true
VarBinaryType r11 =
- (VarBinaryType) SchemaMergingUtils.merge(s11, t11, highestFieldId, true, true);
+ (VarBinaryType)
+ SchemaMergingUtils.merge(s11, t11, highestFieldId, true, true, true);
assertThat(r11.getLength()).isEqualTo(VarBinaryType.DEFAULT_LENGTH);
// bigger length
DataType t12 = new VarBinaryType(5);
VarBinaryType r12 =
- (VarBinaryType) SchemaMergingUtils.merge(s11, t12, highestFieldId, false, true);
+ (VarBinaryType)
+ SchemaMergingUtils.merge(s11, t12, highestFieldId, true, false, true);
assertThat(r12.getLength()).isEqualTo(5);
// CharType -> VarCharType
DataType s13 = new CharType();
DataType t13 = new VarCharType(10);
VarCharType r13 =
- (VarCharType) SchemaMergingUtils.merge(s13, t13, highestFieldId, false, true);
+ (VarCharType) SchemaMergingUtils.merge(s13, t13, highestFieldId, true, false, true);
assertThat(r13.getLength()).isEqualTo(10);
// VarCharType ->CharType
DataType s14 = new VarCharType(10);
DataType t14 = new CharType();
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s14, t14, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s14, t14, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
- CharType r14 = (CharType) SchemaMergingUtils.merge(s14, t14, highestFieldId, true, true);
+ CharType r14 =
+ (CharType) SchemaMergingUtils.merge(s14, t14, highestFieldId, true, true, true);
assertThat(r14.getLength()).isEqualTo(CharType.DEFAULT_LENGTH);
// BinaryType -> VarBinaryType
DataType s15 = new BinaryType();
DataType t15 = new VarBinaryType(10);
VarBinaryType r15 =
- (VarBinaryType) SchemaMergingUtils.merge(s15, t15, highestFieldId, false, true);
+ (VarBinaryType)
+ SchemaMergingUtils.merge(s15, t15, highestFieldId, true, false, true);
assertThat(r15.getLength()).isEqualTo(10);
// VarBinaryType -> BinaryType
DataType s16 = new VarBinaryType(10);
DataType t16 = new BinaryType();
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s16, t16, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s16, t16, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
BinaryType r16 =
- (BinaryType) SchemaMergingUtils.merge(s16, t16, highestFieldId, true, true);
+ (BinaryType) SchemaMergingUtils.merge(s16, t16, highestFieldId, true, true, true);
assertThat(r16.getLength()).isEqualTo(BinaryType.DEFAULT_LENGTH);
// VarCharType -> VarBinaryType
DataType s17 = new VarCharType(10);
DataType t17 = new VarBinaryType();
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s17, t17, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s17, t17, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
VarBinaryType r17 =
- (VarBinaryType) SchemaMergingUtils.merge(s17, t17, highestFieldId, true, true);
+ (VarBinaryType)
+ SchemaMergingUtils.merge(s17, t17, highestFieldId, true, true, true);
assertThat(r17.getLength()).isEqualTo(VarBinaryType.DEFAULT_LENGTH);
}
@@ -648,7 +717,7 @@ public class SchemaMergingUtilsTest {
DataType t1 = new LocalZonedTimestampType();
LocalZonedTimestampType r1 =
(LocalZonedTimestampType)
- SchemaMergingUtils.merge(s1, t1, highestFieldId, false, true);
+ SchemaMergingUtils.merge(s1, t1, highestFieldId, true, false, true);
assertThat(r1.isNullable()).isTrue();
assertThat(r1.getPrecision()).isEqualTo(LocalZonedTimestampType.DEFAULT_PRECISION);
@@ -659,6 +728,7 @@ public class SchemaMergingUtilsTest {
s1,
new LocalZonedTimestampType(3),
highestFieldId,
+ true,
false,
true))
.isInstanceOf(UnsupportedOperationException.class);
@@ -667,27 +737,28 @@ public class SchemaMergingUtilsTest {
DataType t2 = new LocalZonedTimestampType(6);
LocalZonedTimestampType r2 =
(LocalZonedTimestampType)
- SchemaMergingUtils.merge(s1, t2, highestFieldId, false, true);
+ SchemaMergingUtils.merge(s1, t2, highestFieldId, true, false, true);
assertThat(r2.getPrecision()).isEqualTo(6);
// LocalZonedTimestampType -> TimeType
DataType s3 = new LocalZonedTimestampType();
DataType t3 = new TimeType(6);
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s3, t3, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s3, t3, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// LocalZonedTimestampType -> TimestampType
DataType s4 = new LocalZonedTimestampType();
DataType t4 = new TimestampType();
TimestampType r4 =
- (TimestampType) SchemaMergingUtils.merge(s4, t4, highestFieldId, false, true);
+ (TimestampType) SchemaMergingUtils.merge(s4, t4, highestFieldId, true, false, true);
assertThat(r4.getPrecision()).isEqualTo(TimestampType.DEFAULT_PRECISION);
// TimestampType.
DataType s5 = new TimestampType();
DataType t5 = new TimestampType();
TimestampType r5 =
- (TimestampType) SchemaMergingUtils.merge(s5, t5, highestFieldId, false, true);
+ (TimestampType) SchemaMergingUtils.merge(s5, t5, highestFieldId, true, false, true);
assertThat(r5.isNullable()).isTrue();
assertThat(r5.getPrecision()).isEqualTo(TimestampType.DEFAULT_PRECISION);
@@ -695,13 +766,18 @@ public class SchemaMergingUtilsTest {
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- s5, new TimestampType(3), highestFieldId, false, true))
+ s5,
+ new TimestampType(3),
+ highestFieldId,
+ true,
+ false,
+ true))
.isInstanceOf(UnsupportedOperationException.class);
// higher precision
DataType t6 = new TimestampType(9);
TimestampType r6 =
- (TimestampType) SchemaMergingUtils.merge(s5, t6, highestFieldId, false, true);
+ (TimestampType) SchemaMergingUtils.merge(s5, t6, highestFieldId, true, false, true);
assertThat(r6.getPrecision()).isEqualTo(9);
// TimestampType -> LocalZonedTimestampType
@@ -709,19 +785,21 @@ public class SchemaMergingUtilsTest {
DataType t7 = new LocalZonedTimestampType();
LocalZonedTimestampType r7 =
(LocalZonedTimestampType)
- SchemaMergingUtils.merge(s7, t7, highestFieldId, false, true);
+ SchemaMergingUtils.merge(s7, t7, highestFieldId, true, false, true);
assertThat(r7.getPrecision()).isEqualTo(TimestampType.DEFAULT_PRECISION);
// TimestampType -> TimestampType
DataType s8 = new TimestampType();
DataType t8 = new TimeType(6);
- TimeType r8 = (TimeType) SchemaMergingUtils.merge(s8, t8, highestFieldId, false, true);
+ TimeType r8 =
+ (TimeType) SchemaMergingUtils.merge(s8, t8, highestFieldId, true, false, true);
assertThat(r8.getPrecision()).isEqualTo(TimestampType.DEFAULT_PRECISION);
// TimeType.
DataType s9 = new TimeType();
DataType t9 = new TimeType();
- TimeType r9 = (TimeType) SchemaMergingUtils.merge(s9, t9, highestFieldId, false, true);
+ TimeType r9 =
+ (TimeType) SchemaMergingUtils.merge(s9, t9, highestFieldId, true, false, true);
assertThat(r9.isNullable()).isTrue();
assertThat(r9.getPrecision()).isEqualTo(TimeType.DEFAULT_PRECISION);
@@ -730,35 +808,39 @@ public class SchemaMergingUtilsTest {
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- s10, new TimeType(3), highestFieldId, false, true))
+ s10, new TimeType(3), highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// higher precision
DataType t10 = new TimeType(9);
- TimeType r10 = (TimeType) SchemaMergingUtils.merge(s9, t10, highestFieldId, false, true);
+ TimeType r10 =
+ (TimeType) SchemaMergingUtils.merge(s9, t10, highestFieldId, true, false, true);
assertThat(r10.getPrecision()).isEqualTo(9);
// TimeType -> LocalZonedTimestampType
DataType s11 = new TimeType();
DataType t11 = new LocalZonedTimestampType();
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s11, t11, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s11, t11, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// TimeType -> LocalZonedTimestampType with allowExplicitCast = true
LocalZonedTimestampType r11 =
(LocalZonedTimestampType)
- SchemaMergingUtils.merge(s11, t11, highestFieldId, true, true);
+ SchemaMergingUtils.merge(s11, t11, highestFieldId, true, true, true);
assertThat(r11.getPrecision()).isEqualTo(LocalZonedTimestampType.DEFAULT_PRECISION);
// TimeType -> TimestampType
DataType s12 = new TimeType();
DataType t12 = new TimestampType();
- assertThatThrownBy(() -> SchemaMergingUtils.merge(s12, t12, highestFieldId, false, true))
+ assertThatThrownBy(
+ () -> SchemaMergingUtils.merge(s12, t12, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// TimeType -> TimestampType with allowExplicitCast = true
TimestampType r12 =
- (TimestampType) SchemaMergingUtils.merge(s12, t12, highestFieldId, true, true);
+ (TimestampType)
+ SchemaMergingUtils.merge(s12, t12, highestFieldId, true, true, true);
assertThat(r12.getPrecision()).isEqualTo(TimestampType.DEFAULT_PRECISION);
}
@@ -784,237 +866,274 @@ public class SchemaMergingUtilsTest {
DataType dcmTarget = new DecimalType();
// BooleanType
- DataType btRes1 = SchemaMergingUtils.merge(bSource, bTarget, highestFieldId, false, true);
+ DataType btRes1 =
+ SchemaMergingUtils.merge(bSource, bTarget, highestFieldId, true, false, true);
assertThat(btRes1 instanceof BooleanType).isTrue();
// BooleanType -> Numeric Type
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- bSource, tiTarget, highestFieldId, false, true))
+ bSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// BooleanType -> Numeric Type with allowExplicitCast = true
- DataType btRes2 = SchemaMergingUtils.merge(bSource, tiTarget, highestFieldId, true, true);
+ DataType btRes2 =
+ SchemaMergingUtils.merge(bSource, tiTarget, highestFieldId, true, true, true);
assertThat(btRes2 instanceof TinyIntType).isTrue();
// TinyIntType
- DataType tiRes1 = SchemaMergingUtils.merge(tiSource, tiTarget, highestFieldId, false, true);
+ DataType tiRes1 =
+ SchemaMergingUtils.merge(tiSource, tiTarget, highestFieldId, true, false, true);
assertThat(tiRes1 instanceof TinyIntType).isTrue();
// TinyIntType -> SmallIntType
- DataType tiRes2 = SchemaMergingUtils.merge(tiSource, siTarget, highestFieldId, false, true);
+ DataType tiRes2 =
+ SchemaMergingUtils.merge(tiSource, siTarget, highestFieldId, true, false, true);
assertThat(tiRes2 instanceof SmallIntType).isTrue();
// TinyIntType -> IntType
- DataType tiRes3 = SchemaMergingUtils.merge(tiSource, iTarget, highestFieldId, false, true);
+ DataType tiRes3 =
+ SchemaMergingUtils.merge(tiSource, iTarget, highestFieldId, true, false, true);
assertThat(tiRes3 instanceof IntType).isTrue();
// TinyIntType -> BigIntType
- DataType tiRes4 = SchemaMergingUtils.merge(tiSource, biTarget, highestFieldId, false, true);
+ DataType tiRes4 =
+ SchemaMergingUtils.merge(tiSource, biTarget, highestFieldId, true, false, true);
assertThat(tiRes4 instanceof BigIntType).isTrue();
// TinyIntType -> FloatType
- DataType tiRes5 = SchemaMergingUtils.merge(tiSource, fTarget, highestFieldId, false, true);
+ DataType tiRes5 =
+ SchemaMergingUtils.merge(tiSource, fTarget, highestFieldId, true, false, true);
assertThat(tiRes5 instanceof FloatType).isTrue();
// TinyIntType -> DoubleType
- DataType tiRes6 = SchemaMergingUtils.merge(tiSource, dTarget, highestFieldId, false, true);
+ DataType tiRes6 =
+ SchemaMergingUtils.merge(tiSource, dTarget, highestFieldId, true, false, true);
assertThat(tiRes6 instanceof DoubleType).isTrue();
// TinyIntType -> DecimalType
DataType tiRes7 =
- SchemaMergingUtils.merge(tiSource, dcmTarget, highestFieldId, false, true);
+ SchemaMergingUtils.merge(tiSource, dcmTarget, highestFieldId, true, false, true);
assertThat(tiRes7 instanceof DecimalType).isTrue();
// TinyIntType -> BooleanType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- tiSource, bTarget, highestFieldId, false, true))
+ tiSource, bTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// TinyIntType -> BooleanType with allowExplicitCast = true
- DataType tiRes8 = SchemaMergingUtils.merge(tiSource, bTarget, highestFieldId, true, true);
+ DataType tiRes8 =
+ SchemaMergingUtils.merge(tiSource, bTarget, highestFieldId, true, true, true);
assertThat(tiRes8 instanceof BooleanType).isTrue();
// SmallIntType
- DataType siRes1 = SchemaMergingUtils.merge(siSource, siTarget, highestFieldId, false, true);
+ DataType siRes1 =
+ SchemaMergingUtils.merge(siSource, siTarget, highestFieldId, true, false, true);
assertThat(siRes1 instanceof SmallIntType).isTrue();
// SmallIntType -> TinyIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- siSource, tiTarget, highestFieldId, false, true))
+ siSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// SmallIntType -> TinyIntType with allowExplicitCast = true
- DataType siRes2 = SchemaMergingUtils.merge(siSource, tiTarget, highestFieldId, true, true);
+ DataType siRes2 =
+ SchemaMergingUtils.merge(siSource, tiTarget, highestFieldId, true, true, true);
assertThat(siRes2 instanceof TinyIntType).isTrue();
// SmallIntType -> IntType
- DataType siRes3 = SchemaMergingUtils.merge(siSource, iTarget, highestFieldId, false, true);
+ DataType siRes3 =
+ SchemaMergingUtils.merge(siSource, iTarget, highestFieldId, true, false, true);
assertThat(siRes3 instanceof IntType).isTrue();
// SmallIntType -> BigIntType
- DataType siRes4 = SchemaMergingUtils.merge(siSource, biTarget, highestFieldId, false, true);
+ DataType siRes4 =
+ SchemaMergingUtils.merge(siSource, biTarget, highestFieldId, true, false, true);
assertThat(siRes4 instanceof BigIntType).isTrue();
// SmallIntType -> FloatType
- DataType siRes5 = SchemaMergingUtils.merge(siSource, fTarget, highestFieldId, false, true);
+ DataType siRes5 =
+ SchemaMergingUtils.merge(siSource, fTarget, highestFieldId, true, false, true);
assertThat(siRes5 instanceof FloatType).isTrue();
// SmallIntType -> DoubleType
- DataType siRes6 = SchemaMergingUtils.merge(siSource, dTarget, highestFieldId, false, true);
+ DataType siRes6 =
+ SchemaMergingUtils.merge(siSource, dTarget, highestFieldId, true, false, true);
assertThat(siRes6 instanceof DoubleType).isTrue();
// SmallIntType -> DecimalType
DataType siRes7 =
- SchemaMergingUtils.merge(siSource, dcmTarget, highestFieldId, false, true);
+ SchemaMergingUtils.merge(siSource, dcmTarget, highestFieldId, true, false, true);
assertThat(siRes7 instanceof DecimalType).isTrue();
// IntType
- DataType iRes1 = SchemaMergingUtils.merge(iSource, iTarget, highestFieldId, false, true);
+ DataType iRes1 =
+ SchemaMergingUtils.merge(iSource, iTarget, highestFieldId, true, false, true);
assertThat(iRes1 instanceof IntType).isTrue();
// IntType -> TinyIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- iSource, tiTarget, highestFieldId, false, true))
+ iSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// IntType -> TinyIntType with allowExplicitCast = true
- DataType iRes2 = SchemaMergingUtils.merge(iSource, tiTarget, highestFieldId, true, true);
+ DataType iRes2 =
+ SchemaMergingUtils.merge(iSource, tiTarget, highestFieldId, true, true, true);
assertThat(iRes2 instanceof TinyIntType).isTrue();
// IntType -> SmallIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- iSource, siTarget, highestFieldId, false, true))
+ iSource, siTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// IntType -> SmallIntType with allowExplicitCast = true
- DataType iRes3 = SchemaMergingUtils.merge(iSource, siTarget, highestFieldId, true, true);
+ DataType iRes3 =
+ SchemaMergingUtils.merge(iSource, siTarget, highestFieldId, true, true, true);
assertThat(iRes3 instanceof SmallIntType).isTrue();
// IntType -> BigIntType
- DataType iRes4 = SchemaMergingUtils.merge(iSource, biTarget, highestFieldId, false, true);
+ DataType iRes4 =
+ SchemaMergingUtils.merge(iSource, biTarget, highestFieldId, true, false, true);
assertThat(iRes4 instanceof BigIntType).isTrue();
// IntType -> FloatType
- DataType iRes5 = SchemaMergingUtils.merge(iSource, fTarget, highestFieldId, false, true);
+ DataType iRes5 =
+ SchemaMergingUtils.merge(iSource, fTarget, highestFieldId, true, false, true);
assertThat(iRes5 instanceof FloatType).isTrue();
// IntType -> DoubleType
- DataType iRes6 = SchemaMergingUtils.merge(iSource, dTarget, highestFieldId, false, true);
+ DataType iRes6 =
+ SchemaMergingUtils.merge(iSource, dTarget, highestFieldId, true, false, true);
assertThat(iRes6 instanceof DoubleType).isTrue();
// IntType -> DecimalType
- DataType iRes7 = SchemaMergingUtils.merge(iSource, dcmTarget, highestFieldId, false, true);
+ DataType iRes7 =
+ SchemaMergingUtils.merge(iSource, dcmTarget, highestFieldId, true, false, true);
assertThat(iRes7 instanceof DecimalType).isTrue();
// BigIntType
- DataType biRes1 = SchemaMergingUtils.merge(biSource, biTarget, highestFieldId, false, true);
+ DataType biRes1 =
+ SchemaMergingUtils.merge(biSource, biTarget, highestFieldId, true, false, true);
assertThat(biRes1 instanceof BigIntType).isTrue();
// BigIntType -> TinyIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- biSource, tiTarget, highestFieldId, false, true))
+ biSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// BigIntType -> TinyIntType with allowExplicitCast = true
- DataType biRes2 = SchemaMergingUtils.merge(biSource, tiTarget, highestFieldId, true, true);
+ DataType biRes2 =
+ SchemaMergingUtils.merge(biSource, tiTarget, highestFieldId, true, true, true);
assertThat(biRes2 instanceof TinyIntType).isTrue();
// BigIntType -> SmallIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- biSource, siTarget, highestFieldId, false, true))
+ biSource, siTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// BigIntType -> SmallIntType with allowExplicitCast = true
- DataType biRes3 = SchemaMergingUtils.merge(biSource, siTarget, highestFieldId, true, true);
+ DataType biRes3 =
+ SchemaMergingUtils.merge(biSource, siTarget, highestFieldId, true, true, true);
assertThat(biRes3 instanceof SmallIntType).isTrue();
// BigIntType -> IntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- biSource, iTarget, highestFieldId, false, true))
+ biSource, iTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// BigIntType -> IntType with allowExplicitCast = true
- DataType biRes4 = SchemaMergingUtils.merge(biSource, iTarget, highestFieldId, true, true);
+ DataType biRes4 =
+ SchemaMergingUtils.merge(biSource, iTarget, highestFieldId, true, true, true);
assertThat(biRes4 instanceof IntType).isTrue();
// BigIntType -> FloatType
- DataType biRes5 = SchemaMergingUtils.merge(biSource, fTarget, highestFieldId, false, true);
+ DataType biRes5 =
+ SchemaMergingUtils.merge(biSource, fTarget, highestFieldId, true, false, true);
assertThat(biRes5 instanceof FloatType).isTrue();
// BigIntType -> DoubleType
- DataType biRes6 = SchemaMergingUtils.merge(biSource, dTarget, highestFieldId, false, true);
+ DataType biRes6 =
+ SchemaMergingUtils.merge(biSource, dTarget, highestFieldId, true, false, true);
assertThat(biRes6 instanceof DoubleType).isTrue();
// BigIntType -> DecimalType
DataType biRes7 =
- SchemaMergingUtils.merge(biSource, dcmTarget, highestFieldId, false, true);
+ SchemaMergingUtils.merge(biSource, dcmTarget, highestFieldId, true, false, true);
assertThat(biRes7 instanceof DecimalType).isTrue();
// FloatType
- DataType fRes1 = SchemaMergingUtils.merge(fSource, fTarget, highestFieldId, false, true);
+ DataType fRes1 =
+ SchemaMergingUtils.merge(fSource, fTarget, highestFieldId, true, false, true);
assertThat(fRes1 instanceof FloatType).isTrue();
// FloatType -> TinyIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- fSource, tiTarget, highestFieldId, false, true))
+ fSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// FloatType -> TinyIntType with allowExplicitCast = true
- DataType fRes2 = SchemaMergingUtils.merge(fSource, tiTarget, highestFieldId, true, true);
+ DataType fRes2 =
+ SchemaMergingUtils.merge(fSource, tiTarget, highestFieldId, true, true, true);
assertThat(fRes2 instanceof TinyIntType).isTrue();
// FloatType -> SmallIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- fSource, siTarget, highestFieldId, false, true))
+ fSource, siTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// FloatType -> IntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- fSource, iTarget, highestFieldId, false, true))
+ fSource, iTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// FloatType -> IntType with allowExplicitCast = true
- DataType fRes4 = SchemaMergingUtils.merge(fSource, iTarget, highestFieldId, true, true);
+ DataType fRes4 =
+ SchemaMergingUtils.merge(fSource, iTarget, highestFieldId, true, true, true);
assertThat(fRes4 instanceof IntType).isTrue();
// FloatType -> BigIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- fSource, biTarget, highestFieldId, false, true))
+ fSource, biTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// FloatType -> DoubleType
- DataType fRes6 = SchemaMergingUtils.merge(fSource, dTarget, highestFieldId, false, true);
+ DataType fRes6 =
+ SchemaMergingUtils.merge(fSource, dTarget, highestFieldId, true, false, true);
assertThat(fRes6 instanceof DoubleType).isTrue();
// FloatType -> DecimalType
- DataType fRes7 = SchemaMergingUtils.merge(fSource, dcmTarget, highestFieldId, false, true);
+ DataType fRes7 =
+ SchemaMergingUtils.merge(fSource, dcmTarget, highestFieldId, true, false, true);
assertThat(fRes7 instanceof DecimalType).isTrue();
// DoubleType
- DataType dRes1 = SchemaMergingUtils.merge(dSource, dTarget, highestFieldId, false, true);
+ DataType dRes1 =
+ SchemaMergingUtils.merge(dSource, dTarget, highestFieldId, true, false, true);
assertThat(dRes1 instanceof DoubleType).isTrue();
// DoubleType -> TinyIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dSource, tiTarget, highestFieldId, false, true))
+ dSource, tiTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DoubleType -> SmallIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dSource, siTarget, highestFieldId, false, true))
+ dSource, siTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DoubleType -> SmallIntType with allowExplicitCast = true
- DataType dRes3 = SchemaMergingUtils.merge(dSource, siTarget, highestFieldId, true, true);
+ DataType dRes3 =
+ SchemaMergingUtils.merge(dSource, siTarget, highestFieldId, true, true, true);
assertThat(dRes3 instanceof SmallIntType).isTrue();
// DoubleType -> IntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dSource, iTarget, highestFieldId, false, true))
+ dSource, iTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DoubleType -> BigIntType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dSource, biTarget, highestFieldId, false, true))
+ dSource, biTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DoubleType -> BigIntType with allowExplicitCast = true
- DataType dRes5 = SchemaMergingUtils.merge(dSource, biTarget, highestFieldId, true, true);
+ DataType dRes5 =
+ SchemaMergingUtils.merge(dSource, biTarget, highestFieldId, true, true, true);
assertThat(dRes5 instanceof BigIntType).isTrue();
// DoubleType -> FloatType
assertThatThrownBy(
() ->
SchemaMergingUtils.merge(
- dSource, fTarget, highestFieldId, false, true))
+ dSource, fTarget, highestFieldId, true, false, true))
.isInstanceOf(UnsupportedOperationException.class);
// DoubleType -> DecimalType
- DataType dRes7 = SchemaMergingUtils.merge(dSource, dcmTarget, highestFieldId, false, true);
+ DataType dRes7 =
+ SchemaMergingUtils.merge(dSource, dcmTarget, highestFieldId, true, false, true);
assertThat(dRes7 instanceof DecimalType).isTrue();
}
}
diff --git a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/DataFrameWriteTest.scala b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/DataFrameWriteTest.scala
index cb449edb4c..ac5c6c3a17 100644
--- a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/DataFrameWriteTest.scala
+++ b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/DataFrameWriteTest.scala
@@ -40,7 +40,6 @@ class DataFrameWriteTest extends PaimonSparkTestBase {
.option("bucket", "-1")
.option("target-file-size", "256MB")
.option("write.merge-schema", "true")
- .option("write.merge-schema.explicit-cast", "true")
.saveAsTable("test_ctas")
val paimonTable = loadTable("test_ctas")
@@ -53,7 +52,6 @@ class DataFrameWriteTest extends PaimonSparkTestBase {
// non-core options should not be here.
Assertions.assertFalse(paimonTable.options().containsKey("write.merge-schema"))
- Assertions.assertFalse(paimonTable.options().containsKey("write.merge-schema.explicit-cast"))
}
}
diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
index ab4a9bcd9d..d213f257c7 100644
--- a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
+++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
@@ -260,6 +260,7 @@ class PaimonSinkTest extends PaimonSparkTestBase with StreamTest {
.writeStream
.option("checkpointLocation", checkpointDir.getCanonicalPath)
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.option("write.merge-schema.explicit-cast", "true")
.format("paimon")
.start(location)
diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 492d64bbf5..c5f9e02715 100644
--- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -136,6 +136,8 @@ case class MergeIntoPaimonDataEvolutionTable(
lazy val tableSchema: StructType = v2Table.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
+ // Persist the schema that the analyzer evolved in memory (commit deferred to execution).
+ SchemaEvolutionHelper.commitEvolvedSchemaAtExecution(table, targetRelation, sparkSession)
invokeMergeInto(sparkSession)
Seq.empty[Row]
}
diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index 1822929854..89689e108c 100644
--- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -71,6 +71,8 @@ case class MergeIntoPaimonTable(
override def run(sparkSession: SparkSession): Seq[Row] = {
// Avoid that more than one source rows match the same target row.
checkMatchRationality(sparkSession)
+ // Persist the schema that the analyzer evolved in memory (commit deferred to execution).
+ SchemaEvolutionHelper.commitEvolvedSchemaAtExecution(table, relation, sparkSession)
val commitMessages = if (withPrimaryKeys) {
performMergeForPkTable(sparkSession)
} else {
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
index 13305637ee..5217ea0513 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java
@@ -37,14 +37,28 @@ public class SparkConnectorOptions {
.booleanType()
.defaultValue(false)
.withDescription(
- "If true, merge the data schema and the table schema automatically before write data.");
+ "If true, evolve the table schema to accept new columns from the incoming data. "
+ + "Existing column types are preserved and incoming values are cast to them; "
+ + "to also widen existing types, enable 'write.merge-schema.type-widening'.");
+
+ public static final ConfigOption<Boolean> TYPE_WIDENING =
+ key("write.merge-schema.type-widening")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Only effective when 'write.merge-schema' is true. "
+ + "If true, widen an existing column type when the incoming data has a wider "
+ + "compatible type (e.g. INT -> BIGINT, DECIMAL precision increase). "
+ + "Lossy changes are still rejected unless 'write.merge-schema.explicit-cast' is also true.");
public static final ConfigOption<Boolean> EXPLICIT_CAST =
key("write.merge-schema.explicit-cast")
.booleanType()
.defaultValue(false)
.withDescription(
- "If true, allow to merge data types if the two types meet the rules for explicit casting.");
+ "Only effective when 'write.merge-schema.type-widening' is true. "
+ + "If true, also allow lossy type changes between compatible types "
+ + "(e.g. BIGINT -> INT, STRING -> DATE).");
public static final ConfigOption<Boolean> USE_V2_WRITE =
key("write.use-v2-write")
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/MergeSchemaEvolutionHelper.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/MergeSchemaEvolutionHelper.scala
index f4dbbdac96..9275918398 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/MergeSchemaEvolutionHelper.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/MergeSchemaEvolutionHelper.scala
@@ -20,8 +20,7 @@ package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.spark.{SparkTable, SparkTypeUtils}
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
-import org.apache.paimon.spark.commands.SchemaHelper
-import org.apache.paimon.spark.schema.SparkSystemColumns
+import org.apache.paimon.spark.commands.SchemaEvolutionHelper
import org.apache.paimon.spark.util.OptionUtils
import org.apache.paimon.table.FileStoreTable
@@ -29,17 +28,17 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression, ExprId, Literal}
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, DeleteAction, InsertAction, MergeAction, MergeIntoTable}
-import org.apache.spark.sql.connector.catalog.TableCatalog
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.paimon.shims.SparkShimLoader
import org.apache.spark.sql.types.{StructField, StructType}
/**
- * Shared MERGE INTO `merge-schema=true` evolution. Triggers on `UPDATE *` / `INSERT *` (via
- * [[PaimonMergeActionTags]]) or on any explicit assignment whose key resolved to a source-bound
- * attribute (via [[PaimonMergeIntoResolver.resolveAssignments]] fallback — that shape is the
- * marker, no extra tag). Evolution is scoped to source columns referenced by matched / not-matched
- * actions; NOT MATCHED BY SOURCE can't reference source columns.
+ * MERGE INTO schema evolution (merge-schema=true). Computes the evolved schema in memory and
+ * rewrites the merge plan so that action alignment targets the new columns; the actual schema
+ * commit is deferred to execution (the merge command's `run`).
+ *
+ * Triggered by `UPDATE *` / `INSERT *` or explicit source-bound assignment keys. Scoped to source
+ * columns referenced in matched/not-matched actions.
*/
trait MergeSchemaEvolutionHelper extends ExpressionHelper {
@@ -77,26 +76,24 @@ trait MergeSchemaEvolutionHelper extends ExpressionHelper {
}
val fileStoreTable = v2Table.getTable.asInstanceOf[FileStoreTable]
+ // Pass raw source types: the core merge decides whether to keep or widen the target type, and
+ // the action alignment layer casts incoming values to the result.
val sourceSchema = StructType(
merge.sourceTable.output
.filter(a => scopedNames.exists(n => resolver(n, a.name)))
.map(a => StructField(a.name, a.dataType, a.nullable)))
- val filteredSourceSchema = SparkSystemColumns.filterSparkSystemColumns(sourceSchema)
- val allowExplicitCast = OptionUtils.writeMergeSchemaExplicitCastEnabled()
- val caseSensitive = spark.sessionState.conf.caseSensitiveAnalysis
- val updatedFileStoreTable = SchemaHelper
- .mergeAndCommitSchema(fileStoreTable, filteredSourceSchema, allowExplicitCast, caseSensitive)
+ // Compute the evolved schema in memory only; the actual schema commit is deferred to execution
+ // (the merge command's run / the V2 write's toBatch) so analysis stays side-effect-free. The
+ // evolved relation presents the new columns to the plan; existing rows read them as NULL.
+ val updatedFileStoreTable = SchemaEvolutionHelper
+ .evolvedTableInMemory(fileStoreTable, sourceSchema, spark)
.getOrElse(return None)
- // Invalidate Spark catalog cache so subsequent queries see the new schema.
- for (catalog <- relation.catalog; ident <- relation.identifier) {
- catalog.asInstanceOf[TableCatalog].invalidateTable(ident)
- }
-
val updatedV2Table = v2Table.copy(table = updatedFileStoreTable)
- val mergedSparkSchema =
- SparkTypeUtils.fromPaimonRowType(updatedFileStoreTable.schema().logicalRowType())
- val newOutput = buildEvolvedOutput(mergedSparkSchema, relation.output, resolver)
+ val newOutput = buildEvolvedOutput(
+ SparkTypeUtils.fromPaimonRowType(updatedFileStoreTable.schema().logicalRowType()),
+ relation.output,
+ resolver)
val updatedRelation =
SparkShimLoader.shim.copyDataSourceV2Relation(relation, updatedV2Table, newOutput)
val updatedTargetTable = merge.targetTable.transform {
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 5400e9865f..c26eeedd3f 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
@@ -18,12 +18,12 @@
package org.apache.paimon.spark.catalyst.analysis
-import org.apache.paimon.spark.{SparkConnectorOptions, SparkTable}
+import org.apache.paimon.options.Options
+import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable
import org.apache.paimon.spark.catalyst.plans.logical.PaimonDropPartitions
-import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand}
-import org.apache.paimon.spark.util.OptionUtils
+import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonShowColumnsCommand, SchemaEvolutionHelper}
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.{PaimonUtils, SparkSession}
@@ -35,6 +35,8 @@ import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.TableCapability
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation}
+import scala.collection.JavaConverters._
+
class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
import DataSourceV2Implicits._
import PaimonAnalysis._
@@ -43,12 +45,17 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
case a @ PaimonV2WriteCommand(table)
if !paimonWriteResolved(a.query, table) &&
a.query.getTagValue(PAIMON_WRITE_RESOLVED).isEmpty =>
- val mergeSchemaEnabled =
- writeOptions(a).get(SparkConnectorOptions.MERGE_SCHEMA.key()).contains("true") ||
- OptionUtils.writeMergeSchemaEnabled()
+ val options = Options.fromMap(writeOptions(a).asJava)
+ val mergeSchemaEnabled = SchemaEvolutionHelper.mergeSchemaEnabled(options)
+ val expected = SchemaEvolutionHelper.expectedAttrsForCatalogWrite(
+ table,
+ a.query.schema,
+ options,
+ a.isByName,
+ session)
val newQuery = PaimonOutputResolver.resolveOutputColumns(
table.name,
- table.output,
+ expected,
a.query,
a.isByName,
mergeSchemaEnabled)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAssignmentUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAssignmentUtils.scala
index 2b2c13bba8..a1e5dedc75 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAssignmentUtils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAssignmentUtils.scala
@@ -22,9 +22,9 @@ import org.apache.paimon.spark.SparkTypeUtils.CURRENT_DEFAULT_COLUMN_METADATA_KE
import org.apache.paimon.spark.catalyst.Compatibility
import org.apache.paimon.spark.catalyst.analysis.PaimonOutputResolver.MissingFieldBehavior
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{PaimonUtils, SparkSession}
import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, CreateNamedStruct, Expression, GetStructField, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateNamedStruct, Expression, GetStructField, Literal}
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, DeleteAction, InsertAction, InsertStarAction, MergeAction, UpdateStarAction}
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.paimon.shims.SparkShimLoader
@@ -192,7 +192,7 @@ object PaimonAssignmentUtils extends SQLConfHelper {
col.dataType match {
case structType: StructType =>
- val fieldAttrs = toAttributes(structType)
+ val fieldAttrs = PaimonUtils.toAttributes(structType)
val fieldExprs = structType.fields.zipWithIndex.map {
case (field, ordinal) => GetStructField(colExpr, ordinal, Some(field.name))
}
@@ -225,10 +225,6 @@ object PaimonAssignmentUtils extends SQLConfHelper {
attr.withDataType(CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType))
}
- private def toAttributes(structType: StructType): Seq[Attribute] = {
- structType.map(f => AttributeReference(f.name, f.dataType, f.nullable, f.metadata)())
- }
-
/** Parse the column's `CURRENT_DEFAULT` SQL (cast to column type) or fall back to NULL. */
private def getDefaultValueOrNull(attr: Attribute): Expression = {
val nullLit = Literal(null, attr.dataType)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
index eaa5cb19b8..be8424bf9b 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
@@ -66,7 +66,8 @@ case class PaimonMergeInto(spark: SparkSession)
primaryKeys)
}
- // Commit schema changes before alignment so the aligned plan sees new columns.
+ // Evolve the target schema in memory before alignment so the aligned plan sees the new
+ // columns; the commit is deferred to execution (the merge command's run).
val (resolvedMerge, targetOutput) =
evolveTargetIfNeeded(merge, relation, v2Table, spark, resolveNotMatchedBySourceActions)
.map { case (m, r, t) => v2Table = t; (m, r.output) }
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala
index 20da69c321..37ffebe828 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.spark.catalyst.Compatibility
+import org.apache.spark.sql.PaimonUtils
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
@@ -40,7 +41,8 @@ object PaimonOutputResolver extends SQLConfHelper {
* How nested struct field misalignment is handled:
* - [[FailMissing]]: strict — nested missing target / source-extra throws.
* - [[NullForMissing]]: merge-schema for INSERT / explicit UPDATE — missing NULL-fills,
- * source-extras kept so [[org.apache.paimon.spark.commands.SchemaHelper]] evolves the table.
+ * source-extras kept so [[org.apache.paimon.spark.commands.SchemaEvolutionHelper]] evolves
+ * the table.
* - [[PreserveTarget]]: merge-schema for `UPDATE *` struct — missing source field substitutes
* `GetStructField(targetExpr, ordinal)` to keep the current target value.
*/
@@ -277,12 +279,12 @@ object PaimonOutputResolver extends SQLConfHelper {
reorderColumnsByName(
tableName,
fields,
- toAttributes(targetType),
+ PaimonUtils.toAttributes(targetType),
behavior,
targetExpr,
colPath)
} else {
- resolveColumnsByPosition(tableName, fields, toAttributes(targetType), colPath)
+ resolveColumnsByPosition(tableName, fields, PaimonUtils.toAttributes(targetType), colPath)
}
val targetNamedStruct = CreateStruct(resolved)
val res = maybeWrapWithNullPreservation(
@@ -472,10 +474,6 @@ object PaimonOutputResolver extends SQLConfHelper {
case other => other
}
- private def toAttributes(structType: StructType): Seq[Attribute] = {
- structType.map(f => AttributeReference(f.name, f.dataType, f.nullable, f.metadata)())
- }
-
private def restoreActualType(attr: Attribute): Attribute = {
attr.withDataType(CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType))
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 96f8c0c5cc..008f14ed3e 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -136,6 +136,8 @@ case class MergeIntoPaimonDataEvolutionTable(
lazy val tableSchema: StructType = v2Table.schema
override def run(sparkSession: SparkSession): Seq[Row] = {
+ // Persist the schema that the analyzer evolved in memory (commit deferred to execution).
+ SchemaEvolutionHelper.commitEvolvedSchemaAtExecution(table, targetRelation, sparkSession)
invokeMergeInto(sparkSession)
Seq.empty[Row]
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index 1822929854..89689e108c 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -71,6 +71,8 @@ case class MergeIntoPaimonTable(
override def run(sparkSession: SparkSession): Seq[Row] = {
// Avoid that more than one source rows match the same target row.
checkMatchRationality(sparkSession)
+ // Persist the schema that the analyzer evolved in memory (commit deferred to execution).
+ SchemaEvolutionHelper.commitEvolvedSchemaAtExecution(table, relation, sparkSession)
val commitMessages = if (withPrimaryKeys) {
performMergeForPkTable(sparkSession)
} else {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaEvolutionHelper.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaEvolutionHelper.scala
new file mode 100644
index 0000000000..7bfb7ee905
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaEvolutionHelper.scala
@@ -0,0 +1,200 @@
+/*
+ * 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.commands
+
+import org.apache.paimon.options.Options
+import org.apache.paimon.schema.{SchemaMergingUtils, TableSchema}
+import org.apache.paimon.spark.{SparkConnectorOptions, SparkTable, SparkTypeUtils}
+import org.apache.paimon.spark.catalyst.analysis.PaimonOutputResolver
+import org.apache.paimon.spark.schema.SparkSystemColumns
+import org.apache.paimon.spark.util.OptionUtils
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.types.RowType
+
+import org.apache.spark.sql.{DataFrame, PaimonUtils, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.StructType
+
+import scala.collection.JavaConverters._
+
+/** Schema evolution flags resolved from write options and session conf. */
+private[spark] case class SchemaEvolutionFlags(
+ typeWidening: Boolean,
+ allowExplicitCast: Boolean,
+ caseSensitive: Boolean)
+
+/**
+ * Schema evolution entry points for catalog writes. The two `mergeSchema` overloads commit the
+ * evolved schema at execution (`WriteIntoPaimonTable.run` for V1, `PaimonV2Write.toBatch` for V2).
+ */
+private[spark] trait SchemaEvolutionHelper extends WithFileStoreTable {
+
+ val originTable: FileStoreTable
+
+ protected var newTable: Option[FileStoreTable] = None
+
+ override def table: FileStoreTable = newTable.getOrElse(originTable)
+
+ /**
+ * V1 catalog write entry (`WriteIntoPaimonTable.run`). The data is already cast to the evolved
+ * schema by [[PaimonOutputResolver]] during analysis, so this only commits the schema.
+ */
+ def mergeSchema(sparkSession: SparkSession, input: DataFrame, options: Options): Unit =
+ if (SchemaEvolutionHelper.mergeSchemaEnabled(options))
+ commitEvolution(sparkSession, input.schema, options)
+
+ /** V2 catalog write entry (`PaimonV2Write.toBatch`). Commits and returns the write schema. */
+ def mergeSchema(dataSchema: StructType, options: Options): StructType = {
+ if (!SchemaEvolutionHelper.mergeSchemaEnabled(options)) return dataSchema
+ commitEvolution(SparkSession.active, dataSchema, options) match {
+ case Some(evolved) => SparkTypeUtils.fromPaimonRowType(evolved.schema().logicalRowType())
+ case None => dataSchema
+ }
+ }
+
+ /** Commit the evolved schema for the incoming data; updates `newTable` and returns it. */
+ private def commitEvolution(
+ sparkSession: SparkSession,
+ dataSchema: StructType,
+ options: Options): Option[FileStoreTable] =
+ if (SchemaEvolutionHelper.commitSchemaEvolution(table, dataSchema, sparkSession, options)) {
+ val evolved = table.copyWithLatestSchema()
+ newTable = Some(evolved)
+ Some(evolved)
+ } else {
+ None
+ }
+
+ def updateTableWithOptions(options: Map[String, String]): Unit = {
+ newTable = Some(table.copy(options.asJava))
+ }
+}
+
+private[spark] object SchemaEvolutionHelper {
+
+ /**
+ * The single side-effect-free merge primitive (drops system columns; `None` when unchanged),
+ * shared by [[expectedAttrsForCatalogWrite]] and [[evolvedTableInMemory]];
+ * [[commitSchemaEvolution]] is the persisting counterpart.
+ */
+ private def computeMergedSchema(
+ table: FileStoreTable,
+ dataSchema: StructType,
+ flags: SchemaEvolutionFlags): Option[TableSchema] = {
+ val filtered = SparkSystemColumns.filterSparkSystemColumns(dataSchema)
+ val dataRowType = SparkTypeUtils.toPaimonType(filtered).asInstanceOf[RowType]
+ val current = table.schema()
+ val merged =
+ SchemaMergingUtils.mergeSchemas(
+ current,
+ dataRowType,
+ flags.typeWidening,
+ flags.allowExplicitCast,
+ flags.caseSensitive)
+ if (merged.equals(current)) None else Some(merged)
+ }
+
+ /**
+ * Filter system columns, resolve flags, and commit the merge of `dataSchema` into the stored
+ * schema. Returns whether a schema change was committed. Shared by catalog writes and MERGE INTO;
+ * callers that need the evolved table reload it via `copyWithLatestSchema`.
+ */
+ def commitSchemaEvolution(
+ table: FileStoreTable,
+ dataSchema: StructType,
+ sparkSession: SparkSession,
+ options: Options = new Options()): Boolean = {
+ val filtered = SparkSystemColumns.filterSparkSystemColumns(dataSchema)
+ val flags = readFlags(sparkSession, options)
+ val dataRowType = SparkTypeUtils.toPaimonType(filtered).asInstanceOf[RowType]
+ table
+ .store()
+ .mergeSchema(dataRowType, flags.typeWidening, flags.allowExplicitCast, flags.caseSensitive)
+ }
+
+ /**
+ * Persist the schema that MERGE INTO evolved in memory (see [[evolvedTableInMemory]]) and refresh
+ * the catalog cache. Called from the merge command's `run`, so the commit happens at execution,
+ * not during analysis.
+ */
+ def commitEvolvedSchemaAtExecution(
+ table: FileStoreTable,
+ relation: DataSourceV2Relation,
+ sparkSession: SparkSession): Unit = {
+ if (!OptionUtils.writeMergeSchemaEnabled()) return
+ val evolved = SparkTypeUtils.fromPaimonRowType(table.schema().logicalRowType())
+ if (commitSchemaEvolution(table, evolved, sparkSession)) {
+ // Refresh the catalog cache so later queries see the new schema.
+ for (catalog <- relation.catalog; ident <- relation.identifier) {
+ catalog.asInstanceOf[TableCatalog].invalidateTable(ident)
+ }
+ }
+ }
+
+ /**
+ * The in-memory post-evolution table (not persisted), letting MERGE INTO show the new columns in
+ * the plan. `mergeSchemas` assigns the next schema id deterministically, so the deferred
+ * [[commitSchemaEvolution]] reproduces an identical schema. `None` when unchanged.
+ */
+ def evolvedTableInMemory(
+ table: FileStoreTable,
+ dataSchema: StructType,
+ sparkSession: SparkSession,
+ options: Options = new Options()): Option[FileStoreTable] =
+ computeMergedSchema(table, dataSchema, readFlags(sparkSession, options)).map(table.copy)
+
+ /** Whether schema evolution is enabled, from the per-write options or the session conf. */
+ def mergeSchemaEnabled(options: Options): Boolean =
+ options.get(SparkConnectorOptions.MERGE_SCHEMA) || OptionUtils.writeMergeSchemaEnabled()
+
+ /**
+ * Compute the resolver's expected attributes for a catalog write. When type widening is enabled
+ * with `byName` resolution, returns the post-evolution attrs so the resolver can cast incoming
+ * data to the widened target types; otherwise returns `table.output` unchanged.
+ */
+ def expectedAttrsForCatalogWrite(
+ table: DataSourceV2Relation,
+ querySchema: StructType,
+ options: Options,
+ isByName: Boolean,
+ sparkSession: SparkSession): Seq[Attribute] = {
+ val flags = readFlags(sparkSession, options)
+ if (!isByName || !mergeSchemaEnabled(options) || !flags.typeWidening) return table.output
+
+ table.table.asInstanceOf[SparkTable].getTable match {
+ case fst: FileStoreTable =>
+ computeMergedSchema(fst, querySchema, flags)
+ .map(s => PaimonUtils.toAttributes(SparkTypeUtils.fromPaimonRowType(s.logicalRowType())))
+ .getOrElse(table.output)
+ case _ => table.output
+ }
+ }
+
+ /** Resolve schema evolution flags from write options and session conf. */
+ private def readFlags(sparkSession: SparkSession, options: Options): SchemaEvolutionFlags = {
+ val typeWidening = options.get(SparkConnectorOptions.TYPE_WIDENING) || OptionUtils
+ .writeMergeSchemaTypeWideningEnabled()
+ val allowExplicitCast = options.get(SparkConnectorOptions.EXPLICIT_CAST) || OptionUtils
+ .writeMergeSchemaExplicitCastEnabled()
+ val caseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
+ SchemaEvolutionFlags(typeWidening, allowExplicitCast, caseSensitive)
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaHelper.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaHelper.scala
deleted file mode 100644
index 928d6211d1..0000000000
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/SchemaHelper.scala
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * 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.commands
-
-import org.apache.paimon.options.Options
-import org.apache.paimon.spark.{SparkConnectorOptions, SparkTypeUtils}
-import org.apache.paimon.spark.schema.SparkSystemColumns
-import org.apache.paimon.spark.util.OptionUtils
-import org.apache.paimon.table.FileStoreTable
-import org.apache.paimon.types.RowType
-
-import org.apache.spark.sql.{Column, DataFrame, PaimonUtils, SparkSession}
-import org.apache.spark.sql.functions.{col, lit, struct, transform, transform_values}
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType}
-
-import scala.collection.JavaConverters._
-
-private[spark] trait SchemaHelper extends WithFileStoreTable {
-
- val originTable: FileStoreTable
-
- protected var newTable: Option[FileStoreTable] = None
-
- override def table: FileStoreTable = newTable.getOrElse(originTable)
-
- def mergeSchema(sparkSession: SparkSession, input: DataFrame, options: Options): DataFrame = {
- val dataSchema = SparkSystemColumns.filterSparkSystemColumns(input.schema)
- val writeSchema = mergeSchema(sparkSession, dataSchema, options)
- if (!PaimonUtils.sameType(writeSchema, dataSchema)) {
- val resolve = sparkSession.sessionState.conf.resolver
- val cols = SchemaHelper.alignColumns(writeSchema, dataSchema, resolve)
- input.select(cols: _*)
- } else {
- input
- }
- }
-
- def mergeSchema(dataSchema: StructType, options: Options): StructType = {
- mergeSchema(SparkSession.active, dataSchema, options)
- }
-
- def mergeSchema(
- sparkSession: SparkSession,
- dataSchema: StructType,
- options: Options): StructType = {
- val mergeSchemaEnabled =
- options.get(SparkConnectorOptions.MERGE_SCHEMA) || OptionUtils.writeMergeSchemaEnabled()
- if (!mergeSchemaEnabled) {
- return dataSchema
- }
-
- val filteredDataSchema = SparkSystemColumns.filterSparkSystemColumns(dataSchema)
- val allowExplicitCast = options.get(SparkConnectorOptions.EXPLICIT_CAST) || OptionUtils
- .writeMergeSchemaExplicitCastEnabled()
- val caseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
- SchemaHelper
- .mergeAndCommitSchema(table, filteredDataSchema, allowExplicitCast, caseSensitive)
- .foreach { updatedTable => newTable = Some(updatedTable) }
-
- val writeSchema = SparkTypeUtils.fromPaimonRowType(table.schema().logicalRowType())
- if (!PaimonUtils.sameType(writeSchema, filteredDataSchema)) {
- writeSchema
- } else {
- filteredDataSchema
- }
- }
-
- def updateTableWithOptions(options: Map[String, String]): Unit = {
- newTable = Some(table.copy(options.asJava))
- }
-}
-
-private[spark] object SchemaHelper {
-
- /**
- * Merge the given dataSchema into the table's schema. If the schema changed, commit the change
- * and return the updated table; otherwise return None.
- */
- def mergeAndCommitSchema(
- table: FileStoreTable,
- dataSchema: StructType,
- allowExplicitCast: Boolean,
- caseSensitive: Boolean = true): Option[FileStoreTable] = {
- val dataRowType = SparkTypeUtils.toPaimonType(dataSchema).asInstanceOf[RowType]
- if (table.store().mergeSchema(dataRowType, allowExplicitCast, caseSensitive)) {
- Some(table.copyWithLatestSchema())
- } else {
- None
- }
- }
-
- /**
- * Recursively align columns from dataSchema to targetSchema by name. For nested struct fields,
- * reorder and fill nulls for missing sub-fields.
- */
- def alignColumns(
- targetSchema: StructType,
- dataSchema: StructType,
- resolve: (String, String) => Boolean): Seq[Column] = {
- targetSchema.map {
- targetField =>
- dataSchema.find(f => resolve(f.name, targetField.name)) match {
- case Some(dataField) =>
- alignColumn(col(dataField.name), dataField.dataType, targetField, resolve)
- case _ =>
- lit(null).cast(targetField.dataType).as(targetField.name)
- }
- }
- }
-
- private def alignColumn(
- sourceCol: Column,
- sourceType: DataType,
- targetField: StructField,
- resolve: (String, String) => Boolean): Column = {
- (sourceType, targetField.dataType) match {
- case (s: StructType, t: StructType) if !PaimonUtils.sameType(s, t) =>
- alignStruct(sourceCol, s, t, resolve).as(targetField.name)
- case (ArrayType(s: StructType, _), ArrayType(t: StructType, _))
- if !PaimonUtils.sameType(s, t) =>
- transform(sourceCol, elem => alignStruct(elem, s, t, resolve))
- .as(targetField.name)
- case (MapType(sKey, sVal: StructType, _), MapType(tKey, tVal: StructType, _))
- if !PaimonUtils.sameType(sVal, tVal) =>
- transform_values(sourceCol, (_, v) => alignStruct(v, sVal, tVal, resolve))
- .as(targetField.name)
- case _ =>
- sourceCol.as(targetField.name)
- }
- }
-
- private def alignStruct(
- sourceCol: Column,
- sourceType: StructType,
- targetType: StructType,
- resolve: (String, String) => Boolean): Column = {
- val subCols = targetType.map {
- subTargetField =>
- sourceType.find(f => resolve(f.name, subTargetField.name)) match {
- case Some(subDataField) =>
- alignColumn(
- sourceCol.getField(subDataField.name),
- subDataField.dataType,
- subTargetField,
- resolve)
- case _ =>
- lit(null).cast(subTargetField.dataType).as(subTargetField.name)
- }
- }
- struct(subCols: _*)
- }
-}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
index 623c88d72f..fcf061ec67 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
@@ -41,7 +41,7 @@ case class WriteIntoPaimonTable(
batchId: Option[Long] = None)
extends RunnableCommand
with ExpressionHelper
- with SchemaHelper
+ with SchemaEvolutionHelper
with Logging {
override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -49,7 +49,7 @@ case class WriteIntoPaimonTable(
PaimonUtils.createDataset(
sparkSession,
ReplacePaimonFunctions(sparkSession)(_data.queryExecution.analyzed))
- val data = mergeSchema(sparkSession, replacedData, options)
+ mergeSchema(sparkSession, replacedData, options)
val (dynamicPartitionOverwriteMode, overwritePartition) = parseSaveMode()
// use the extra options to rebuild the table object
@@ -60,7 +60,7 @@ case class WriteIntoPaimonTable(
if (overwritePartition != null) {
writer.writeBuilder.withOverwrite(overwritePartition.asJava)
}
- val commitMessages = writer.write(data)
+ val commitMessages = writer.write(replacedData)
writer.commit(commitMessages)
Seq.empty
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
index 62712950e7..9d0a1795b5 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/sources/PaimonSink.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.sources
import org.apache.paimon.options.Options
import org.apache.paimon.spark.{InsertInto, Overwrite}
-import org.apache.paimon.spark.commands.{SchemaHelper, WriteIntoPaimonTable}
+import org.apache.paimon.spark.commands.{SchemaEvolutionHelper, WriteIntoPaimonTable}
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.{DataFrame, PaimonUtils, SQLContext}
@@ -35,7 +35,7 @@ class PaimonSink(
outputMode: OutputMode,
options: Options)
extends Sink
- with SchemaHelper {
+ with SchemaEvolutionHelper {
override def addBatch(batchId: Long, data: DataFrame): Unit = {
val saveMode = if (outputMode == OutputMode.Complete()) {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
index 7a6fa547c2..66b75bf122 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala
@@ -113,6 +113,10 @@ object OptionUtils extends SQLConfHelper with Logging {
getOptionString(SparkConnectorOptions.EXPLICIT_CAST).toBoolean
}
+ def writeMergeSchemaTypeWideningEnabled(): Boolean = {
+ getOptionString(SparkConnectorOptions.TYPE_WIDENING).toBoolean
+ }
+
def v1FunctionEnabled(): Boolean = {
getOptionString(SparkCatalogOptions.V1FUNCTION_ENABLED).toBoolean
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
index c7a9a9ff3a..2ae1dd53a3 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
@@ -21,7 +21,7 @@ package org.apache.paimon.spark.write
import org.apache.paimon.CoreOptions.ChangelogProducer
import org.apache.paimon.options.Options
import org.apache.paimon.spark._
-import org.apache.paimon.spark.commands.SchemaHelper
+import org.apache.paimon.spark.commands.SchemaEvolutionHelper
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
import org.apache.paimon.table.BucketMode.BUCKET_UNAWARE
import org.apache.paimon.table.FileStoreTable
@@ -44,10 +44,9 @@ class PaimonV2Write(
options: Options
) extends Write
with RequiresDistributionAndOrdering
- with SchemaHelper
+ with SchemaEvolutionHelper
with Logging {
- private val writeSchema = mergeSchema(dataSchema, options)
private val writeRequirement = PaimonWriteRequirement(table)
override def requiredDistribution(): Distribution = {
@@ -63,6 +62,8 @@ class PaimonV2Write(
}
override def toBatch: BatchWrite = {
+ // Commit the evolved schema at execution (not at planning), then write to the evolved table.
+ val writeSchema = mergeSchema(dataSchema, options)
SparkShimLoader.shim.createPaimonBatchWrite(
table,
writeSchema,
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/PaimonUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/PaimonUtils.scala
index d8b70f68f2..16d34ecf4c 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/PaimonUtils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/PaimonUtils.scala
@@ -22,7 +22,7 @@ import org.apache.spark.executor.OutputMetrics
import org.apache.spark.rdd.InputFileBlockHolder
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.connector.expressions.FieldReference
import org.apache.spark.sql.connector.expressions.filter.Predicate
@@ -136,6 +136,10 @@ object PaimonUtils {
DataType.equalsIgnoreCompatibleNullability(from, to)
}
+ /** `StructType` to fresh `AttributeReference`s (the `StructType.toAttributes` removed in 3.4+). */
+ def toAttributes(schema: StructType): Seq[AttributeReference] =
+ schema.map(f => AttributeReference(f.name, f.dataType, f.nullable, f.metadata)())
+
def classIsLoadable(clazz: String): Boolean = {
SparkUtils.classIsLoadable(clazz)
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
index dcb99224cd..f22dab4b9c 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PaimonSinkTest.scala
@@ -263,6 +263,7 @@ class PaimonSinkTest extends PaimonSparkTestBase with StreamTest {
.writeStream
.option("checkpointLocation", checkpointDir.getCanonicalPath)
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.option("write.merge-schema.explicit-cast", "true")
.format("paimon")
.start(location)
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
index 87d80cb897..de31d8ccc4 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DataFrameWriteTestBase.scala
@@ -166,7 +166,6 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
.option("bucket", "-1")
.option("target-file-size", "256MB")
.option("write.merge-schema", "true")
- .option("write.merge-schema.explicit-cast", "true")
.saveAsTable("test_ctas")
val paimonTable = loadTable("test_ctas")
@@ -179,7 +178,6 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
// non-core options should not be here.
Assertions.assertFalse(paimonTable.options().containsKey("write.merge-schema"))
- Assertions.assertFalse(paimonTable.options().containsKey("write.merge-schema.explicit-cast"))
}
}
@@ -597,6 +595,7 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
.format("paimon")
.mode("append")
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.save(location)
val expected3 = if (hasPk) {
Row(1L, "a2", BigDecimal.decimal(123), Map("k" -> 11.1)) :: Row(
@@ -641,6 +640,7 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
.format("paimon")
.mode("append")
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.save(location)
val expected4 =
expected3 ++ Seq(Row(99L, "df4", BigDecimal.decimal(4.0), Map("4" -> 4.1)))
@@ -712,19 +712,21 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
"c",
"d")
- // throw UnsupportedOperationException if write.merge-schema.explicit-cast = false
+ // throw UnsupportedOperationException when type-widening is on but explicit-cast = false
assertThrows[UnsupportedOperationException] {
df3.write
.format("paimon")
.mode("append")
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.save(location)
}
- // merge schema and write data when write.merge-schema.explicit-cast = true
+ // merge schema and write data when type-widening + explicit-cast = true
df3.write
.format("paimon")
.mode("append")
.option("write.merge-schema", "true")
+ .option("write.merge-schema.type-widening", "true")
.option("write.merge-schema.explicit-cast", "true")
.save(location)
val expected3 = if (hasPk) {
@@ -864,56 +866,50 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
}
test("Paimon Schema Evolution: some columns is absent in the coming data") {
+ withTable("T") {
+ spark.sql("CREATE TABLE T (a INT, b STRING)")
- spark.sql(s"""
- |CREATE TABLE T (a INT, b STRING)
- |""".stripMargin)
+ val df1 = Seq((1, "2023-08-01"), (2, "2023-08-02")).toDF("a", "b")
+ df1.write.format("paimon").mode("append").saveAsTable("T")
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY a, b"),
+ Row(1, "2023-08-01") :: Row(2, "2023-08-02") :: Nil)
+
+ // Case 1: two additional fields: DoubleType and TimestampType
+ val ts = java.sql.Timestamp.valueOf("2023-08-01 10:00:00.0")
+ val df2 = Seq((1, "2023-08-01", 12.3d, ts), (3, "2023-08-03", 34.5d, ts))
+ .toDF("a", "b", "c", "d")
+ df2.write
+ .format("paimon")
+ .mode("append")
+ .option("write.merge-schema", "true")
+ .saveAsTable("T")
- val paimonTable = loadTable("T")
- val location = paimonTable.location().toString
-
- val df1 = Seq((1, "2023-08-01"), (2, "2023-08-02")).toDF("a", "b")
- df1.write.format("paimon").mode("append").save(location)
- checkAnswer(
- spark.sql("SELECT * FROM T ORDER BY a, b"),
- Row(1, "2023-08-01") :: Row(2, "2023-08-02") :: Nil)
-
- // Case 1: two additional fields: DoubleType and TimestampType
- val ts = java.sql.Timestamp.valueOf("2023-08-01 10:00:00.0")
- val df2 = Seq((1, "2023-08-01", 12.3d, ts), (3, "2023-08-03", 34.5d, ts))
- .toDF("a", "b", "c", "d")
- df2.write
- .format("paimon")
- .mode("append")
- .option("write.merge-schema", "true")
- .save(location)
-
- // Case 2: colum b and d are absent in the coming data
- val df3 = Seq((4, 45.6d), (5, 56.7d))
- .toDF("a", "c")
- df3.write
- .format("paimon")
- .mode("append")
- .option("write.merge-schema", "true")
- .save(location)
- val expected3 =
- Row(1, "2023-08-01", null, null) :: Row(1, "2023-08-01", 12.3d, ts) :: Row(
- 2,
- "2023-08-02",
- null,
- null) :: Row(3, "2023-08-03", 34.5d, ts) :: Row(4, null, 45.6d, null) :: Row(
- 5,
- null,
- 56.7d,
- null) :: Nil
- checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"), expected3)
+ // Case 2: column b and d are absent in the coming data
+ val df3 = Seq((4, 45.6d), (5, 56.7d))
+ .toDF("a", "c")
+ df3.write
+ .format("paimon")
+ .mode("append")
+ .option("write.merge-schema", "true")
+ .saveAsTable("T")
+ val expected3 =
+ Row(1, "2023-08-01", null, null) :: Row(1, "2023-08-01", 12.3d, ts) :: Row(
+ 2,
+ "2023-08-02",
+ null,
+ null) :: Row(3, "2023-08-03", 34.5d, ts) :: Row(4, null, 45.6d, null) :: Row(
+ 5,
+ null,
+ 56.7d,
+ null) :: Nil
+ checkAnswer(spark.sql("SELECT * FROM T ORDER BY a, b"), expected3)
+ }
}
test("Paimon write merge-schema conflict: deep nested array element bigint -> string") {
for (useV2Write <- Seq("true", "false")) {
- withSparkSQLConf(
- "spark.paimon.write.use-v2-write" -> useV2Write,
- "spark.paimon.write.merge-schema.explicit-cast" -> "true") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2Write) {
withTable("target") {
sql("""
|CREATE TABLE target (
@@ -953,9 +949,7 @@ abstract class DataFrameWriteTestBase extends PaimonSparkTestBase {
test("Paimon write merge-schema conflict: top-level same-name column string vs bigint") {
for (useV2Write <- Seq("true", "false")) {
- withSparkSQLConf(
- "spark.paimon.write.use-v2-write" -> useV2Write,
- "spark.paimon.write.merge-schema.explicit-cast" -> "true") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2Write) {
withTable("target") {
sql("CREATE TABLE target (id STRING, value BIGINT) USING paimon")
sql("INSERT INTO target VALUES ('r0', 1000L)")
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteMergeSchemaTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteMergeSchemaTest.scala
index 2a17d6297e..b1bf57193b 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteMergeSchemaTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/V2WriteMergeSchemaTest.scala
@@ -30,6 +30,7 @@ class V2WriteMergeSchemaTest extends PaimonSparkTestBase {
.set("spark.sql.catalog.paimon.cache-enabled", "false")
.set("spark.paimon.write.use-v2-write", "true")
.set("spark.paimon.write.merge-schema", "true")
+ .set("spark.paimon.write.merge-schema.type-widening", "true")
.set("spark.paimon.write.merge-schema.explicit-cast", "true")
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMergeSchemaTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMergeSchemaTest.scala
index ae830bcd97..094bd60e48 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMergeSchemaTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/WriteMergeSchemaTest.scala
@@ -482,7 +482,7 @@ class WriteMergeSchemaTest extends PaimonSparkTestBase {
withTable("t") {
withSparkSQLConf(
"spark.paimon.write.merge-schema" -> "true",
- "spark.paimon.write.merge-schema.explicit-cast" -> "true") {
+ "spark.paimon.write.merge-schema.type-widening" -> "true") {
sql("CREATE TABLE t (id INT, info STRUCT<f1 INT, f2 STRING>)")
sql("INSERT INTO t VALUES (1, struct(10, 'a'))")
@@ -641,6 +641,28 @@ class WriteMergeSchemaTest extends PaimonSparkTestBase {
}
}
+ test("Merge into with merge-schema: append-only target evolves schema") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, name STRING) USING paimon")
+ sql("INSERT INTO t VALUES (1, 'a'), (2, 'b')")
+
+ spark
+ .sql("SELECT 1 AS id, 'a2' AS name, 100 AS value UNION ALL SELECT 3 AS id, 'c' AS name, 30 AS value")
+ .createOrReplaceTempView("s")
+
+ withSparkSQLConf("spark.paimon.write.merge-schema" -> "true") {
+ sql("""MERGE INTO t USING s ON t.id = s.id
+ | WHEN MATCHED THEN UPDATE SET *
+ | WHEN NOT MATCHED THEN INSERT *""".stripMargin)
+ }
+
+ assert(spark.table("t").schema.fieldNames.toSeq == Seq("id", "name", "value"))
+ checkAnswer(
+ sql("SELECT id, name, value FROM t ORDER BY id"),
+ Seq(Row(1, "a2", 100), Row(2, "b", null), Row(3, "c", 30)))
+ }
+ }
+
test("Merge into with merge-schema: extra column with case-mismatched existing columns") {
withTable("t") {
sql("""CREATE TABLE t (id INT, name STRING)
@@ -775,6 +797,46 @@ class WriteMergeSchemaTest extends PaimonSparkTestBase {
}
}
+ test("Merge into without merge-schema: ARRAY<INT> target with ARRAY<BIGINT> source") {
+ withTable("t") {
+ sql("""CREATE TABLE t (id INT, ports ARRAY<INT>)
+ | USING paimon
+ | TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')""".stripMargin)
+ sql("INSERT INTO t VALUES (1, array(80, 443))")
+
+ spark
+ .sql("SELECT 1 AS id, array(cast(8080 as bigint), cast(9090 as bigint)) AS ports")
+ .createOrReplaceTempView("s")
+
+ sql("""MERGE INTO t USING s ON t.id = s.id
+ | WHEN MATCHED THEN UPDATE SET *
+ | WHEN NOT MATCHED THEN INSERT *""".stripMargin)
+
+ val portsType = spark.table("t").schema("ports").dataType
+ assert(
+ portsType.simpleString == "array<int>",
+ s"Expected array<int> but got ${portsType.simpleString}")
+
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, Seq(8080, 9090))))
+ }
+ }
+
+ test("Write merge schema: INT to BIGINT keeps target type") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, value INT)")
+ sql("INSERT INTO t VALUES (1, 100)")
+
+ withSparkSQLConf("spark.paimon.write.merge-schema" -> "true") {
+ sql("INSERT INTO t BY NAME SELECT 2 AS id, cast(200 as bigint) AS value")
+ }
+
+ val valueType = spark.table("t").schema("value").dataType
+ assert(valueType.simpleString == "int", s"Expected int but got ${valueType.simpleString}")
+
+ checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(1, 100), Row(2, 200)))
+ }
+ }
+
test("Merge into with merge-schema: case-sensitive mode treats different case as new columns") {
withTable("t") {
sql("""CREATE TABLE t (id INT, name STRING)
@@ -849,6 +911,32 @@ class WriteMergeSchemaTest extends PaimonSparkTestBase {
}
}
+ test("Merge into with merge-schema: ARRAY<INT> to ARRAY<BIGINT> keeps target type") {
+ withTable("t") {
+ sql("""CREATE TABLE t (id INT, ports ARRAY<INT>)
+ | USING paimon
+ | TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')""".stripMargin)
+ sql("INSERT INTO t VALUES (1, array(80, 443))")
+
+ spark
+ .sql("SELECT 1 AS id, array(cast(8080 as bigint), cast(9090 as bigint)) AS ports")
+ .createOrReplaceTempView("s")
+
+ withSparkSQLConf("spark.paimon.write.merge-schema" -> "true") {
+ sql("""MERGE INTO t USING s ON t.id = s.id
+ | WHEN MATCHED THEN UPDATE SET *
+ | WHEN NOT MATCHED THEN INSERT *""".stripMargin)
+ }
+
+ val portsType = spark.table("t").schema("ports").dataType
+ assert(
+ portsType.simpleString == "array<int>",
+ s"Expected array<int> but got ${portsType.simpleString}")
+
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, Seq(8080, 9090))))
+ }
+ }
+
test("Write merge schema: case-sensitive mode treats different case as new columns") {
withTable("t") {
sql("CREATE TABLE t (id INT, name STRING)")
@@ -866,4 +954,51 @@ class WriteMergeSchemaTest extends PaimonSparkTestBase {
s"Expected 5 columns (id, name, ID, NAME, extra) but got ${columnNames.length}: ${columnNames.mkString(", ")}")
}
}
+
+ test("Merge into with type-widening=true: INT to BIGINT widens existing column") {
+ withTable("t") {
+ sql("""CREATE TABLE t (id INT, v INT)
+ | USING paimon
+ | TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 10)")
+
+ spark.sql("SELECT 1 AS id, cast(200 AS bigint) AS v").createOrReplaceTempView("s")
+
+ withSparkSQLConf(
+ "spark.paimon.write.merge-schema" -> "true",
+ "spark.paimon.write.merge-schema.type-widening" -> "true") {
+ sql("""MERGE INTO t USING s ON t.id = s.id
+ | WHEN MATCHED THEN UPDATE SET *""".stripMargin)
+ }
+
+ assert(spark.table("t").schema("v").dataType.simpleString == "bigint")
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1, 200L)))
+ }
+ }
+
+ test("Merge into with type-widening=true: ARRAY element widening throws (known limitation)") {
+ withTable("t") {
+ sql("""CREATE TABLE t (id INT, ports ARRAY<INT>)
+ | USING paimon
+ | TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')""".stripMargin)
+ sql("INSERT INTO t VALUES (1, array(80))")
+
+ spark
+ .sql("SELECT 2 AS id, array(cast(9090 AS bigint)) AS ports")
+ .createOrReplaceTempView("s")
+
+ // KNOWN LIMITATION: widening the element type of ARRAY/MAP is not yet supported by
+ // SchemaManager.generateTableSchema (CastExecutors has no ARRAY<INT> -> ARRAY<BIGINT>
+ // rule). Tracked as a follow-up; until then type-widening on complex element types throws.
+ withSparkSQLConf(
+ "spark.paimon.write.merge-schema" -> "true",
+ "spark.paimon.write.merge-schema.type-widening" -> "true") {
+ intercept[Exception] {
+ sql("""MERGE INTO t USING s ON t.id = s.id
+ | WHEN MATCHED THEN UPDATE SET *
+ | WHEN NOT MATCHED THEN INSERT *""".stripMargin)
+ }
+ }
+ }
+ }
}