morningman commented on code in PR #68874:
URL: https://github.com/apache/doris/pull/68874#discussion_r4237265974
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/handle/ConnectorWriteHandle.java:
##########
@@ -75,6 +76,20 @@ default Set<String> getStaticPartitionNullKeys() {
return Collections.emptySet();
}
+ /**
+ * The static partition values cast to the types of their columns, the
cast a written row gets, 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. 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() {
Review Comment:
Not bumping here, on purpose. The connector SPI has not shipped in a
release: the FE and every connector plugin are built from the same tree and
released together, so there is no Paimon plugin built against an older 12.0
surface for the gate to keep out. Until the SPI ships, methods are added
without moving `connector.plugin.api.version`. The frozen surface baseline
(`connector-plugin-surface.txt`) is refreshed in this PR, so the surface test
still pins the new method.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java:
##########
@@ -66,16 +69,51 @@ public Set<String> getStaticPartitionNullKeys() {
public void setStaticPartitionSpecFromExpressions(Map<String, Expression>
partitionValues) {
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) {
spec.put(entry.getKey(), ((Literal)
entry.getValue()).getStringValue());
+ literals.put(entry.getKey(), (Literal) entry.getValue());
if (entry.getValue() instanceof NullLiteral) {
nullKeys.add(entry.getKey());
}
}
}
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}. The cast is
the one constant folding applies to
+ * the value BindSink materializes into each row, 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()) {
+ Expression cast =
value.checkedCastTo(DataType.fromCatalogType(column.get().getType()));
Review Comment:
Agreed, fixed in 1d4999b3fbb. BindSink does not cast a string written into a
string column. With `enable_insert_value_auto_cast` on and `enable_strict_cast`
off (the defaults) it cuts the value to the CHAR / VARCHAR length with
`Substring`, which counts code points; otherwise it writes the value as is.
`castStaticPartitionSpec` now does the same for a string literal and a string
column: the session check is shared with BindSink
(`BindSink.truncatesStringOnInsert()`) and the cut is the constant-folding
`substring`. Other combinations keep the cast.
On a local file-system Paimon catalog:
| Statement | before | after |
|---|---|---|
| `INSERT INTO t PARTITION (p = 'abcd') VALUES (1)`, `p CHAR(2)` | `Cast
from abcd to CHAR(2) not supported` | row written with `p = 'ab'` |
| `INSERT OVERWRITE TABLE t PARTITION (p = 'abcd') VALUES (10)` | the same
error | partition `ab` replaced |
| empty overwrite `PARTITION (p = '😀x')`, `p VARCHAR(1)`, partitions `😀` and
`?` exist | partition `?` deleted | partition `😀` deleted |
`PluginDrivenTableSinkBindingTest.staticPartitionStringValueIsCutLikeTheWrittenRows`
covers CHAR, VARCHAR with a surrogate pair, STRING, and a session that does
not truncate.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -78,26 +84,75 @@ static FileStoreTable configureTableForWrite(FileStoreTable
table, boolean overw
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 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.
+ */
+ static Map<String, String> resolveStaticPartition(FileStoreTable table,
ConnectorWriteHandle writeHandle,
+ String sessionTimeZone) {
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> castValues =
writeHandle.getCastStaticPartitionSpec();
Map<String, String> result = new LinkedHashMap<>();
- for (Map.Entry<String, String> entry : requested.entrySet()) {
- String canonicalName = canonicalNames.get(entry.getKey());
+ 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);
+ 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(value, isTimestampTz(writeHandle,
canonicalName), sessionTimeZone);
+ }
+ if (writeHandle.isOverwrite() &&
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;
}
+ private static boolean isTimestampTz(ConnectorWriteHandle writeHandle,
String columnName) {
+ for (ConnectorColumn column : writeHandle.getBoundTargetColumns()) {
+ if (column.getName().equalsIgnoreCase(columnName)) {
+ return
"TIMESTAMPTZ".equalsIgnoreCase(column.getType().getTypeName());
+ }
+ }
+ throw new DorisConnectorException("Paimon partition column is missing
from the write schema: "
+ + columnName);
+ }
+
+ /**
+ * Paimon parses a static overwrite value of a TIMESTAMP WITH LOCAL TIME
ZONE column as local time in the FE
+ * JVM's default zone, so the value is moved there from the zone it was
written in: the session zone for a
+ * DATETIMEV2 value, UTC for a TIMESTAMPTZ value, which carries its
offset. Paimon's parser takes a space,
+ * not ISO's 'T', between the date and the time.
+ */
+ private static String toSdkLocalTime(String value, boolean timestampTz,
String sessionTimeZone) {
+ String isoValue = value.replace(' ', 'T');
+ ZonedDateTime instant = timestampTz
+ ? OffsetDateTime.parse(isoValue,
DateTimeFormatter.ISO_OFFSET_DATE_TIME).toZonedDateTime()
+ : LocalDateTime.parse(isoValue,
DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+ .atZone(ZoneId.of(sessionTimeZone,
PaimonConnectorMetadata.SESSION_TIME_ZONE_ALIASES));
+ return
instant.withZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime()
Review Comment:
Agreed, fixed in 1d4999b3fbb the way branch-4.x does it (#68532): an
overwrite rejects an LTZ value whose local time in the FE JVM zone falls in a
DST overlap, because both instants of the overlap format to the same
offset-free value. An INSERT keeps the value, since it overwrites nothing.
Reproduced on a local catalog with the FE JVM in Europe/London, a UTC
session, and partitions at 2024-10-27 00:30:45Z and 01:30:45Z (both 01:30:45 in
London). Before, an empty overwrite of `ts = '2024-10-27 01:30:45'` deleted the
00:30:45Z partition and kept the one it named. Now it fails with `Static LTZ
partition value for column 'ts' is ambiguous in FE JVM time zone Europe/London
and cannot be represented in a static overwrite`, and both partitions stay.
`PaimonWriteBindingTest.staticLtzOverwriteRejectsAnInstantTheSdkZoneRepeats`
covers both instants of an overlap for a TIMESTAMPTZ value and your DATETIMEV2
example; `staticLtzPartitionOutsideAnOverlapRoundTripsThroughPaimon` checks
with Paimon's parser that the same instants still pass where they are
unambiguous.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonWriteBinding.java:
##########
@@ -78,26 +84,75 @@ static FileStoreTable configureTableForWrite(FileStoreTable
table, boolean overw
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 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.
+ */
+ static Map<String, String> resolveStaticPartition(FileStoreTable table,
ConnectorWriteHandle writeHandle,
+ String sessionTimeZone) {
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> castValues =
writeHandle.getCastStaticPartitionSpec();
Map<String, String> result = new LinkedHashMap<>();
- for (Map.Entry<String, String> entry : requested.entrySet()) {
- String canonicalName = canonicalNames.get(entry.getKey());
+ 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);
+ 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(value, isTimestampTz(writeHandle,
canonicalName), sessionTimeZone);
+ }
+ if (writeHandle.isOverwrite() &&
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;
}
+ private static boolean isTimestampTz(ConnectorWriteHandle writeHandle,
String columnName) {
+ for (ConnectorColumn column : writeHandle.getBoundTargetColumns()) {
+ if (column.getName().equalsIgnoreCase(columnName)) {
+ return
"TIMESTAMPTZ".equalsIgnoreCase(column.getType().getTypeName());
+ }
+ }
+ throw new DorisConnectorException("Paimon partition column is missing
from the write schema: "
+ + columnName);
+ }
+
+ /**
+ * Paimon parses a static overwrite value of a TIMESTAMP WITH LOCAL TIME
ZONE column as local time in the FE
+ * JVM's default zone, so the value is moved there from the zone it was
written in: the session zone for a
+ * DATETIMEV2 value, UTC for a TIMESTAMPTZ value, which carries its
offset. Paimon's parser takes a space,
+ * not ISO's 'T', between the date and the time.
+ */
+ private static String toSdkLocalTime(String value, boolean timestampTz,
String sessionTimeZone) {
+ String isoValue = value.replace(' ', 'T');
+ ZonedDateTime instant = timestampTz
+ ? OffsetDateTime.parse(isoValue,
DateTimeFormatter.ISO_OFFSET_DATE_TIME).toZonedDateTime()
+ : LocalDateTime.parse(isoValue,
DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+ .atZone(ZoneId.of(sessionTimeZone,
PaimonConnectorMetadata.SESSION_TIME_ZONE_ALIASES));
Review Comment:
Agreed. `DateV2Value::unix_timestamp` goes through `cctz::convert`, which
returns the transition instant for a skipped civil time, while
`LocalDateTime.atZone` moves it forward by the length of the gap. Fixed in
1d4999b3fbb by rejecting the value instead of copying cctz's rule into the
connector: an overwrite whose DATETIMEV2 value does not exist in the session
zone fails with `... does not exist in session time zone ...`. A repeated
session-local time needs nothing, because both sides take the earlier offset
there (cctz returns `pre`, and Java takes the earlier offset of an overlap).
Reproduced with a Europe/London session: the row written with `2024-03-31
01:30:45` lands at 01:00:00Z, and before this change an empty overwrite of that
value deleted the 01:30:45Z partition (the row written with `02:30:45`)
instead. Now the overwrite is rejected and both partitions stay. Covered by
`PaimonWriteBindingTest.staticLtzOverwriteRejectsASessionTimeADstChangeSkips`.
##########
regression-test/suites/external_table_p0/paimon/write/test_paimon_write_transaction.groovy:
##########
@@ -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"""))
Review Comment:
The rule is right, but these cases cannot get a generated `.out` from this
branch. The suite creates its tables through Spark in the Paimon docker
environment, which only the External pipeline has, so a `qt_` block here would
have to be written by hand, which the same rule forbids. I kept explicit
assertions, as the suite already does with `assertTableEquals`: the expected
ids are the point of each case (which partition the overwrite replaced), and
the External run checks them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
