morningman commented on code in PR #68874:
URL: https://github.com/apache/doris/pull/68874#discussion_r4237342401
##########
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:
Update after rebasing onto #68786 (b9874e4334a): Paimon LTZ columns are now
always bound as TIMESTAMPTZ, so the static value and the row value both come
from the FE cast of the literal in the session zone, which puts a skipped time
at the instant of the DST change, as cctz does. The DATETIMEV2 path is gone,
and this check with it. Checked on a local catalog with the TIMESTAMPTZ binding
and a Europe/London session: the row written with `2024-03-31 01:30:45` lands
at 01:00:00Z, `INSERT OVERWRITE ... PARTITION (ts = '2024-03-31 01:30:45')
VALUES (10)` writes into that same partition, and an empty overwrite of that
value deletes it and leaves the 02:30:45Z partition. The DATETIMEV2 example in
`staticLtzOverwriteRejectsAnInstantTheSdkZoneRepeats` went too; the overlap
rejection (M3) is unchanged and covered for TIMESTAMPTZ values.
--
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]
