github-actions[bot] commented on code in PR #68874:
URL: https://github.com/apache/doris/pull/68874#discussion_r4237037980
##########
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:
[P1] Derive the static partition value from the bound row's actual string
coercion. Under default non-strict inserts, BindSink applies Substring to a
bounded CHAR/VARCHAR row, while this separate checkedCastTo does something
else: CHAR(2) with p='abcd' binds a row value of 'ab' but throws here even for
an append; VARCHAR(1) with p='😀x' writes the whole emoji, but
VarcharLiteral(value, 1) keeps one UTF-16 surrogate and Paimon encodes it as
'?'. An empty overwrite can then delete a preexisting '?' partition instead of
the emoji partition. FE constant folding and BE substring both count code
points, so they do not remove this mismatch.
##########
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:
[P2] Use generated query output for these deterministic results. The new
fixed-row assertEquals checks here and at lines 309 and 323 contradict root
AGENTS.md Testing Standards item 6, which requires qt_sql-style queries with
script-generated .out files for determined expected results. These queries
already ORDER BY id, so they can be converted directly to named ordered query
tests and their outputs generated by the regression runner.
##########
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:
[P1] Resolve a spring DST gap the same way as the BE row writer, or reject
the nonexistent local time. For session America/Los_Angeles, UTC FE JVM, and
static LTZ value 2024-03-10 02:30:45, Java atZone shifts the value to 03:30:45
PDT (10:30:45Z). The BE DateTimeV2 Arrow path uses CCTZ convert, which resolves
this skipped civil time to the transition instant (10:00:00Z). The overwrite
target and written row differ, so an empty overwrite can delete an unrelated
partition. This source-zone gap is separate from the destination-zone overlap
above. See the [CCTZ v2.5 conversion
rule](https://github.com/google/cctz/blob/v2.5/include/cctz/time_zone.h#L234-L244).
##########
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:
[P1] Bump the connector plugin API major for this new SPI method. The frozen
surface now includes getCastStaticPartitionSpec, but
connector.plugin.api.version and its pinned test remain 12.0. ApiVersionGate
checks the major only, and the FE parent-loads the SPI, so a new Paimon plugin
can be admitted by an older 12.0 FE whose interface lacks this method; the call
in PaimonWriteBinding then fails with NoSuchMethodError. Advance the manifest
version and test pin together with the surface change.
##########
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:
[P1] Preserve or validate the instant at a fall DST overlap. With an
America/Los_Angeles FE JVM, UTC session, and static LTZ value 2024-11-03
09:30:45, the written row is 09:30:45Z (the second 01:30 occurrence). This
conversion emits offset-free 01:30:45; Paimon 1.4.2 parses it via
LocalDateTime.atZone and chooses the first occurrence, 08:30:45Z. withOverwrite
can therefore target and, for an empty overwrite, delete the other partition.
Round-trip against Paimon's parser and reject unrepresentable instants, or pass
a typed/offset-preserving target. See [Paimon's LTZ
parser](https://github.com/apache/paimon/blob/release-1.4.2/paimon-common/src/main/java/org/apache/paimon/utils/DateTimeUtils.java#L504-L509).
--
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]
