This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 20d2b4c6f12 [fix](paimon) Cast static overwrite partition values to
their column types (#68874)
20d2b4c6f12 is described below
commit 20d2b4c6f1286d604132280c7382f3d74112deb5
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Oct 14 22:30:22 2026 +0800
[fix](paimon) Cast static overwrite partition values to their column types
(#68874)
### What problem does this PR solve?
Issue Number: None
Related PR: #65868
Problem Summary:
**In short.** This ports the static-partition part of branch-4.1 #65868
to the Paimon connector. An `INSERT OVERWRITE ... PARTITION (...)` named
the Paimon partition by the literal as the user wrote it, while the rows
carry that literal cast to the column type. Overwrites whose literal is
not spelled the column type's way failed, and overwrites of a TIMESTAMP
WITH LOCAL TIME ZONE partition failed whenever the session zone differed
from the FE JVM zone; when no rows were written to trip Paimon's check,
the overwrite silently deleted the wrong partition. Connectors now get
the cast values through a new write-handle getter, and Paimon uses them
for an overwrite.
**Background.** On a Paimon table, the bind step materializes the
literal of `PARTITION (col = <literal>)` into every row with
`castIfNotSameType`, so the rows carry the value in the column's type.
The static partition spec handed to connectors is built separately:
`PluginDrivenInsertCommandContext` keeps `Literal.getStringValue()` of
the literal as written, and `PaimonWriteBinding` passed that string to
Paimon's `withOverwrite`. Paimon parses the string with the column type
to find the partition to replace, parses an LTZ value as local time in
the FE JVM zone, and reads the table's `partition.default-name` as the
NULL partition. branch-4.1 #65868 casts each literal to its column type
first, moves an LTZ value from the session zone to the JVM zone, and
rejects a value equal to `partition.default-name`; master has none of
this.
**The problem, and what it cost.** On a local FE + BE with a filesystem
Paimon catalog, session zone Asia/Shanghai, FE JVM zone Europe/London:
| Statement | master |
|---|---|
| `INSERT OVERWRITE t PARTITION (p = true) VALUES (...)`, `p INT` |
fails: `NumberFormatException: For input string: "true"` |
| `... PARTITION (dt = 20240101) ...`, `dt DATE` | fails: `the changes
in dt=19723/ does not belong to this partition` (Paimon reads 20240101
as a day number) |
| `... PARTITION (ts = '2024-01-15 08:30:45') ...`, LTZ `ts` | fails:
`the changes in ts=2024-01-15T00%3A30%3A45/ does not belong to this
partition` |
| the same LTZ overwrite with no rows (`SELECT ... LIMIT 0`) | succeeds,
deletes the partition `2024-01-15 16:30:45` (session time) and keeps
`08:30:45` |
| an overwrite with no rows of `region = '<partition.default-name>'` |
succeeds, deletes the NULL partition |
The two empty overwrites are silent data loss: the statement reports
success and the deleted partition is not the one the user named. These
were measured before #68786, which now binds every Paimon LTZ column as
TIMESTAMPTZ; it leaves the static partition spec as written, so the LTZ
rows still apply.
**How this PR fixes it.**
- `ConnectorWriteHandle#getCastStaticPartitionSpec` (new): the static
partition values cast to the types of their columns, keyed like
`getStaticPartitionSpec`, which still carries the literals as written.
SQL NULL keys stay in `getStaticPartitionNullKeys` only; a key that
names no column (an Iceberg partition field) keeps its literal. The
default, for a handle that does not cast, is the spec as written without
its null keys.
- fe-core: `InsertIntoTableCommand` records the sink's bound target
schema on `PluginDrivenInsertCommandContext`, and the write handle
computes the cast values on demand with `checkedCastTo`, the cast
constant folding applies to the row value. A string written into a
string column follows BindSink instead, which does not cast it: the
value is cut to a CHAR / VARCHAR length in code points when the session
truncates strings on insert (`BindSink.truncatesStringOnInsert()`,
shared with BindSink), and kept as written otherwise. Only a connector
that asks pays for the cast and can fail on it, so Iceberg, MaxCompute
and Hive, which keep reading the spec as written, see no change. A value
that does not fit its column type fails the cast instead of becoming
NULL, as on branch-4.x. A binary value is spelled `0x` and its bytes in
hex, as in the spec as written.
- Paimon: `PaimonWriteBinding` resolves the static partition of an
overwrite from the cast values. An INSERT writes each row to the
partition its own values name and never reads the static partition, so
Paimon asks for the cast values only for an overwrite, and an INSERT is
planned as before (branch-4.x casts for an INSERT too). Doris binds an
LTZ column as TIMESTAMPTZ (#68786), so the cast value of an LTZ
partition is the instant in UTC with its offset; it is moved to the FE
JVM zone and written with a space between date and time, which Paimon's
parser needs. An overwrite value equal to `partition.default-name` is
rejected. An overwrite also rejects an LTZ value whose local time in the
JVM zone falls in a DST overlap, since both instants format to the same
value and Paimon reads it as the earlier one (as branch-4.2 does,
#68532). A static overwrite that gives a binary partition column a value
other than NULL is rejected: Paimon reads a static value as text and
takes its UTF-8 bytes, so neither the cast value nor the `0x4142` that
master passes names the partition, and an overwrite with no rows
committed against another one.
**Results.** The table above, with this PR:
| Statement | this PR |
|---|---|
| `PARTITION (p = true)` | replaces `p = 1` |
| `PARTITION (dt = 20240101)` | replaces `dt = 2024-01-01` |
| LTZ overwrite | replaces the named partition |
| LTZ overwrite with no rows | deletes the named partition only |
| `region = '<partition.default-name>'`, with or without rows | fails:
`... cannot be represented in a static overwrite`, no data changes |
| `INSERT OVERWRITE ... PARTITION (dt = '2024-13-45')` | fails:
`date/datetime literal [2024-13-45T00:00:00] is invalid` |
| `INSERT INTO ... PARTITION (dt = '2024-13-45')` | unchanged: writes
`dt = NULL` |
| `PARTITION (p = 'abcd')` on a CHAR(2) column, INSERT or OVERWRITE |
writes and replaces `p = 'ab'`, where the rows go |
| overwrite with no rows of `p = '😀x'` on a VARCHAR(1) column | deletes
the partition `😀` only |
| LTZ overwrite whose value is repeated in the FE JVM zone (UTC session,
`2024-10-27 01:30:45` with the JVM in Europe/London) | fails: `... is
ambiguous in FE JVM time zone Europe/London ...`, no data changes |
| LTZ overwrite of a value skipped in the session zone (`2024-03-31
01:30:45` in a Europe/London session) | replaces the partition its rows
are written to, the instant of the DST change (01:00:00Z); an empty one
deletes that partition |
| `PARTITION (p = X'4142')` on a BYTES column, with or without rows |
fails: `Static overwrite of binary partition column 'p' is not
supported`, no data changes (before, an empty one committed and left the
partition `AB` in place) |
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../doris/connector/paimon/PaimonWriteBinding.java | 91 ++++++--
.../connector/paimon/PaimonWritePlanProvider.java | 3 +-
.../paimon/PaimonBackendBoundTableTest.java | 46 +++-
.../paimon/PaimonConnectorTransactionTest.java | 30 ++-
.../connector/paimon/PaimonWriteBindingTest.java | 255 +++++++++++++++++++--
.../connector/spi/handle/ConnectorWriteHandle.java | 17 ++
.../spi/handle/ConnectorWriteHandleTest.java | 26 ++-
.../test/resources/connector-plugin-surface.txt | 1 +
.../doris/nereids/rules/analysis/BindSink.java | 15 +-
.../commands/insert/InsertIntoTableCommand.java | 1 +
.../insert/PluginDrivenInsertCommandContext.java | 78 ++++++-
.../doris/planner/PluginDrivenTableSink.java | 17 +-
.../planner/PluginDrivenTableSinkBindingTest.java | 169 +++++++++++++-
.../write/test_paimon_write_transaction.groovy | 49 ++++
14 files changed, 740 insertions(+), 58 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
index 26865c542c0..6328de70355 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java
@@ -18,17 +18,23 @@
package org.apache.doris.connector.paimon;
import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.utils.InstantiationUtil;
import java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
import java.util.Base64;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
-import java.util.Set;
+import java.util.Objects;
import java.util.TreeMap;
/** Statement-scoped Paimon write target shared by sink planning and
transaction commit. */
@@ -55,10 +61,12 @@ final class PaimonWriteBinding {
}
static PaimonWriteBinding create(PaimonTableHandle handle, FileStoreTable
table,
- Map<String, String> hadoopConfig, boolean overwrite,
- Map<String, String> requestedStaticPartition, Set<String>
staticPartitionNullKeys) {
- Map<String, String> staticPartition = resolveStaticPartition(
- table, requestedStaticPartition, staticPartitionNullKeys);
+ Map<String, String> hadoopConfig, ConnectorWriteHandle
writeHandle) {
+ boolean overwrite = writeHandle.isOverwrite();
+ // Only a static overwrite reads the partition values. An INSERT
writes each row to the partition its own
+ // values name, so it leaves them alone: casting them could only fail
a write the rows would accept.
+ Map<String, String> staticPartition = overwrite
+ ? resolveStaticPartition(table, writeHandle) :
Collections.emptyMap();
FileStoreTable writeTable = configureTableForWrite(table, overwrite,
staticPartition);
return new PaimonWriteBinding(handle.getDatabaseName() + "." +
handle.getTableName(),
writeTable, hadoopConfig, overwrite, staticPartition);
@@ -78,26 +86,81 @@ final class PaimonWriteBinding {
return table.copy(Collections.singletonMap(dynamicOverwriteKey,
Boolean.FALSE.toString()));
}
- private static Map<String, String> resolveStaticPartition(FileStoreTable
table,
- Map<String, String> requested, Set<String> nullKeys) {
+ /**
+ * The static partition of an overwrite, as Paimon's static overwrite
parses it: SQL NULL becomes the table's
+ * {@code partition.default-name}, and every other value is the one the
written rows carry, cast to the
+ * column type. A binary column can only be statically overwritten with
NULL.
+ */
+ static Map<String, String> resolveStaticPartition(FileStoreTable table,
ConnectorWriteHandle writeHandle) {
Map<String, String> canonicalNames = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
for (String partitionKey : table.partitionKeys()) {
canonicalNames.put(partitionKey, partitionKey);
}
- String defaultPartitionName =
CoreOptions.fromMap(table.options()).partitionDefaultName();
- Map<String, String> result = new LinkedHashMap<>();
- for (Map.Entry<String, String> entry : requested.entrySet()) {
- String canonicalName = canonicalNames.get(entry.getKey());
+ Map<String, String> requestedColumns = new LinkedHashMap<>();
+ for (String key : writeHandle.getStaticPartitionSpec().keySet()) {
+ String canonicalName = canonicalNames.get(key);
if (canonicalName == null) {
- throw new DorisConnectorException("Column '" + entry.getKey()
+ throw new DorisConnectorException("Column '" + key
+ "' is not a partition column of Paimon table");
}
- String value = entry.getValue();
- result.put(canonicalName, nullKeys.contains(entry.getKey()) ?
defaultPartitionName : value);
+ DataTypeRoot typeRoot =
table.rowType().getField(canonicalName).type().getTypeRoot();
+ if ((typeRoot == DataTypeRoot.BINARY || typeRoot ==
DataTypeRoot.VARBINARY)
+ &&
!writeHandle.getStaticPartitionNullKeys().contains(key)) {
+ // Paimon's static overwrite takes the UTF-8 bytes of a
value's string, which cannot name every
+ // binary partition, and a wrong name overwrites another
partition without an error. Checked
+ // before the values are cast.
+ throw new DorisConnectorException("Static overwrite of binary
partition column '"
+ + canonicalName + "' is not supported");
+ }
+ requestedColumns.put(key, canonicalName);
+ }
+ String defaultPartitionName =
CoreOptions.fromMap(table.options()).partitionDefaultName();
+ Map<String, String> castValues =
writeHandle.getCastStaticPartitionSpec();
+ Map<String, String> result = new LinkedHashMap<>();
+ for (Map.Entry<String, String> requested :
requestedColumns.entrySet()) {
+ String key = requested.getKey();
+ String canonicalName = requested.getValue();
+ if (writeHandle.getStaticPartitionNullKeys().contains(key)) {
+ result.put(canonicalName, defaultPartitionName);
+ continue;
+ }
+ String value = Objects.requireNonNull(castValues.get(key),
+ () -> "missing the cast value of static partition column "
+ key);
+ if (table.rowType().getField(canonicalName).type().getTypeRoot()
+ == DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
+ value = toSdkLocalTime(canonicalName, value);
+ }
+ if (defaultPartitionName.equals(value)) {
+ // Paimon's static overwrite reads this string as the NULL
partition of any partition type, so
+ // a value equal to it would overwrite the NULL partition
instead.
+ throw new DorisConnectorException("Static partition value for
column '" + canonicalName
+ + "' equals Paimon partition.default-name '" +
defaultPartitionName
+ + "' and cannot be represented in a static overwrite");
+ }
+ result.put(canonicalName, value);
}
return result;
}
+ /**
+ * Paimon parses a static overwrite value of a TIMESTAMP WITH LOCAL TIME
ZONE column as local time in the FE
+ * JVM's default zone. Doris binds such a column as TIMESTAMPTZ, so the
cast value is the instant in UTC with
+ * its offset; it is moved to the JVM zone, with a space between the date
and the time as Paimon's parser
+ * needs. An instant whose local time a DST change repeats there is
rejected: both instants of the overlap
+ * format to the same value, which Paimon reads as the earlier one.
+ */
+ private static String toSdkLocalTime(String column, String value) {
+ ZoneId sdkZone = ZoneId.systemDefault();
+ LocalDateTime sdkLocal = OffsetDateTime.parse(value.replace(' ', 'T'),
DateTimeFormatter.ISO_OFFSET_DATE_TIME)
+ .atZoneSameInstant(sdkZone).toLocalDateTime();
+ if (sdkZone.getRules().getValidOffsets(sdkLocal).size() > 1) {
+ throw new DorisConnectorException("Static LTZ partition value for
column '" + column
+ + "' is ambiguous in FE JVM time zone " + sdkZone
+ + " and cannot be represented in a static overwrite");
+ }
+ return
sdkLocal.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME).replace('T', ' ');
+ }
+
private static String serialize(FileStoreTable table) {
try {
return
Base64.getEncoder().encodeToString(InstantiationUtil.serializeObject(table));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
index f5d793dd5aa..620439e7cab 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWritePlanProvider.java
@@ -152,8 +152,7 @@ public class PaimonWritePlanProvider implements
ConnectorWritePlanProvider {
}
PaimonConnectorTransaction transaction = currentTransaction(session);
PaimonWriteBinding binding = PaimonWriteBinding.create(
- tableHandle, table, buildHadoopConfig(), handle.isOverwrite(),
- handle.getStaticPartitionSpec(),
handle.getStaticPartitionNullKeys());
+ tableHandle, table, buildHadoopConfig(), handle);
transaction.bind(binding);
TPaimonTableSink sink = new TPaimonTableSink();
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
index bca0c1544e0..f618ba4d816 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonBackendBoundTableTest.java
@@ -17,7 +17,10 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.spi.ConnectorColumn;
import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
@@ -63,6 +66,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
/**
* The table {@link PaimonScanPlanProvider#getScanNodeProperties} serializes
to the BE must carry no
@@ -116,8 +120,8 @@ public class PaimonBackendBoundTableTest {
FileStoreTable table = newTable(warehouse,
catalogEnvironment(catalog), C1);
PaimonWriteBinding binding = PaimonWriteBinding.create(
- dataHandle(), table, Collections.emptyMap(), false,
- Collections.emptyMap(), Collections.emptySet());
+ dataHandle(), table, Collections.emptyMap(),
+ writeHandle(false, Collections.emptyMap(),
Collections.emptySet()));
FileStoreTable backendTable =
deserializeTable(binding.getSerializedTable());
Assertions.assertNotNull(binding.getTable().catalogEnvironment().catalogLoader());
@@ -138,9 +142,9 @@ public class PaimonBackendBoundTableTest {
Map<String, String> requested = Collections.singletonMap("pt", "NULL");
PaimonWriteBinding literal = PaimonWriteBinding.create(dataHandle(),
table,
- Collections.emptyMap(), false, requested,
Collections.emptySet());
+ Collections.emptyMap(), writeHandle(true, requested,
Collections.emptySet()));
PaimonWriteBinding sqlNull = PaimonWriteBinding.create(dataHandle(),
table,
- Collections.emptyMap(), false, requested,
Collections.singleton("pt"));
+ Collections.emptyMap(), writeHandle(true, requested,
Collections.singleton("pt")));
Assertions.assertEquals("NULL",
literal.getStaticPartition().get("pt"));
Assertions.assertEquals(table.coreOptions().partitionDefaultName(),
@@ -817,6 +821,40 @@ public class PaimonBackendBoundTableTest {
return new PaimonTableHandle("db", "tbl", Collections.emptyList(),
Collections.emptyList());
}
+ /**
+ * A write whose static partition values need no cast, so the handle's
default cast spec applies. Only an
+ * overwrite resolves the static partition.
+ */
+ private static ConnectorWriteHandle writeHandle(boolean overwrite,
Map<String, String> staticPartition,
+ Set<String> nullKeys) {
+ return new ConnectorWriteHandle() {
+ @Override
+ public ConnectorTableHandle getTableHandle() {
+ return null;
+ }
+
+ @Override
+ public List<ConnectorColumn> getColumns() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isOverwrite() {
+ return overwrite;
+ }
+
+ @Override
+ public Map<String, String> getStaticPartitionSpec() {
+ return staticPartition;
+ }
+
+ @Override
+ public Set<String> getStaticPartitionNullKeys() {
+ return nullKeys;
+ }
+ };
+ }
+
private static PaimonTableHandle sysHandle(String sysTableType,
FileStoreTable base) {
PaimonTableHandle handle = PaimonTableHandle.forSystemTable("db",
"tbl", sysTableType,
PaimonScanParams.requiresPaimonReader(sysTableType));
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
index 34b46e29fcf..cb601a6ceba 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorTransactionTest.java
@@ -17,7 +17,10 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.spi.ConnectorColumn;
import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
import org.apache.doris.thrift.TPaimonCommitMessage;
import org.apache.paimon.CoreOptions;
@@ -204,10 +207,35 @@ public class PaimonConnectorTransactionTest {
new PaimonConnectorTransaction(TRANSACTION_ID, new
RecordingConnectorContext());
transaction.bind(PaimonWriteBinding.create(
new PaimonTableHandle("db", "tbl", Collections.emptyList(),
Collections.emptyList()),
- target, Collections.emptyMap(), false, Collections.emptyMap(),
Collections.emptySet()));
+ target, Collections.emptyMap(), plainInsert()));
return transaction;
}
+ /** An INSERT without a PARTITION clause. */
+ private static ConnectorWriteHandle plainInsert() {
+ return new ConnectorWriteHandle() {
+ @Override
+ public ConnectorTableHandle getTableHandle() {
+ return null;
+ }
+
+ @Override
+ public List<ConnectorColumn> getColumns() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isOverwrite() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> getStaticPartitionSpec() {
+ return Collections.emptyMap();
+ }
+ };
+ }
+
/** A DPCM-framed payload with one empty commit message, as the BE writer
reports it. */
private static byte[] commitPayload(int bucket) throws Exception {
CommitMessage message = new CommitMessageImpl(BinaryRow.EMPTY_ROW,
bucket, 2,
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
index 6cb05d51810..ca2f37d8776 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonWriteBindingTest.java
@@ -17,38 +17,73 @@
package org.apache.doris.connector.paimon;
+import org.apache.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.connector.spi.handle.ConnectorWriteHandle;
+
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.Timestamp;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.TypeUtils;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
+import java.time.Instant;
+import java.time.OffsetDateTime;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.function.Supplier;
/**
- * Pins how {@link PaimonWriteBinding#configureTableForWrite} maps a Doris
INSERT OVERWRITE onto Paimon's
- * {@code dynamic-partition-overwrite} option. Paimon defaults that option to
true, which replaces only the
- * partitions the new rows reach; a Doris full-table overwrite replaces the
whole table and a static-partition
- * overwrite replaces exactly the named partition.
+ * How a static partition reaches Paimon's static overwrite. The written rows
carry each PARTITION literal cast
+ * to its column type, so the overwrite must name the partition by that cast
value, in the form Paimon parses:
+ * a value spelled differently selects another partition, or none, and the
commit then fails or replaces the
+ * wrong rows.
*/
public class PaimonWriteBindingTest {
+ private static final String DEFAULT_NAME = "__CUSTOM_DEFAULT__";
private static final String DYNAMIC_OVERWRITE =
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key();
- @TempDir
- Path warehouse;
+ private TimeZone jvmZone;
+
+ @BeforeEach
+ public void pinJvmZone() {
+ // Paimon parses a static LTZ value in the JVM zone. Keep it away from
UTC, the zone of the cast values,
+ // so a value passed through unchanged cannot pass by accident.
+ jvmZone = TimeZone.getDefault();
+ TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"));
+ }
+
+ @AfterEach
+ public void restoreJvmZone() {
+ TimeZone.setDefault(jvmZone);
+ }
+
+ // configureTableForWrite maps a Doris INSERT OVERWRITE onto Paimon's
dynamic-partition-overwrite option. Paimon
+ // defaults that option to true, which replaces only the partitions the
new rows reach; a Doris full-table
+ // overwrite replaces the whole table and a static-partition overwrite
replaces exactly the named partition.
@Test
- public void fullOverwriteUsesStaticSemanticsByDefault() {
- FileStoreTable table = partitionedTable(Collections.emptyMap());
+ public void fullOverwriteUsesStaticSemanticsByDefault(@TempDir Path
warehouse) {
+ FileStoreTable table = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "pt", DataTypes.STRING()));
FileStoreTable configured = PaimonWriteBinding.configureTableForWrite(
table, true, Collections.emptyMap());
@@ -59,20 +94,22 @@ public class PaimonWriteBindingTest {
}
@Test
- public void fullOverwriteHonorsExplicitDynamicSemantics() {
+ public void fullOverwriteHonorsExplicitDynamicSemantics(@TempDir Path
warehouse) {
// A table that sets the option itself asked for dynamic overwrite, so
the binding keeps the
// loaded table instead of copying it with the static override.
- FileStoreTable table =
partitionedTable(Collections.singletonMap(DYNAMIC_OVERWRITE, "true"));
+ FileStoreTable table = partitionedTable(warehouse,
Collections.singletonMap(DYNAMIC_OVERWRITE, "true"),
+ new DataField(1, "pt", DataTypes.STRING()));
Assertions.assertSame(table, PaimonWriteBinding.configureTableForWrite(
table, true, Collections.emptyMap()));
}
@Test
- public void staticPartitionOverwriteAlwaysUsesStaticSemantics() {
+ public void staticPartitionOverwriteAlwaysUsesStaticSemantics(@TempDir
Path warehouse) {
// A static partition names the partition to replace. Dynamic
semantics would leave it intact
// when the query produces no row for it, so the table option does not
apply here.
- FileStoreTable table =
partitionedTable(Collections.singletonMap(DYNAMIC_OVERWRITE, "true"));
+ FileStoreTable table = partitionedTable(warehouse,
Collections.singletonMap(DYNAMIC_OVERWRITE, "true"),
+ new DataField(1, "pt", DataTypes.STRING()));
FileStoreTable configured = PaimonWriteBinding.configureTableForWrite(
table, true, Collections.singletonMap("pt", "1"));
@@ -81,11 +118,197 @@ public class PaimonWriteBindingTest {
Assertions.assertFalse(configured.coreOptions().dynamicPartitionOverwrite());
}
- private FileStoreTable partitionedTable(Map<String, String> options) {
- TableSchema schema = new TableSchema(0L,
- Arrays.asList(new DataField(0, "id", DataTypes.INT()), new
DataField(1, "pt", DataTypes.STRING())),
- 1, Collections.singletonList("pt"), Collections.emptyList(),
options, "");
+ @Test
+ public void staticPartitionUsesTheValueCastToTheColumnType(@TempDir Path
warehouse) {
+ // PARTITION (part = true) on an INT column and PARTITION (dt =
20240101) on a DATE column: Paimon reads
+ // "true" as no INT and "20240101" as the 20240101st day after the
epoch. MUTATION: resolving from the spec
+ // as written -> both red.
+ FileStoreTable intTable = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "part", DataTypes.INT()));
+ Assertions.assertEquals("1",
PaimonWriteBinding.resolveStaticPartition(intTable,
+ write(true, "part", "true", "1")).get("part"));
+
+ FileStoreTable dateTable = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "dt", DataTypes.DATE()));
+ Assertions.assertEquals("2024-01-01",
PaimonWriteBinding.resolveStaticPartition(dateTable,
+ write(true, "dt", "20240101", "2024-01-01")).get("dt"));
+ }
+
+ @Test
+ public void staticOverwriteRejectsTheDefaultPartitionName(@TempDir Path
warehouse) {
+ // Paimon's static overwrite reads partition.default-name as the NULL
partition of every type, so this
+ // value would replace the NULL partition. MUTATION: dropping the
check -> red.
+ FileStoreTable table = partitionedTable(warehouse,
+ Collections.singletonMap("partition.default-name",
DEFAULT_NAME),
+ new DataField(1, "part", DataTypes.STRING()));
+
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> PaimonWriteBinding.resolveStaticPartition(table,
+ write(true, "part", DEFAULT_NAME, DEFAULT_NAME)));
+ Assertions.assertTrue(error.getMessage().contains("cannot be
represented"), error.getMessage());
+ }
+
+ @Test
+ public void staticLtzPartitionMovesTheInstantToTheSdkZone(@TempDir Path
warehouse) {
+ // Doris binds an LTZ column as TIMESTAMPTZ, so the cast value is the
instant in UTC with its offset,
+ // which Paimon's parser does not take: it reads an offset-free value
as local time in the JVM zone.
+ // MUTATION: passing the value through, or dropping its offset without
moving it -> red.
+ FileStoreTable table = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "part",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)));
+
+ String value = PaimonWriteBinding.resolveStaticPartition(table,
+ write(true, "part", "2024-01-15 08:30:45.123456", "2024-01-15
00:30:45.123456+00:00")).get("part");
+
+ Assertions.assertEquals("2024-01-14 16:30:45.123456", value);
+ Assertions.assertEquals(Instant.parse("2024-01-15T00:30:45.123456Z"),
parsedByPaimon(table, value));
+ }
+
+ @Test
+ public void staticLtzOverwriteRejectsAnInstantTheSdkZoneRepeats(@TempDir
Path warehouse) {
+ // 2023-11-05 08:30Z and 09:30Z are both 01:30 in Los Angeles, the JVM
zone here. Either value would name
+ // the same partition, which Paimon reads as the earlier instant.
MUTATION: dropping the check -> red.
+ FileStoreTable table = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "part",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)));
+ for (String instant : Arrays.asList("2023-11-05 08:30:00.123456+00:00",
+ "2023-11-05 09:30:00.123456+00:00")) {
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> PaimonWriteBinding.resolveStaticPartition(table,
write(true, "part", instant, instant)));
+ Assertions.assertTrue(error.getMessage().contains("ambiguous"),
error.getMessage());
+
Assertions.assertTrue(error.getMessage().contains("America/Los_Angeles"),
error.getMessage());
+ }
+ }
+
+ @Test
+ public void
staticLtzPartitionOutsideAnOverlapRoundTripsThroughPaimon(@TempDir Path
warehouse) {
+ // Both instants that overlap in Los Angeles are distinct in UTC, and
an instant after the overlap is
+ // distinct in Los Angeles. MUTATION: rejecting every instant near a
DST change -> red.
+ FileStoreTable table = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "part",
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(6)));
+ TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
+ for (String instant : Arrays.asList("2023-11-05 08:30:00.123456+00:00",
+ "2023-11-05 09:30:00.123456+00:00")) {
+ assertStaticLtzRoundTrip(table, instant);
+ }
+ TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"));
+ assertStaticLtzRoundTrip(table, "2023-11-05 10:30:00.123456+00:00");
+ }
+
+ @Test
+ public void staticOverwriteRejectsABinaryPartitionValue(@TempDir Path
warehouse) {
+ // Paimon's static overwrite takes the UTF-8 bytes of a value's
string: "0x4142" names the partition of
+ // those six bytes, not of the two bytes 41 42 the rows carry, so an
overwrite without rows would report
+ // success and leave the partition it was given in place. The check
comes before the values are cast,
+ // which a binary value may not survive. MUTATION: dropping the check
-> red.
+ Assertions.assertArrayEquals("0x4142".getBytes(StandardCharsets.UTF_8),
+ (byte[]) TypeUtils.castFromString("0x4142",
DataTypes.BYTES()));
+ for (DataType type : Arrays.asList(DataTypes.BYTES(),
DataTypes.VARBINARY(2), DataTypes.BINARY(2))) {
+ FileStoreTable table = partitionedTable(warehouse,
Collections.emptyMap(),
+ new DataField(1, "part", type));
+ ConnectorWriteHandle overwrite = handle(true,
Collections.singletonMap("part", "0x4142"), () -> {
+ throw new AssertionError("a binary static partition value must
be rejected before it is cast");
+ }, Collections.emptySet());
+
+ DorisConnectorException error =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> PaimonWriteBinding.resolveStaticPartition(table,
overwrite));
+ Assertions.assertTrue(error.getMessage().contains("binary
partition column 'part'"),
+ error.getMessage());
+ }
+ }
+
+ @Test
+ public void staticPartitionNullUsesTheDefaultPartitionName(@TempDir Path
warehouse) {
+ // A binary column too: SQL NULL names its partition by
partition.default-name, not by bytes.
+ for (DataType type : Arrays.asList(DataTypes.INT(),
DataTypes.BYTES())) {
+ FileStoreTable table = partitionedTable(warehouse,
+ Collections.singletonMap("partition.default-name",
DEFAULT_NAME),
+ new DataField(1, "part", type));
+ ConnectorWriteHandle write = handle(true,
Collections.singletonMap("part", "NULL"),
+ Collections::emptyMap, Collections.singleton("part"));
+
+ Assertions.assertEquals(DEFAULT_NAME,
+ PaimonWriteBinding.resolveStaticPartition(table,
write).get("part"), type.toString());
+ }
+ }
+
+ @Test
+ public void insertLeavesTheStaticPartitionValuesAlone(@TempDir Path
warehouse) {
+ // Only a static overwrite reads the static partition. An INSERT
writes each row to the partition its own
+ // values name, so a value the cast rejects (a row keeps it as NULL,
or as a fitting binary) or an
+ // overwrite cannot represent must not fail it. MUTATION: resolving
the static partition of an INSERT
+ // -> red.
+ FileStoreTable table = partitionedTable(warehouse,
+ Collections.singletonMap("partition.default-name",
DEFAULT_NAME),
+ new DataField(1, "part", DataTypes.STRING()));
+ ConnectorWriteHandle insert = handle(false,
Collections.singletonMap("part", DEFAULT_NAME), () -> {
+ throw new AssertionError("an INSERT must not cast its static
partition values");
+ }, Collections.emptySet());
+
+ PaimonWriteBinding binding = PaimonWriteBinding.create(
+ new PaimonTableHandle("db", "tbl", Collections.emptyList(),
Collections.emptyList()),
+ table, Collections.emptyMap(), insert);
+
+ Assertions.assertFalse(binding.isOverwrite());
+ Assertions.assertTrue(binding.getStaticPartition().isEmpty(),
binding.getStaticPartition().toString());
+ }
+
+ private static void assertStaticLtzRoundTrip(FileStoreTable table, String
instant) {
+ String value = PaimonWriteBinding.resolveStaticPartition(table,
write(true, "part", instant, instant))
+ .get("part");
+ Assertions.assertEquals(OffsetDateTime.parse(instant.replace(' ',
'T')).toInstant(),
+ parsedByPaimon(table, value), instant);
+ }
+
+ private static Instant parsedByPaimon(FileStoreTable table, String value) {
+ return ((Timestamp) TypeUtils.castFromString(value,
table.rowType().getTypeAt(1))).toInstant();
+ }
+
+ private static FileStoreTable partitionedTable(Path warehouse, Map<String,
String> options,
+ DataField partition) {
+ List<DataField> fields = Arrays.asList(new DataField(0, "id",
DataTypes.INT()), partition);
+ TableSchema schema = new TableSchema(0L, fields, 1,
Collections.singletonList(partition.name()),
+ Collections.emptyList(), options, "");
return FileStoreTableFactory.create(LocalFileIO.create(),
new org.apache.paimon.fs.Path("file://" + warehouse +
"/db.db/tbl"), schema);
}
+
+ private static ConnectorWriteHandle write(boolean overwrite, String name,
String writtenValue,
+ String castValue) {
+ return handle(overwrite, Collections.singletonMap(name, writtenValue),
+ () -> Collections.singletonMap(name, castValue),
Collections.emptySet());
+ }
+
+ private static ConnectorWriteHandle handle(boolean overwrite, Map<String,
String> spec,
+ Supplier<Map<String, String>> castSpec, Set<String> nullKeys) {
+ return new ConnectorWriteHandle() {
+ @Override
+ public ConnectorTableHandle getTableHandle() {
+ return null;
+ }
+
+ @Override
+ public List<ConnectorColumn> getColumns() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isOverwrite() {
+ return overwrite;
+ }
+
+ @Override
+ public Map<String, String> getStaticPartitionSpec() {
+ return spec;
+ }
+
+ @Override
+ public Set<String> getStaticPartitionNullKeys() {
+ return nullKeys;
+ }
+
+ @Override
+ public Map<String, String> getCastStaticPartitionSpec() {
+ return castSpec.get();
+ }
+ };
+ }
}
diff --git
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandle.java
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandle.java
index 79b0d489ccc..db7234d92a2 100644
---
a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandle.java
+++
b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandle.java
@@ -21,6 +21,7 @@ import org.apache.doris.connector.spi.ConnectorColumn;
import org.apache.doris.thrift.TSortInfo;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -75,6 +76,22 @@ public interface ConnectorWriteHandle {
return Collections.emptySet();
}
+ /**
+ * The static partition values cast to the types of their columns, as the
written rows carry them (a string
+ * longer than its CHAR / VARCHAR column is cut as the INSERT cuts it), in
the string form of the cast value
+ * and keyed like {@link #getStaticPartitionSpec}, which keeps each
literal as written:
+ * {@code PARTITION (p = true)} on an INT column is {@code "true"} there
and {@code "1"} here. A binary value
+ * is {@code 0x} followed by its bytes in hex, as in {@link
#getStaticPartitionSpec}. The
+ * {@link #getStaticPartitionNullKeys null keys} are absent, and a key
that names no column (an Iceberg
+ * partition field) keeps its literal. A value that cannot be cast to its
column type fails the call. The
+ * default, for a handle that does not cast, is the spec as written
without its null keys.
+ */
+ default Map<String, String> getCastStaticPartitionSpec() {
+ Map<String, String> spec = new
LinkedHashMap<>(getStaticPartitionSpec());
+ spec.keySet().removeAll(getStaticPartitionNullKeys());
+ return spec;
+ }
+
/**
* The kind of DML write (INSERT / OVERWRITE / DELETE / UPDATE / MERGE). A
single
* {@code planWrite} dispatches on this to pick the connector's Thrift
sink dialect, and a
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandleTest.java
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandleTest.java
index b4bff4173fc..1027c0f8a30 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandleTest.java
+++
b/fe/fe-connector/fe-connector-spi/src/test/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandleTest.java
@@ -23,9 +23,11 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
/**
* Pins the {@link ConnectorWriteHandle#getWriteOperation()} default
(P6.3-T03, deferred from T01).
@@ -39,7 +41,7 @@ import java.util.Optional;
public class ConnectorWriteHandleTest {
/** Minimal handle that overrides nothing write-op related, to read the
SPI default. */
- private static final class BareWriteHandle implements ConnectorWriteHandle
{
+ private static class BareWriteHandle implements ConnectorWriteHandle {
@Override
public ConnectorTableHandle getTableHandle() {
return null;
@@ -99,4 +101,26 @@ public class ConnectorWriteHandleTest {
Assertions.assertEquals(Optional.empty(), new
BareWriteHandle().getBranchName(),
"a write handle that declares no branch must default to
Optional.empty()");
}
+
+ @Test
+ public void castStaticPartitionSpecDefaultsToTheSpecWithoutNullKeys() {
+ // A handle that does not cast hands its values on as written, and a
SQL NULL value is a null key only.
+ Map<String, String> spec = new LinkedHashMap<>();
+ spec.put("pt", "20240101");
+ spec.put("region", "NULL");
+ ConnectorWriteHandle handle = new BareWriteHandle() {
+ @Override
+ public Map<String, String> getStaticPartitionSpec() {
+ return spec;
+ }
+
+ @Override
+ public Set<String> getStaticPartitionNullKeys() {
+ return Collections.singleton("region");
+ }
+ };
+
+ Assertions.assertEquals(Collections.singletonMap("pt", "20240101"),
handle.getCastStaticPartitionSpec());
+ Assertions.assertEquals(2, spec.size(), "the default must not modify
the spec");
+ }
}
diff --git
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
index f0ae0af27d1..3f9e7f2703d 100644
---
a/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
+++
b/fe/fe-connector/fe-connector-spi/src/test/resources/connector-plugin-surface.txt
@@ -118,6 +118,7 @@
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBeExecVersion():in
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBoundTargetColumns():java.util.List
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBoundWriteMetadataIdentity():java.lang.String
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getBranchName():java.util.Optional
+org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getCastStaticPartitionSpec():java.util.Map
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getColumns():java.util.List
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getSortInfo():org.apache.doris.thrift.TSortInfo
org.apache.doris.connector.spi.handle.ConnectorWriteHandle#getStaticPartitionNullKeys():java.util.Set
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
index a426a76dddc..5a235fd995b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
@@ -312,10 +312,7 @@ public class BindSink implements AnalysisRuleFactory {
// add cast project
List<NamedExpression> castExprs = Lists.newArrayList();
- ConnectContext connCtx = ConnectContext.get();
- final boolean truncateString = needTruncateStringWhenInsert
- && (connCtx == null ||
connCtx.getSessionVariable().enableInsertValueAutoCast)
- && !SessionVariable.enableStrictCast();
+ final boolean truncateString = needTruncateStringWhenInsert &&
truncatesStringOnInsert();
for (int i = 0; i < tableSchema.size(); ++i) {
Column col = tableSchema.get(i);
NamedExpression expr = columnToOutput.get(col.getName()); //
relative outputExpr
@@ -361,6 +358,16 @@ public class BindSink implements AnalysisRuleFactory {
return fullOutputProject;
}
+ /**
+ * Whether the session lets an INSERT write a string into a shorter CHAR /
VARCHAR column cut to the column
+ * length, counted in code points (the {@link Substring} above), instead
of as is.
+ */
+ public static boolean truncatesStringOnInsert() {
+ ConnectContext connCtx = ConnectContext.get();
+ return (connCtx == null ||
connCtx.getSessionVariable().enableInsertValueAutoCast)
+ && !SessionVariable.enableStrictCast();
+ }
+
@VisibleForTesting
static Expression coerceSinkExpression(Expression expression, DataType
targetType) {
return TypeCoercionUtils.castIfNotSameType(expression, targetType);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
index ea963183659..db001e34998 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -631,6 +631,7 @@ public class InsertIntoTableCommand extends Command
// commit at the branch (iceberg validates it in beginWrite).
The guard above already
// rejected @branch for connectors without
supportsWriteBranch().
branchName.ifPresent(notUsed ->
pluginCtx.setBranchName(branchName));
+
pluginCtx.setBoundTargetSchema(((PhysicalConnectorTableSink<?>)
physicalSink).getBoundTargetSchema());
if (pluginCtx.getStaticPartitionSpec().isEmpty()
&& originLogicalQuery instanceof
UnboundConnectorTableSink) {
UnboundConnectorTableSink<?> pluginSink =
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java
index e0921f7d4d5..7d96149086a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java
@@ -19,14 +19,23 @@ package
org.apache.doris.nereids.trees.plans.commands.insert;
import org.apache.doris.catalog.Column;
import
org.apache.doris.datasource.connector.converter.ConnectorWriteValueConverter;
+import org.apache.doris.nereids.rules.analysis.BindSink;
import org.apache.doris.nereids.trees.expressions.Expression;
+import
org.apache.doris.nereids.trees.expressions.functions.executable.StringArithmetic;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.coercion.CharacterType;
+
+import com.google.common.base.Preconditions;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -50,18 +59,15 @@ public class PluginDrivenInsertCommandContext extends
BaseExternalTableInsertCom
private Map<String, String> staticPartitionSpec = Collections.emptyMap();
private Set<String> staticPartitionNullKeys = Collections.emptySet();
+ private Map<String, Literal> staticPartitionLiterals =
Collections.emptyMap();
+ // The target schema the sink was bound to, whose column types the static
partition values are cast to.
+ private List<Column> boundTargetSchema = Collections.emptyList();
private Optional<String> branchName = Optional.empty();
public Map<String, String> getStaticPartitionSpec() {
return staticPartitionSpec;
}
- public void setStaticPartitionSpec(Map<String, String>
staticPartitionSpec) {
- this.staticPartitionSpec =
- staticPartitionSpec == null ? Collections.emptyMap() :
staticPartitionSpec;
- this.staticPartitionNullKeys = Collections.emptySet();
- }
-
public Set<String> getStaticPartitionNullKeys() {
return staticPartitionNullKeys;
}
@@ -75,6 +81,7 @@ public class PluginDrivenInsertCommandContext extends
BaseExternalTableInsertCom
public void setStaticPartitionSpecFromExpressions(Map<String, Expression>
partitionValues, List<Column> schema) {
Map<String, String> spec = new HashMap<>();
Set<String> nullKeys = new HashSet<>();
+ Map<String, Literal> literals = new LinkedHashMap<>();
for (Map.Entry<String, Expression> entry : partitionValues.entrySet())
{
if (entry.getValue() instanceof Literal) {
Literal literal = (Literal) entry.getValue();
@@ -84,9 +91,8 @@ public class PluginDrivenInsertCommandContext extends
BaseExternalTableInsertCom
// Commit metadata must use the same semantic bytes as the
materialized partition row.
literal = (Literal)
ConnectorWriteValueConverter.convert(column, literal);
}
- // Binary partition keys must bypass character decoding,
including empty and non-UTF-8 bytes.
- spec.put(entry.getKey(), literal instanceof VarBinaryLiteral
- ? "0x" + literal.toString() :
literal.getStringValue());
+ spec.put(entry.getKey(), specValue(literal));
+ literals.put(entry.getKey(), literal);
if (entry.getValue() instanceof NullLiteral) {
nullKeys.add(entry.getKey());
}
@@ -94,6 +100,60 @@ public class PluginDrivenInsertCommandContext extends
BaseExternalTableInsertCom
}
this.staticPartitionSpec = spec;
this.staticPartitionNullKeys = nullKeys;
+ this.staticPartitionLiterals = literals;
+ }
+
+ public void setBoundTargetSchema(List<Column> boundTargetSchema) {
+ this.boundTargetSchema = boundTargetSchema;
+ }
+
+ /**
+ * Casts the static partition values to the types of their columns in the
bound target schema; see
+ * {@code ConnectorWriteHandle#getCastStaticPartitionSpec}. Each value is
the one BindSink materializes into
+ * the rows, after constant folding, but a value that does not fit its
type fails here instead of becoming
+ * NULL.
+ */
+ public Map<String, String> castStaticPartitionSpec() {
+ Preconditions.checkState(staticPartitionLiterals.isEmpty() ||
!boundTargetSchema.isEmpty(),
+ "static partition values are cast before the sink is bound to
a target schema");
+ Map<String, String> spec = new LinkedHashMap<>();
+ for (Map.Entry<String, Literal> entry :
staticPartitionLiterals.entrySet()) {
+ if (staticPartitionNullKeys.contains(entry.getKey())) {
+ continue;
+ }
+ Literal value = entry.getValue();
+ Optional<Column> column = boundTargetSchema.stream()
+ .filter(candidate ->
candidate.getName().equalsIgnoreCase(entry.getKey()))
+ .findFirst();
+ if (column.isPresent()) {
+ value = writtenValue(value,
DataType.fromCatalogType(column.get().getType()));
+ }
+ spec.put(entry.getKey(), specValue(value));
+ }
+ return spec;
+ }
+
+ private static String specValue(Literal literal) {
+ // Binary partition keys must bypass character decoding, including
empty and non-UTF-8 bytes.
+ return literal instanceof VarBinaryLiteral ? "0x" + literal.toString()
: literal.getStringValue();
+ }
+
+ private static Literal writtenValue(Literal value, DataType columnType) {
+ if (!value.getDataType().isStringLikeType() ||
!columnType.isStringLikeType()) {
+ Expression cast = value.checkedCastTo(columnType);
+ Preconditions.checkState(cast instanceof Literal, "cast of literal
%s is not a literal", value);
+ return (Literal) cast;
+ }
+ // BindSink does not cast a string written into a string column. It
keeps the value, cut to the length of
+ // a CHAR / VARCHAR column in code points when the session truncates
strings on insert; a cast would cut
+ // UTF-16 units instead, or reject a CHAR value that is too long.
+ int length = ((CharacterType) columnType).getLen();
+ if (length < 0 || !BindSink.truncatesStringOnInsert()) {
+ return value;
+ }
+ Preconditions.checkState(value instanceof StringLikeLiteral, "string
literal %s has no string value", value);
+ return (Literal)
StringArithmetic.substringVarcharIntInt((StringLikeLiteral) value,
+ new IntegerLiteral(1), new IntegerLiteral(length));
}
public Optional<String> getBranchName() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/PluginDrivenTableSink.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PluginDrivenTableSink.java
index fbbc7bfcaa3..32fe4c24e54 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/planner/PluginDrivenTableSink.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/planner/PluginDrivenTableSink.java
@@ -46,6 +46,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Supplier;
/**
* Generic data sink for plugin-driven external tables.
@@ -147,7 +148,7 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
// EXPLAIN), so the connector derives the detail from the write handle.
ConnectorWriteHandle handle = new PluginDrivenWriteHandle(
tableHandle, connectorColumns, boundTargetColumns, false,
Collections.emptyMap(),
- Collections.emptySet(), null, null, Optional.empty(),
writeOperation,
+ Collections.emptySet(), Collections::emptyMap, null, null,
Optional.empty(), writeOperation,
writesDataFiles, requireMergeCardinalityCheck, beExecVersion);
writePlanProvider.appendExplainInfo(sb, prefix, connectorSession,
handle);
return sb.toString();
@@ -175,12 +176,14 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
boolean overwrite = false;
Map<String, String> writeContext = Collections.emptyMap();
Set<String> staticPartitionNullKeys = Collections.emptySet();
+ Supplier<Map<String, String>> castStaticPartitionSpec =
Collections::emptyMap;
Optional<String> branchName = Optional.empty();
if (insertCtx.isPresent() && insertCtx.get() instanceof
PluginDrivenInsertCommandContext) {
PluginDrivenInsertCommandContext ctx =
(PluginDrivenInsertCommandContext) insertCtx.get();
overwrite = ctx.isOverwrite();
writeContext = ctx.getStaticPartitionSpec();
staticPartitionNullKeys = ctx.getStaticPartitionNullKeys();
+ castStaticPartitionSpec = ctx::castStaticPartitionSpec;
branchName = ctx.getBranchName();
}
ConnectorTableHandle boundTableHandle = tableHandle;
@@ -195,7 +198,7 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
}
ConnectorWriteHandle handle = new PluginDrivenWriteHandle(
boundTableHandle, connectorColumns, boundTargetColumns,
overwrite, writeContext,
- staticPartitionNullKeys, writeSortInfo,
+ staticPartitionNullKeys, castStaticPartitionSpec,
writeSortInfo,
boundWriteMetadataIdentity, branchName, writeOperation,
writesDataFiles,
requireMergeCardinalityCheck, beExecVersion);
ConnectorSinkPlan sinkPlan =
writePlanProvider.planWrite(connectorSession, handle);
@@ -230,6 +233,9 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
private final boolean overwrite;
private final Map<String, String> writeContext;
private final Set<String> staticPartitionNullKeys;
+ // Computed on demand, so only a connector that reads the cast values
can fail on a value that does not
+ // fit its column type.
+ private final Supplier<Map<String, String>> castStaticPartitionSpec;
private final TSortInfo sortInfo;
private final String boundWriteMetadataIdentity;
private final Optional<String> branchName;
@@ -241,6 +247,7 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
private PluginDrivenWriteHandle(ConnectorTableHandle tableHandle,
List<ConnectorColumn> columns,
List<ConnectorColumn> boundTargetColumns, boolean overwrite,
Map<String, String> writeContext, Set<String>
staticPartitionNullKeys,
+ Supplier<Map<String, String>> castStaticPartitionSpec,
TSortInfo sortInfo, String boundWriteMetadataIdentity,
Optional<String> branchName, WriteOperation writeOperation,
boolean writesDataFiles, boolean requireMergeCardinalityCheck,
@@ -251,6 +258,7 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
this.overwrite = overwrite;
this.writeContext = writeContext;
this.staticPartitionNullKeys = staticPartitionNullKeys;
+ this.castStaticPartitionSpec = castStaticPartitionSpec;
this.sortInfo = sortInfo;
this.boundWriteMetadataIdentity = boundWriteMetadataIdentity;
this.branchName = branchName == null ? Optional.empty() :
branchName;
@@ -320,6 +328,11 @@ public class PluginDrivenTableSink extends
BaseExternalTableDataSink {
return staticPartitionNullKeys;
}
+ @Override
+ public Map<String, String> getCastStaticPartitionSpec() {
+ return castStaticPartitionSpec.get();
+ }
+
@Override
public WriteOperation getWriteOperation() {
return writeOperation;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/PluginDrivenTableSinkBindingTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/PluginDrivenTableSinkBindingTest.java
index 120bfbb2c96..2da58e50325 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/PluginDrivenTableSinkBindingTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/PluginDrivenTableSinkBindingTest.java
@@ -18,6 +18,9 @@
package org.apache.doris.planner;
import org.apache.doris.analysis.TableScanParams;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.connector.ConnectorSessionBuilder;
@@ -32,7 +35,16 @@ import
org.apache.doris.connector.spi.write.ConnectorWritePlanProvider;
import org.apache.doris.datasource.mvcc.MvccUtil;
import org.apache.doris.datasource.mvcc.PluginDrivenMvccSnapshot;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.DecimalV3Literal;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
import
org.apache.doris.nereids.trees.plans.commands.insert.PluginDrivenInsertCommandContext;
+import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TDataSink;
import org.junit.jupiter.api.Assertions;
@@ -40,8 +52,11 @@ import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.math.BigDecimal;
+import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -69,9 +84,7 @@ public class PluginDrivenTableSinkBindingTest {
PluginDrivenInsertCommandContext ctx = new
PluginDrivenInsertCommandContext();
ctx.setOverwrite(true);
- Map<String, String> staticSpec = new HashMap<>();
- staticSpec.put("pt", "20240101");
- ctx.setStaticPartitionSpec(staticSpec);
+
ctx.setStaticPartitionSpecFromExpressions(Collections.singletonMap("pt", new
StringLiteral("20240101")));
sink.bindDataSink(Optional.of(ctx));
@@ -79,10 +92,145 @@ public class PluginDrivenTableSinkBindingTest {
Assertions.assertNotNull(handle, "planWrite must be invoked with a
bound write handle");
Assertions.assertTrue(handle.isOverwrite(),
"INSERT OVERWRITE must propagate ctx.isOverwrite()=true to the
connector write handle");
- Assertions.assertEquals(staticSpec, handle.getStaticPartitionSpec(),
+ Assertions.assertEquals(Collections.singletonMap("pt", "20240101"),
handle.getStaticPartitionSpec(),
"PARTITION(col=val) must propagate the static partition spec
to the write handle");
}
+ @Test
+ public void staticPartitionValuesCastToTheirColumnTypesFlowToWriteHandle()
throws AnalysisException {
+ // The rows carry each PARTITION literal cast to its column type; a
connector that names the partition
+ // by value (Paimon's static overwrite) must get that value, not the
literal as written. MUTATION:
+ // returning the spec as written from getCastStaticPartitionSpec ->
red.
+ RecordingWritePlanProvider provider = new RecordingWritePlanProvider();
+ PluginDrivenTableSink sink = newPlanProviderSink(provider);
+ PluginDrivenInsertCommandContext ctx = new
PluginDrivenInsertCommandContext();
+ Map<String, Expression> partition = new LinkedHashMap<>();
+ partition.put("P", BooleanLiteral.TRUE);
+ partition.put("dt", new IntegerLiteral(20240101));
+ partition.put("amount", new DecimalV3Literal(new BigDecimal("1.5")));
+ partition.put("region", new NullLiteral());
+ partition.put("p_bucket", new StringLiteral("x"));
+ ctx.setStaticPartitionSpecFromExpressions(partition);
+ ctx.setBoundTargetSchema(Arrays.asList(new Column("p", Type.INT), new
Column("dt", Type.DATEV2),
+ new Column("amount", ScalarType.createDecimalV3Type(10, 2)),
new Column("region", Type.STRING)));
+
+ sink.bindDataSink(Optional.of(ctx));
+
+ ConnectorWriteHandle handle = provider.capturedHandle;
+ Assertions.assertEquals("20240101",
handle.getStaticPartitionSpec().get("dt"));
+ Map<String, String> expected = new LinkedHashMap<>();
+ expected.put("P", "1");
+ expected.put("dt", "2024-01-01");
+ expected.put("amount", "1.50");
+ // A key that names no column, such as an Iceberg partition field,
keeps its literal; SQL NULL stays a null key.
+ expected.put("p_bucket", "x");
+ Assertions.assertEquals(expected, handle.getCastStaticPartitionSpec());
+ Assertions.assertEquals(Collections.singleton("region"),
handle.getStaticPartitionNullKeys());
+ }
+
+ @Test
+ public void staticPartitionDateTimeValuesCastInTheSessionZone() throws
AnalysisException {
+ // A DATETIMEV2 value stays local time in the session zone, and a
TIMESTAMPTZ value becomes the instant
+ // in UTC with its offset; Paimon moves either one to the zone its
overwrite parser uses.
+ ConnectContext previous = ConnectContext.get();
+ ConnectContext context = new ConnectContext();
+ context.getSessionVariable().setTimeZone("Asia/Shanghai");
+ context.setThreadLocalInfo();
+ try {
+ RecordingWritePlanProvider provider = new
RecordingWritePlanProvider();
+ PluginDrivenTableSink sink = newPlanProviderSink(provider);
+ PluginDrivenInsertCommandContext ctx = new
PluginDrivenInsertCommandContext();
+ Map<String, Expression> partition = new LinkedHashMap<>();
+ partition.put("ts", new StringLiteral("2024-01-15
08:30:45.123456"));
+ partition.put("tz", new StringLiteral("2024-01-15
08:30:45.123456"));
+ ctx.setStaticPartitionSpecFromExpressions(partition);
+ ctx.setBoundTargetSchema(Arrays.asList(new Column("ts",
ScalarType.createDatetimeV2Type(6)),
+ new Column("tz", ScalarType.createTimeStampTzType(6))));
+
+ sink.bindDataSink(Optional.of(ctx));
+
+ Map<String, String> expected = new LinkedHashMap<>();
+ expected.put("ts", "2024-01-15 08:30:45.123456");
+ expected.put("tz", "2024-01-15 00:30:45.123456+00:00");
+ Assertions.assertEquals(expected,
provider.capturedHandle.getCastStaticPartitionSpec());
+ } finally {
+ if (previous == null) {
+ ConnectContext.remove();
+ } else {
+ previous.setThreadLocalInfo();
+ }
+ }
+ }
+
+ @Test
+ public void staticPartitionStringValueIsCutLikeTheWrittenRows() throws
AnalysisException {
+ // BindSink writes a string into a shorter CHAR / VARCHAR column cut
to the column length in code points,
+ // not cast: a cast rejects the CHAR value and splits the emoji's
surrogate pair, so the overwrite would
+ // fail or name another partition. MUTATION: casting string values
like the others -> red.
+ ConnectContext previous = ConnectContext.get();
+ ConnectContext context = new ConnectContext();
+ context.setThreadLocalInfo();
+ try {
+ // One code point, two UTF-16 units.
+ String emoji = new String(Character.toChars(0x1F600));
+ Map<String, Expression> partition = new LinkedHashMap<>();
+ partition.put("c", new VarcharLiteral("abcd"));
+ partition.put("v", new VarcharLiteral(emoji + "x"));
+ partition.put("s", new VarcharLiteral("abcd"));
+ List<Column> schema = Arrays.asList(new Column("c",
ScalarType.createCharType(2)),
+ new Column("v", ScalarType.createVarcharType(1)), new
Column("s", Type.STRING));
+
+ Map<String, String> expected = new LinkedHashMap<>();
+ expected.put("c", "ab");
+ expected.put("v", emoji);
+ expected.put("s", "abcd");
+ Assertions.assertEquals(expected,
castStaticPartitionSpec(partition, schema));
+
+ // A session that does not truncate writes the strings as they
are, and they stay so here.
+ context.getSessionVariable().enableInsertValueAutoCast = false;
+ Map<String, String> asWritten = new LinkedHashMap<>();
+ asWritten.put("c", "abcd");
+ asWritten.put("v", emoji + "x");
+ asWritten.put("s", "abcd");
+ Assertions.assertEquals(asWritten,
castStaticPartitionSpec(partition, schema));
+ } finally {
+ if (previous == null) {
+ ConnectContext.remove();
+ } else {
+ previous.setThreadLocalInfo();
+ }
+ }
+ }
+
+ @Test
+ public void staticPartitionBinaryValueKeepsItsBytesWhenCast() throws
AnalysisException {
+ // A binary value is 0x and its bytes in hex in both maps. Its string
value would be the identity of the
+ // byte array ([B@...), which names no partition. MUTATION:
getStringValue() for a cast binary value -> red.
+ Map<String, Expression> partition = Collections.singletonMap("b",
+ new VarBinaryLiteral(new byte[] {0x41, 0x42}));
+ // The type Paimon's VARBINARY maps to.
+ List<Column> schema = Collections.singletonList(
+ new Column("b",
ScalarType.createVarbinaryType(ScalarType.MAX_VARBINARY_LENGTH)));
+
+ Assertions.assertEquals(Collections.singletonMap("b", "0x4142"),
castStaticPartitionSpec(partition, schema));
+ }
+
+ @Test
+ public void staticPartitionValueThatDoesNotFitItsTypeFailsOnlyWhenCast()
throws AnalysisException {
+ // Only a connector that reads the cast values may fail on them; the
others keep reading the literal.
+ RecordingWritePlanProvider provider = new RecordingWritePlanProvider();
+ PluginDrivenTableSink sink = newPlanProviderSink(provider);
+ PluginDrivenInsertCommandContext ctx = new
PluginDrivenInsertCommandContext();
+
ctx.setStaticPartitionSpecFromExpressions(Collections.singletonMap("dt", new
StringLiteral("2024-13-45")));
+ ctx.setBoundTargetSchema(Collections.singletonList(new Column("dt",
Type.DATEV2)));
+
+ sink.bindDataSink(Optional.of(ctx));
+
+ ConnectorWriteHandle handle = provider.capturedHandle;
+ Assertions.assertEquals(Collections.singletonMap("dt", "2024-13-45"),
handle.getStaticPartitionSpec());
+ Assertions.assertThrows(RuntimeException.class,
handle::getCastStaticPartitionSpec);
+ }
+
@Test
public void absentContextDefaultsToNonOverwriteEmptySpec() throws
AnalysisException {
RecordingWritePlanProvider provider = new RecordingWritePlanProvider();
@@ -133,6 +281,17 @@ public class PluginDrivenTableSinkBindingTest {
Assertions.assertSame(pinnedHandle,
provider.capturedHandle.getTableHandle());
}
+ private static Map<String, String> castStaticPartitionSpec(Map<String,
Expression> partition,
+ List<Column> boundTargetSchema) throws AnalysisException {
+ RecordingWritePlanProvider provider = new RecordingWritePlanProvider();
+ PluginDrivenTableSink sink = newPlanProviderSink(provider);
+ PluginDrivenInsertCommandContext ctx = new
PluginDrivenInsertCommandContext();
+ ctx.setStaticPartitionSpecFromExpressions(partition);
+ ctx.setBoundTargetSchema(boundTargetSchema);
+ sink.bindDataSink(Optional.of(ctx));
+ return provider.capturedHandle.getCastStaticPartitionSpec();
+ }
+
private static PluginDrivenTableSink
newPlanProviderSink(ConnectorWritePlanProvider provider) {
ConnectorSession session =
ConnectorSessionBuilder.create().withCatalogName("mc_cat").build();
ConnectorTableHandle tableHandle = new ConnectorTableHandle() { };
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
index 0cdbee408e9..6a0db1e9dbc 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy
@@ -91,6 +91,18 @@ suite("test_paimon_write_transaction", "p0,external,paimon")
{
'partition.default-name' = '__CUSTOM_DEFAULT_PARTITION__'
);
+ DROP TABLE IF EXISTS paimon.${dbName}.t_static_cast;
+ CREATE TABLE paimon.${dbName}.t_static_cast (
+ id INT, name STRING, p INT, dt DATE
+ ) USING paimon
+ PARTITIONED BY (p, dt);
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_static_ltz;
+ CREATE TABLE paimon.${dbName}.t_static_ltz (
+ id INT, ts TIMESTAMP
+ ) USING paimon
+ PARTITIONED BY (ts);
+
DROP TABLE IF EXISTS paimon.${dbName}.t_dynamic_multi;
CREATE TABLE paimon.${dbName}.t_dynamic_multi (
id INT, name STRING, region STRING
@@ -273,6 +285,43 @@ suite("test_paimon_write_transaction",
"p0,external,paimon") {
FROM t_static_default ORDER BY id"""
assertTableEquals("t_static_default", "ORDER BY id")
+ // Paimon's static overwrite reads partition.default-name as the NULL
+ // partition, so the same string as a value cannot be overwritten.
+ test {
+ sql """INSERT OVERWRITE TABLE t_static_default
+ PARTITION (region = '__CUSTOM_DEFAULT_PARTITION__') SELECT 1,
'unused' LIMIT 0"""
+ exception "cannot be represented in a static overwrite"
+ }
+ assertTableEquals("t_static_default", "ORDER BY id")
+
+ // A static partition value is cast to its column type, as the written
+ // rows are: PARTITION (p = true) names p = 1 and dt = 20240101 names
+ // 2024-01-01.
+ sql """INSERT INTO t_static_cast VALUES
+ (1, 'p1_d1', 1, '2024-01-01'),
+ (2, 'p1_d2', 1, '2024-01-02'),
+ (3, 'p2_d1', 2, '2024-01-01')"""
+ sql """INSERT OVERWRITE TABLE t_static_cast
+ PARTITION (p = true) VALUES (10, 'p1_new', '2024-01-01')"""
+ assertEquals([[3], [10]], sql("""SELECT id FROM t_static_cast ORDER BY
id"""))
+ sql """INSERT OVERWRITE TABLE t_static_cast
+ PARTITION (p = 2, dt = 20240101) VALUES (20, 'p2_new')"""
+ assertEquals([[10], [20]], sql("""SELECT id FROM t_static_cast ORDER
BY id"""))
+ assertTableEquals("t_static_cast", "ORDER BY id")
+
+ // An LTZ static partition value is local time in the session zone,
+ // while Paimon parses the overwrite value in the FE JVM zone; keep the
+ // session zone away from the JVM zone so a value left unconverted
+ // names another partition.
+ sql """set time_zone = 'Asia/Kolkata'"""
+ sql """INSERT INTO t_static_ltz VALUES
+ (1, '2024-01-15 08:30:45'), (2, '2024-01-15 16:30:45')"""
+ sql """INSERT OVERWRITE TABLE t_static_ltz
+ PARTITION (ts = '2024-01-15 08:30:45') VALUES (10)"""
+ def staticLtzIds = sql """SELECT id FROM t_static_ltz ORDER BY id"""
+ sql """set time_zone = default"""
+ assertEquals([[2], [10]], staticLtzIds)
+
// A partial static specification must use typed partition identity.
// NULL, blank, the literal "null", escaped path characters, and DATE
// subpartitions must remain distinct even when display paths overlap.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
