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 1386e1eac91 [fix](iceberg) Reject row-level DML on files whose
partition source column was dropped (#69229)
1386e1eac91 is described below
commit 1386e1eac91a2389a4a07c8b375450b0ea9092ae
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 19 08:11:10 2026 +0100
[fix](iceberg) Reject row-level DML on files whose partition source column
was dropped (#69229)
### What problem does this PR solve?
Issue Number: None
Related PR: #68481, #68124
Problem Summary:
**In short.** Ports the rest of branch-4.1 #68481 to master. When the
source column of a historical Iceberg partition field has been dropped,
DELETE and UPDATE on master commit a delete file under partition NULL,
after which Doris can no longer read the table. They now fail with a
clear error before anything is written, ordinary reads keep working, and
TIME partition values can be parsed back at commit.
**Background.** #68481 fixed Iceberg row-level DML after a table evolves
to an unpartitioned spec. Its main part, keeping each file's own spec ID
and partition tuple in the scan range once any spec is partitioned,
reached master with #68124; the regression added here confirms that
REPLACE to an unpartitioned table and dropping the last partition field
work on master. Row IDs carry each file's `partition_data_json`, and BE
writes the position-delete file of a DELETE, UPDATE or MERGE under that
partition.
**The problem, and what it cost.**
- Dropping the source column of a historical partition field leaves the
field with no type, so the file's partition tuple cannot be serialized.
`IcebergPartitionUtils.getPartitionValues` swallowed the
`UnsupportedOperationException` and wrote a null entry. A DELETE on such
a table then committed a delete file under partition NULL for a data
file in partition 7. Spark still reads the table, but every later Doris
query of it fails:
`Mismatched partition tuples (PartitionData{partition_key=null},
PartitionData{partition_key=7}) for delete file
s3://.../delete_pos_....parquet and data file
s3a://.../partition_key=7/00000-....parquet`
- `parsePartitionValueFromString` had no TIME branch, so the value
`serializePartitionValue` renders for a TIME partition could not be
parsed back when the delete file is committed
(`convertToPartitionData`).
**How this PR fixes it.**
- `getPartitionValues` throws for a non-null value it cannot serialize
instead of returning a null entry. A null needs no type and stays null,
so a historical NULL partition of the dropped column, and the `void(p)`
field a v1 table keeps for a removed partition field, still serialize.
- When a data file's partition tuple cannot be serialized, its range
carries no `partition_data_json`. It keeps the raw values instead
(binary as hex), which still identify the file's partition for EXPLAIN
`partition=N/M` and `sql_block_rule` `partition_num`, as legacy
`IcebergScanNode` counted distinct `PartitionData`. Ordinary reads do
not use the payload, so they are unaffected.
- A scan that produces row IDs over such a range is rejected: `Cannot
produce Iceberg row IDs with unsupported partition types in spec N.
Rewrite historical data files before DELETE or UPDATE.` The check runs
on every `planScan` call, including one served by the statement scan
reuse (its key leaves out the projection, as for the metadata-column
reader check), and on each range of a streaming split source.
- `getColumnHandles` returns a handle for the hidden row-id column, as
it does for `_file` and `_pos`, so a DML scan request carries it.
Before, the column had no handle and never reached the connector. The
field-id schema dictionary leaves it out, since it is not a table
column.
- `parsePartitionValueFromString` parses TIME (microseconds since
midnight).
Branch-4.x renders `partition_data_json` only for scans that produce row
IDs. Master keeps rendering it for every scan, because
`specId|partitionDataJson` is the partition identity behind EXPLAIN
`partition=N/M` and `sql_block_rule` `partition_num`.
**Results.** Local cluster with the Iceberg docker environment (tables
written by Spark):
| Case | Before | After |
|---|---|---|
| Spark REPLACE a partitioned table with an unpartitioned one, then two
Doris DELETEs | Works (#68124) | Same |
| Drop the last partition field, then DELETE and UPDATE across both
specs | Works | Same |
| Drop the partition field, then its source column; Doris DELETE
(Parquet and ORC) | Commits a delete file under partition NULL; Doris
reads of the table then fail with `Mismatched partition tuples` |
Rejected with the error above; no delete file, rows unchanged in Doris
and Spark |
| Same table, Doris UPDATE | Same as DELETE | Rejected |
| Same table, Doris DELETE of the row in the `partition_key = NULL` file
| Works | Same |
| DELETE on historical binary partitions; DELETE and UPDATE on pre-1970
timestamp partitions | Works | Same |
Co-authored-by: Gabriel <[email protected]>
---
.../iceberg/IcebergConnectorMetadata.java | 8 +-
.../connector/iceberg/IcebergPartitionUtils.java | 60 ++++--
.../connector/iceberg/IcebergScanPlanProvider.java | 67 ++++++-
.../doris/connector/iceberg/IcebergScanRange.java | 31 +++-
.../iceberg/IcebergConnectorMetadataTest.java | 4 +-
.../iceberg/IcebergPartitionUtilsTest.java | 73 ++++++++
.../iceberg/IcebergScanPlanProviderTest.java | 181 ++++++++++++++++++
...t_iceberg_delete_unpartitioned_evolution.groovy | 204 +++++++++++++++++++++
8 files changed, 596 insertions(+), 32 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index a1259061585..a2f63a69239 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -125,6 +125,9 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
private static final String ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL =
"_last_updated_sequence_number";
private static final String ICEBERG_FILE_PATH_COL = "_file";
private static final String ICEBERG_ROW_POSITION_COL = "_pos";
+ // The hidden DML row-id column, fe-core Column.ICEBERG_ROWID_COL (pinned
by
+ // IcebergScanPlanProviderClassifyColumnTest through the scan provider's
copy).
+ private static final String DORIS_ICEBERG_ROWID_COL =
"__DORIS_ICEBERG_ROWID_COL__";
private static final int ICEBERG_ROW_ID_FIELD_ID = 2147483540;
private static final int ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_FIELD_ID =
2147483539;
private static final int ICEBERG_ROW_LINEAGE_MIN_VERSION = 3;
@@ -783,7 +786,7 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
rejectReservedMetadataColumns(schema);
}
Map<String, ConnectorColumnHandle> handles = new LinkedHashMap<>(
- fields.size() + (appendDataFileMetadataColumns ? 2 : 0));
+ fields.size() + (appendDataFileMetadataColumns ? 3 : 0));
for (Types.NestedField field : fields) {
String name = field.name();
handles.put(name, new IcebergColumnHandle(name, field.fieldId()));
@@ -791,6 +794,9 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
if (appendDataFileMetadataColumns) {
handles.put(ICEBERG_FILE_PATH_COL, new
IcebergColumnHandle(ICEBERG_FILE_PATH_COL, -1));
handles.put(ICEBERG_ROW_POSITION_COL, new
IcebergColumnHandle(ICEBERG_ROW_POSITION_COL, -1));
+ // Synthesized by the reader for DML and never in the schema. Its
handle tells the scan provider that a
+ // scan produces row IDs
(IcebergScanPlanProvider#rejectRowIdScanWithoutPartitionData).
+ handles.put(DORIS_ICEBERG_ROWID_COL, new
IcebergColumnHandle(DORIS_ICEBERG_ROWID_COL, -1));
}
return handles;
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
index 46f7659f698..24094ea76da 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java
@@ -195,9 +195,12 @@ final class IcebergPartitionUtils {
/**
* The serialized value for <b>every</b> partition field (identity +
transform), in spec order, used for
- * {@code partition_data_json}. Mirrors legacy {@code
IcebergUtils.getPartitionValues}. A field whose
- * value cannot be serialized (BINARY/FIXED) yields a {@code null} entry
(not dropped) to keep positional
- * alignment with the spec.
+ * {@code partition_data_json}. Mirrors legacy {@code
IcebergUtils.getPartitionValues}. A non-null value whose
+ * type cannot be serialized (a historical field whose source column was
dropped) throws
+ * {@link UnsupportedOperationException}: these values also identify the
partition of the delete files
+ * BE writes for DML, so an unsupported value must never become a {@code
null} entry. A null needs no type,
+ * so the void field a v1 table keeps for a removed partition field stays
{@code null} after its source
+ * column is dropped.
*/
static List<String> getPartitionValues(PartitionData partitionData,
PartitionSpec partitionSpec, ZoneId zone) {
List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
@@ -206,15 +209,8 @@ final class IcebergPartitionUtils {
List<String> partitionValues = new ArrayList<>(fields.size());
for (int i = 0; i < fields.size(); i++) {
- NestedField field = fields.get(i);
Object value = partitionData.get(i);
- try {
- partitionValues.add(serializePartitionValue(field.type(),
value, zone));
- } catch (UnsupportedOperationException e) {
- LOG.warn("Failed to serialize Iceberg partition value for
field {}: {}", field.name(),
- e.getMessage());
- partitionValues.add(null);
- }
+ partitionValues.add(value == null ? null :
serializePartitionValue(fields.get(i).type(), value, zone));
}
return partitionValues;
}
@@ -225,9 +221,32 @@ final class IcebergPartitionUtils {
* content is identical.
*/
static String getPartitionDataJson(PartitionData partitionData,
PartitionSpec partitionSpec, ZoneId zone) {
- List<String> partitionValues = getPartitionValues(partitionData,
partitionSpec, zone);
+ return toJsonArray(getPartitionValues(partitionData, partitionSpec,
zone));
+ }
+
+ /**
+ * Identifies a partition tuple {@link #getPartitionValues} cannot
serialize, for the scanned-partition count
+ * only: a JSON array of the raw values in spec order (binary as hex), so
each distinct tuple counts once, as
+ * legacy {@code IcebergScanNode} keyed the count by the {@link
PartitionData} itself.
+ */
+ static String getRawPartitionKey(PartitionData partitionData) {
+ List<String> values = new ArrayList<>(partitionData.size());
+ for (int i = 0; i < partitionData.size(); i++) {
+ Object value = partitionData.get(i);
+ if (value == null) {
+ values.add(null);
+ } else if (value instanceof java.nio.ByteBuffer) {
+ values.add(binaryHex((java.nio.ByteBuffer) value));
+ } else {
+ values.add(value.toString());
+ }
+ }
+ return toJsonArray(values);
+ }
+
+ private static String toJsonArray(List<String> values) {
try {
- return JsonUtil.mapper().writeValueAsString(partitionValues);
+ return JsonUtil.mapper().writeValueAsString(values);
} catch (com.fasterxml.jackson.core.JsonProcessingException e) {
throw new RuntimeException("Failed to serialize iceberg partition
data to JSON, error message is:"
+ e.getMessage(), e);
@@ -382,10 +401,7 @@ final class IcebergPartitionUtils {
if (value == null) {
return null;
}
- java.nio.ByteBuffer bytes = ((java.nio.ByteBuffer)
value).duplicate();
- byte[] copy = new byte[bytes.remaining()];
- bytes.get(copy);
- return "0x" + java.util.HexFormat.of().formatHex(copy);
+ return binaryHex((java.nio.ByteBuffer) value);
case DATE:
if (value == null) {
return null;
@@ -430,6 +446,13 @@ final class IcebergPartitionUtils {
}
}
+ private static String binaryHex(java.nio.ByteBuffer value) {
+ java.nio.ByteBuffer bytes = value.duplicate();
+ byte[] copy = new byte[bytes.remaining()];
+ bytes.get(copy);
+ return "0x" + java.util.HexFormat.of().formatHex(copy);
+ }
+
// Canonical partition-timestamp format ("yyyy-MM-dd HH:mm:ss" with an
optional micro/nano fraction),
// the form BE renders human-readable partition values in. Legacy parsed
via the nereids multi-format
// DateLiteral.parseDateTime (connector-forbidden); the canonical form is
the only one BE emits, so this
@@ -491,6 +514,9 @@ final class IcebergPartitionUtils {
case DATE:
// Iceberg date is days since epoch (1970-01-01).
return (int) LocalDate.parse(valueStr,
DateTimeFormatter.ISO_LOCAL_DATE).toEpochDay();
+ case TIME:
+ // Iceberg time is microseconds since midnight, the
inverse of serializePartitionValue.
+ return LocalTime.parse(valueStr,
DateTimeFormatter.ISO_LOCAL_TIME).toNanoOfDay() / 1000;
case TIMESTAMP_NANO:
// Match the sink cast: a date-only static partition
denotes midnight.
if (valueStr.length() == 10) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index 515d67b5f56..2f07b5ac0a7 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -441,8 +441,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// validate every call before a cached range list can be returned.
validateMetadataColumnReader(session, request.getColumns());
if (session == null || !session.isExternalScanTaskReuseEnabled()) {
- return planScanInternal(session, icebergHandle,
request.getColumns(),
- request.getFilter(), request.isCountPushdown());
+ return
rejectRowIdScanWithoutPartitionData(request.getColumns(),
planScanInternal(session,
+ icebergHandle, request.getColumns(),
request.getFilter(), request.isCountPushdown()));
}
if (icebergHandle.isSystemTable()) {
// System tables read connector metadata through their own
readers; never reuse them.
@@ -469,10 +469,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
Map<IcebergScanReuseKey, List<ConnectorScanRange>> scanReuse =
session.getStatementScope()
.computeIfAbsentForPlanning(memoKey, () -> new
ConcurrentHashMap<>());
IcebergScanReuseKey reuseKey = new
IcebergScanReuseKey(icebergHandle, request);
- return scanReuse.computeIfAbsent(reuseKey,
+ return rejectRowIdScanWithoutPartitionData(request.getColumns(),
scanReuse.computeIfAbsent(reuseKey,
k -> Collections.unmodifiableList(planScanInternal(session,
icebergHandle, request.getColumns(),
request.getFilter(),
- request.isCountPushdown())));
+ request.isCountPushdown()))));
} catch (RuntimeException e) {
// Normal data scans and native position_deletes run on File
Scanner V2. Keep the serialized JNI
// system-table route untouched because its deferred reads belong
to the V1 scanner contract.
@@ -599,7 +599,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
scan, session, table, filter, sliceSize);
return new IcebergStreamingSplitSource(tasks, table,
formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer, sliceSize,
- iceHandle.getRewriteFileScope(), iceHandle);
+ iceHandle.getRewriteFileScope(), iceHandle,
requiresRowId(columns));
} catch (RuntimeException e) {
throw IcebergExceptionUtils.wrapMetadataReadFailure(iceHandle, e);
}
@@ -664,6 +664,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private final long sliceSize;
private final Set<String> rewriteScope;
private final IcebergTableHandle handle;
+ private final boolean requiresRowId;
// Lazily opened on first hasNext() so the ctor never throws —
iceberg's ParallelIterable submits
// manifest readers in tasks.iterator(), which can fail; opening it
eagerly here would throw out of
// streamSplits() BEFORE the source is returned, leaking the
planFiles() iterable (the engine pump's
@@ -679,7 +680,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
IcebergStreamingSplitSource(CloseableIterable<FileScanTask> tasks,
Table table, int formatVersion,
boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
UnaryOperator<String> uriNormalizer, long sliceSize,
Set<String> rewriteScope,
- IcebergTableHandle handle) {
+ IcebergTableHandle handle, boolean requiresRowId) {
this.tasks = tasks;
this.table = table;
this.formatVersion = formatVersion;
@@ -690,6 +691,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
this.sliceSize = sliceSize;
this.rewriteScope = rewriteScope;
this.handle = handle;
+ this.requiresRowId = requiresRowId;
}
@Override
@@ -705,6 +707,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
IcebergScanRange range =
buildRangeForTask(iterator.next(), table, formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer,
sliceSize, rewriteScope, null, scratch);
if (range != null) {
+ if (requiresRowId) {
+ rejectRowIdRangeWithoutPartitionData(range);
+ }
buffered = range;
return true;
}
@@ -1622,6 +1627,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private DataFile file;
private Integer partitionSpecId;
private String partitionDataJson;
+ private String rawPartitionKey;
private Map<String, String> partitionValues = Collections.emptyMap();
// The table's CURRENT spec isPartitioned() — scan-invariant, memoized
with the rest so the display
// gate (see IcebergScanRange#getScannedPartitionKey) costs one
table.spec() per file, not per slice.
@@ -1676,6 +1682,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
.formatVersion(formatVersion)
.partitionSpecId(file.partitionSpecId)
.partitionDataJson(file.partitionDataJson)
+ .rawPartitionKey(file.rawPartitionKey)
.firstRowId(file.firstRowId)
.lastUpdatedSequenceNumber(file.lastUpdatedSequenceNumber)
.partitionValues(file.partitionValues)
@@ -1701,13 +1708,21 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
perFileInvariantComputeCount++;
Integer partitionSpecId = null;
String partitionDataJson = null;
+ String rawPartitionKey = null;
Map<String, String> partitionValues = Collections.emptyMap();
if (partitioned && dataFile.partition() instanceof PartitionData) {
PartitionData partitionData = (PartitionData) dataFile.partition();
int specId = dataFile.specId();
PartitionSpec spec = table.specs().get(specId);
partitionSpecId = specId;
- partitionDataJson =
IcebergPartitionUtils.getPartitionDataJson(partitionData, spec, zone);
+ try {
+ partitionDataJson =
IcebergPartitionUtils.getPartitionDataJson(partitionData, spec, zone);
+ } catch (UnsupportedOperationException e) {
+ // A historical partition field whose source column was
dropped has no serializable type. Only
+ // row IDs need the tuple, so the file stays readable and a
row-ID scan is rejected per call; the
+ // raw values still count the file's partition in EXPLAIN
partition=N/M and sql_block_rule.
+ rawPartitionKey =
IcebergPartitionUtils.getRawPartitionKey(partitionData);
+ }
// Order the identity values as the path_partition_keys list
(legacy getOrderedPathPartitionKeys),
// filtered to keys this file carries — so columns-from-path
matches legacy ordering exactly.
Map<String, String> identityMap =
@@ -1747,6 +1762,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
file.file = dataFile;
file.partitionSpecId = partitionSpecId;
file.partitionDataJson = partitionDataJson;
+ file.rawPartitionKey = rawPartitionKey;
file.partitionValues = partitionValues;
// Display-only gate (see IcebergScanRange#getScannedPartitionKey): a
table whose CURRENT spec is
// unpartitioned reports no partitions, so the old-spec partitions its
files still carry must not
@@ -2160,8 +2176,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* The lowercased names of the requested (pruned) columns — the
authoritative Doris scan slots the field-id
* dictionary keys its {@code -1} entry off (so its top-level names == the
BE scan-slot names BY
* CONSTRUCTION; CI #969249). The names come straight from the {@link
IcebergColumnHandle}s
- * {@code IcebergConnectorMetadata.getColumnHandles} produced (already
lowercased). An empty list (count-only
- * scan / no column handles) makes the dictionary fall back to all
top-level columns.
+ * {@code IcebergConnectorMetadata.getColumnHandles} produced (already
lowercased), minus the synthesized
+ * {@code _file} / {@code _pos} / hidden row-id columns, which are not in
the table schema. An empty list
+ * (count-only scan / no column handles) makes the dictionary fall back to
all top-level columns.
*/
private static List<String>
requestedLowerNames(List<ConnectorColumnHandle> columns) {
if (columns == null || columns.isEmpty()) {
@@ -2171,7 +2188,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
for (ConnectorColumnHandle column : columns) {
String name = ((IcebergColumnHandle) column).getName();
if (!ICEBERG_FILE_PATH_COL.equalsIgnoreCase(name)
- && !ICEBERG_ROW_POSITION_COL.equalsIgnoreCase(name)) {
+ && !ICEBERG_ROW_POSITION_COL.equalsIgnoreCase(name)
+ && !DORIS_ICEBERG_ROWID_COL.equalsIgnoreCase(name)) {
names.add(name);
}
}
@@ -3093,6 +3111,35 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return Boolean.parseBoolean(raw.trim());
}
+ private static boolean requiresRowId(List<ConnectorColumnHandle> columns) {
+ return columns.stream()
+ .filter(column -> column instanceof IcebergColumnHandle)
+ .anyMatch(column ->
DORIS_ICEBERG_ROWID_COL.equalsIgnoreCase(((IcebergColumnHandle)
column).getName()));
+ }
+
+ /**
+ * Rejects a scan that produces Iceberg row IDs (DELETE, UPDATE, MERGE)
when a file's partition tuple could not
+ * be serialized. Row IDs carry the partition of the delete files BE
writes, and a missing value must not become
+ * a NULL partition. Checked on every call, like {@link
#validateMetadataColumnReader}, because the scan reuse
+ * key leaves the projection out.
+ */
+ private static List<ConnectorScanRange>
rejectRowIdScanWithoutPartitionData(
+ List<ConnectorColumnHandle> columns, List<ConnectorScanRange>
ranges) {
+ if (requiresRowId(columns)) {
+ for (ConnectorScanRange range : ranges) {
+ rejectRowIdRangeWithoutPartitionData((IcebergScanRange) range);
+ }
+ }
+ return ranges;
+ }
+
+ private static void rejectRowIdRangeWithoutPartitionData(IcebergScanRange
range) {
+ if (range.isPartitionDataUnavailable()) {
+ throw new DorisConnectorException("Cannot produce Iceberg row IDs
with unsupported partition types in spec "
+ + range.getPartitionSpecId() + ". Rewrite historical data
files before DELETE or UPDATE.");
+ }
+ }
+
private static boolean requiresMetadataColumns(List<ConnectorColumnHandle>
columns) {
return columns.stream()
.filter(column -> column instanceof IcebergColumnHandle)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
index a22ac634122..ef48c73ed2d 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanRange.java
@@ -62,6 +62,11 @@ public class IcebergScanRange implements ConnectorScanRange {
private final int formatVersion;
private final Integer partitionSpecId;
private final String partitionDataJson;
+ // FE-only, set when the file's partition tuple has a value that cannot be
serialized (a historical partition
+ // field whose source column was dropped), so partitionDataJson is absent:
the raw values identify the
+ // partition for getScannedPartitionKey, and a scan that produces row IDs
is rejected
+ // (IcebergScanPlanProvider#rejectRowIdScanWithoutPartitionData). Ordinary
reads need neither.
+ private final String rawPartitionKey;
private final Long firstRowId;
private final Long lastUpdatedSequenceNumber;
// Identity partition column (lowercased) -> serialized value, already
ordered as the path_partition_keys
@@ -134,6 +139,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
this.formatVersion = builder.formatVersion;
this.partitionSpecId = builder.partitionSpecId;
this.partitionDataJson = builder.partitionDataJson;
+ this.rawPartitionKey = builder.rawPartitionKey;
this.firstRowId = builder.firstRowId;
this.lastUpdatedSequenceNumber = builder.lastUpdatedSequenceNumber;
this.partitionValues = builder.partitionValues != null
@@ -233,17 +239,30 @@ public class IcebergScanRange implements
ConnectorScanRange {
* A distinct-faithful key for the native iceberg partition this file
belongs to (FIX-L12), or
* {@code null} when the file carries no {@code PartitionData}
(unpartitioned / current spec
* unpartitioned). Combines the partition-spec id with the serialized
{@code PartitionData}
- * ({@code partitionDataJson}) so that two files are keyed equal iff they
share the same spec and
+ * ({@code partitionDataJson}), or with its raw values when they cannot be
serialized
+ * ({@code rawPartitionKey}), so that two files are keyed equal iff they
share the same spec and
* partition tuple — mirroring legacy {@code IcebergScanNode}'s de-dup by
* {@code (PartitionData) file().partition()}, and disambiguating
cross-spec value collisions
* (e.g. {@code identity(id)=2} vs {@code bucket(id)=2}) that the
value-only json would merge.
* {@code IcebergScanPlanProvider} counts distinct non-null keys for
{@code selectedPartitionNum}.
*/
String getScannedPartitionKey() {
- if (partitionDataJson == null || !countsAsScannedPartition) {
+ if (!countsAsScannedPartition) {
return null;
}
- return partitionSpecId + "|" + partitionDataJson;
+ if (partitionDataJson != null) {
+ return partitionSpecId + "|" + partitionDataJson;
+ }
+ // The marker keeps a raw tuple apart from a serialized one of the
same spec.
+ return rawPartitionKey == null ? null : partitionSpecId + "|raw" +
rawPartitionKey;
+ }
+
+ Integer getPartitionSpecId() {
+ return partitionSpecId;
+ }
+
+ boolean isPartitionDataUnavailable() {
+ return rawPartitionKey != null;
}
/**
@@ -478,6 +497,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
private int formatVersion = 2;
private Integer partitionSpecId;
private String partitionDataJson;
+ private String rawPartitionKey;
private Long firstRowId;
private Long lastUpdatedSequenceNumber;
private Map<String, String> partitionValues;
@@ -569,6 +589,11 @@ public class IcebergScanRange implements
ConnectorScanRange {
return this;
}
+ public Builder rawPartitionKey(String rawPartitionKey) {
+ this.rawPartitionKey = rawPartitionKey;
+ return this;
+ }
+
/**
* Whether this range counts toward the scanned-partition display
(default {@code true}); pass the
* table's CURRENT spec {@code isPartitioned()} — see {@link
IcebergScanRange#getScannedPartitionKey()}.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
index 2e174c1ff99..6d3efdbfe42 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
@@ -1395,11 +1395,13 @@ public class IcebergConnectorMetadataTest {
Map<String, ConnectorColumnHandle> handles =
metadataWith(ops).getColumnHandles(null, new
IcebergTableHandle("db1", "t1"));
- Assertions.assertEquals(4, handles.size());
+ Assertions.assertEquals(5, handles.size());
Assertions.assertTrue(handles.containsKey("ID"));
Assertions.assertTrue(handles.containsKey("Name"));
Assertions.assertTrue(handles.containsKey("_file"));
Assertions.assertTrue(handles.containsKey("_pos"));
+ // The hidden DML row-id column, so the scan provider can tell a scan
that produces row IDs.
+ Assertions.assertEquals(-1, ((IcebergColumnHandle)
handles.get("__DORIS_ICEBERG_ROWID_COL__")).getFieldId());
Assertions.assertFalse(handles.containsKey("id"), "post-#65094 the
handle key keeps the iceberg case");
Assertions.assertEquals(7, ((IcebergColumnHandle)
handles.get("ID")).getFieldId());
Assertions.assertEquals(9, ((IcebergColumnHandle)
handles.get("Name")).getFieldId());
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
index 323d28a73d8..cb0bd8d2aee 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsTest.java
@@ -506,6 +506,68 @@ public class IcebergPartitionUtilsTest {
IcebergPartitionUtils.getPartitionDataJson(pd, spec,
ZoneOffset.UTC));
}
+ @Test
+ public void partitionDataJsonRejectsFieldWhoseSourceColumnWasDropped() {
+ // From branch-4.x #68481: once the source column of a historical
identity field is dropped, the field has no
+ // type and its value cannot be serialized. The JSON is the partition
of the delete files BE writes for DML,
+ // so it must fail rather than carry a null entry. MUTATION: restoring
the null fallback -> "[null]" -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "p", Types.IntegerType.get()));
+ Table table = tableWith(schema,
PartitionSpec.builderFor(schema).identity("p").build());
+ table.updateSpec().removeField("p").commit();
+ table.updateSchema().deleteColumn("p").commit();
+ PartitionSpec historical = table.specs().get(0);
+ PartitionData pd = new PartitionData(historical.partitionType());
+ pd.set(0, 7);
+
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> IcebergPartitionUtils.getPartitionDataJson(pd,
historical, ZoneOffset.UTC));
+ }
+
+ @Test
+ public void partitionDataJsonKeepsNullOfFieldWhoseSourceColumnWasDropped()
{
+ // A null needs no type: the void field a v1 table keeps for a removed
partition field is always null, and a
+ // historical NULL partition is one, so both still serialize after the
source column is dropped.
+ // MUTATION: serializing a null by its type ->
UnsupportedOperationException -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "p", Types.IntegerType.get()));
+ Table table = tableWith(schema,
PartitionSpec.builderFor(schema).identity("p").build());
+ table.updateSpec().removeField("p").commit();
+ table.updateSchema().deleteColumn("p").commit();
+ PartitionSpec historical = table.specs().get(0);
+ PartitionData pd = new PartitionData(historical.partitionType());
+ pd.set(0, null);
+
+ Assertions.assertEquals("[null]",
IcebergPartitionUtils.getPartitionDataJson(pd, historical, ZoneOffset.UTC));
+ }
+
+ @Test
+ public void rawPartitionKeyIdentifiesTupleThatCannotBeSerialized() {
+ // The scanned-partition count keys such a tuple by its raw values:
binary as hex (a ByteBuffer has no
+ // content-based toString), and a null apart from the string "null".
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "p", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "b", Types.BinaryType.get()),
+ Types.NestedField.optional(3, "s", Types.StringType.get()));
+ Table table = tableWith(schema,
PartitionSpec.builderFor(schema).identity("p").identity("b").identity("s")
+ .build());
+ table.updateSpec().removeField("p").removeField("b").commit();
+ table.updateSchema().deleteColumn("p").deleteColumn("b").commit();
+ PartitionSpec historical = table.specs().get(0);
+ PartitionData pd = new PartitionData(historical.partitionType());
+ pd.set(0, 7);
+ pd.set(1, ByteBuffer.wrap(new byte[] {0x61, 0x62}));
+ pd.set(2, null);
+
+ Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> IcebergPartitionUtils.getPartitionDataJson(pd,
historical, ZoneOffset.UTC));
+ Assertions.assertEquals("[\"7\",\"0x6162\",null]",
IcebergPartitionUtils.getRawPartitionKey(pd));
+ pd.set(2, "null");
+ Assertions.assertEquals("[\"7\",\"0x6162\",\"null\"]",
IcebergPartitionUtils.getRawPartitionKey(pd));
+ }
+
// ---- parsePartitionValueFromString: legacy
IcebergUtils.parsePartitionValueFromString matrix (T04) ----
// The write-direction inverse of serializePartitionValue: BE sends
human-readable partition strings, the
// connector converts them to iceberg internal partition objects for
PartitionData. Mirrors legacy cell-by-cell.
@@ -596,6 +658,17 @@ public class IcebergPartitionUtilsTest {
"2021-01-01 00:00:00.123456",
Types.TimestampType.withoutZone(), ZoneOffset.UTC));
}
+ @Test
+ public void parseTimeIsInverseOfSerialize() {
+ // From branch-4.x #68481: a DELETE on a file of a historical TIME
partition commits its delete file under
+ // the partition parsed back from partition_data_json, so the parse
must accept what serialize renders.
+ long micros = 3661_123_456L;
+ String serialized =
IcebergPartitionUtils.serializePartitionValue(Types.TimeType.get(), micros,
ZoneOffset.UTC);
+ Assertions.assertEquals("01:01:01.123456", serialized);
+ Assertions.assertEquals(Long.valueOf(micros),
IcebergPartitionUtils.parsePartitionValueFromString(
+ serialized, Types.TimeType.get(), ZoneOffset.UTC));
+ }
+
@Test
public void parseUnsupportedTypeThrows() {
Assertions.assertThrows(IllegalArgumentException.class, () ->
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 37135377acc..633595a680a 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -1139,6 +1139,167 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals("[\"7\"]",
desc.getTableFormatParams().getIcebergParams().getPartitionDataJson());
}
+ @Test
+ public void rowIdScanRejectsFilesWhosePartitionSourceColumnWasDropped() {
+ // From branch-4.x #68481: dropping the source column of a historical
partition field leaves the field with
+ // no type, so the file's partition tuple cannot be serialized. An
ordinary read does not need it and keeps
+ // working. A scan producing row IDs (DELETE / UPDATE / MERGE) is
rejected: the row IDs carry the partition
+ // of the delete files BE writes, and a NULL partition there leaves a
delete file Doris cannot read back.
+ IcebergScanPlanProvider provider =
providerOver(droppedPartitionSourceTable());
+ ConnectorSession session = new FakeScanSession("UTC",
Collections.emptyMap());
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "pt");
+
+ List<ConnectorScanRange> ranges = provider.planScan(session,
scanOf(handle).build());
+
+ Assertions.assertEquals(1, ranges.size());
+ TFileRangeDesc desc = populate(ranges.get(0));
+ Assertions.assertEquals(0,
desc.getTableFormatParams().getIcebergParams().getPartitionSpecId());
+
Assertions.assertFalse(desc.getTableFormatParams().getIcebergParams().isSetPartitionDataJson());
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.planScan(session, rowIdScanOf(handle).build()));
+ Assertions.assertEquals("Cannot produce Iceberg row IDs with
unsupported partition types in spec 0. "
+ + "Rewrite historical data files before DELETE or UPDATE.",
ex.getMessage());
+ }
+
+ @Test
+ public void statementReuseStillRejectsRowIdScanWithoutPartitionData() {
+ // The reuse key leaves the projection out, so a row-ID scan served
from an ordinary scan's memoized ranges
+ // must still be checked. MUTATION: checking only freshly planned
ranges -> the second call returns -> red.
+ IcebergScanPlanProvider provider =
providerOver(droppedPartitionSourceTable());
+ ConnectorSession session = reuseSession(new TestStatementScope(), 0L);
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "pt");
+
+ provider.planScan(session, scanOf(handle).build());
+
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.planScan(session, rowIdScanOf(handle).build()));
+ }
+
+ @Test
+ public void streamSplitsRejectsRowIdScanWithoutPartitionData() throws
IOException {
+ // Batch mode builds its ranges lazily on the split-pump thread, so it
checks each range as it is built.
+ IcebergScanPlanProvider provider =
providerOver(droppedPartitionSourceTable());
+ try (ConnectorSplitSource source = provider.streamSplits(new
FakeScanSession("UTC", Collections.emptyMap()),
+ rowIdScanOf(new IcebergTableHandle("db1",
"pt")).filter(Optional.empty()).limit(-1L).build())) {
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class, source::hasNext);
+ Assertions.assertTrue(ex.getMessage().contains("unsupported
partition types in spec 0"), ex.getMessage());
+ }
+ }
+
+ /** A file written under identity(p), after p is removed from the spec and
then dropped from the schema. */
+ private static Table droppedPartitionSourceTable() {
+ PartitionSpec spec =
PartitionSpec.builderFor(PART_SCHEMA).identity("p").build();
+ Table table = createTable("pt", PART_SCHEMA, spec);
+ table.newAppend()
+ .appendFile(dataFile(spec, "s3://b/db/pt/p=7/old.parquet",
1024, null, "p=7"))
+ .commit();
+ table.updateSpec().removeField("p").commit();
+ table.updateSchema().deleteColumn("p").commit();
+ return table;
+ }
+
+ private static ConnectorScanRequest.Builder rowIdScanOf(IcebergTableHandle
handle) {
+ return ConnectorScanRequest.builder(handle, Arrays.asList(new
IcebergColumnHandle("id", 1),
+ new IcebergColumnHandle("__DORIS_ICEBERG_ROWID_COL__", -1)));
+ }
+
+ @Test
+ public void
rowIdScanKeepsNullOfRetainedVoidFieldWhoseSourceColumnWasDropped() {
+ // A v1 table keeps a removed partition field as void(p), whose value
is always null. Dropping p leaves the
+ // field without a type, but a null needs none: DML must keep working
on files written under the void spec,
+ // which rewriting historical files cannot get rid of. MUTATION:
serializing a null by its type -> red.
+ Table table = retainedVoidFieldTable("pv", true);
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ rowIdScanOf(new IcebergTableHandle("db1", "pv")).build());
+
+ Assertions.assertEquals(1, ranges.size());
+ TIcebergFileDesc desc =
populate(ranges.get(0)).getTableFormatParams().getIcebergParams();
+ Assertions.assertEquals(1, desc.getPartitionSpecId());
+ Assertions.assertEquals("[null,\"5\"]", desc.getPartitionDataJson());
+ }
+
+ @Test
+ public void rowIdScanAcceptsAllVoidSpecWhoseSourceColumnWasDropped() {
+ // The same evolution when void(p) is the only field left: the spec is
no longer partitioned.
+ Table table = retainedVoidFieldTable("pav", false);
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ rowIdScanOf(new IcebergTableHandle("db1", "pav")).build());
+
+ Assertions.assertEquals(1, ranges.size());
+ TIcebergFileDesc desc =
populate(ranges.get(0)).getTableFormatParams().getIcebergParams();
+ Assertions.assertEquals(1, desc.getPartitionSpecId());
+ }
+
+ @Test
+ public void
scannedPartitionCountKeepsPartitionsWhoseSourceColumnWasDropped() {
+ // Files of a historical identity(p) whose source column was dropped
have no partition_data_json, but they
+ // are still scanned partitions: EXPLAIN partition=N/M and
sql_block_rule partition_num must count them,
+ // each distinct tuple once, as legacy IcebergScanNode did by keying
on PartitionData.
+ // MUTATION: counting only ranges with partition_data_json -> 1 -> red.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "p", Types.IntegerType.get()),
+ Types.NestedField.optional(3, "q", Types.IntegerType.get()));
+ PartitionSpec byP =
PartitionSpec.builderFor(schema).identity("p").build();
+ Table table = createTable("pc", schema, byP);
+ table.newAppend()
+ .appendFile(dataFile(byP, "s3://b/db/pc/p=7/a.parquet", 1024,
null, "p=7"))
+ .appendFile(dataFile(byP, "s3://b/db/pc/p=7/b.parquet", 1024,
null, "p=7"))
+ .appendFile(dataFile(byP, "s3://b/db/pc/p=8/c.parquet", 1024,
null, "p=8"))
+ .commit();
+ table.updateSpec().removeField("p").addField("q").commit();
+ table.updateSchema().deleteColumn("p").commit();
+ table.newAppend()
+ .appendFile(dataFile(table.spec(),
"s3://b/db/pc/q=1/d.parquet", 1024, null, "q=1"))
+ .commit();
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ scanOf(new IcebergTableHandle("db1", "pc")).build());
+
+ Assertions.assertEquals(4, ranges.size());
+ Assertions.assertEquals(OptionalLong.of(3),
provider.scannedPartitionCount(ranges));
+ }
+
+ /**
+ * A v1 table partitioned by p (and q when {@code keepQ}): p is removed
from the spec, which v1 keeps as
+ * void(p), a file is written under that spec, p is dropped from the
schema and the table is upgraded to v2.
+ */
+ private static Table retainedVoidFieldTable(String name, boolean keepQ) {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "p", Types.IntegerType.get()),
+ Types.NestedField.optional(3, "q", Types.IntegerType.get()));
+ PartitionSpec.Builder spec =
PartitionSpec.builderFor(schema).identity("p");
+ if (keepQ) {
+ spec.identity("q");
+ }
+ Table table = createTable(name, schema, spec.build(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "1"));
+ table.updateSpec().removeField("p").commit();
+ PartitionSpec voidSpec = table.spec();
+ PartitionData partition = new PartitionData(voidSpec.partitionType());
+ if (keepQ) {
+ partition.set(1, 5);
+ }
+ table.newAppend()
+ .appendFile(DataFiles.builder(voidSpec)
+ .withPath("s3://b/db/" + name + "/new.parquet")
+ .withFileSizeInBytes(1024)
+ .withRecordCount(10)
+ .withFormat(FileFormat.PARQUET)
+ .withPartition(partition)
+ .build())
+ .commit();
+ table.updateSchema().deleteColumn("p").commit();
+ table.updateProperties().set(TableProperties.FORMAT_VERSION,
"2").commit();
+ return table;
+ }
+
@Test
public void
planScanKeepsUnpartitionedSpecIdentityAfterEvolvingToPartitioned() {
// Guard for the DML $row_id contract: a file written before
identity(p) was added must still report
@@ -1533,6 +1694,26 @@ public class IcebergScanPlanProviderTest {
.getFields().get(0).getFieldPtr().getName());
}
+ @Test
+ public void getScanNodePropertiesLeavesHiddenRowIdOutOfDict() throws
Exception {
+ // getColumnHandles returns a handle for the hidden DML row-id column
so a DML scan request carries it, but
+ // the column is synthesized by the reader and not in the schema.
MUTATION: keeping it in the requested
+ // names -> "requested column ... not found in the table schema" ->
red.
+ Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
+ IcebergScanPlanProvider provider = providerOver(table);
+ List<ConnectorColumnHandle> columns = Arrays.asList(new
IcebergColumnHandle("id", 1),
+ new IcebergColumnHandle("__DORIS_ICEBERG_ROWID_COL__", -1));
+
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "t1"), columns,
Optional.empty());
+
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+ Assertions.assertEquals(1,
params.getHistorySchemaInfo().get(0).getRootField().getFieldsSize());
+ Assertions.assertEquals("id",
params.getHistorySchemaInfo().get(0).getRootField()
+ .getFields().get(0).getFieldPtr().getName());
+ }
+
@Test
public void populateScanLevelParamsAdvertisesIcebergScanSemanticsV2()
throws Exception {
// #65784: every iceberg scan this connector plans must stamp
iceberg_scan_semantics_version=2 on the
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
new file mode 100644
index 00000000000..7ae07744445
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_delete_unpartitioned_evolution.groovy
@@ -0,0 +1,204 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_delete_unpartitioned_evolution",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ logger.info("disable iceberg test")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_iceberg_delete_unpartitioned_evolution"
+ String dbName = "iceberg_delete_unpartitioned_evolution_db"
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )
+ """
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+
+ def checkRows = { String table, def expected ->
+ sql """refresh table ${dbName}.${table}"""
+ spark_iceberg """refresh table demo.${dbName}.${table}"""
+ def actual = sql """select record_key, metric from ${table} order by
record_key"""
+ assertSparkDorisResultEquals(expected, actual)
+ assertSparkDorisResultEquals(spark_iceberg("""
+ select record_key, metric from demo.${dbName}.${table} order by
record_key
+ """), actual)
+ }
+
+ // REPLACE retains historical spec 0 while writing new files under an
unpartitioned spec.
+ // Spark's replace transaction exercises the same Iceberg metadata
transition as Trino RTAS.
+ spark_iceberg """drop table if exists
demo.${dbName}.replace_to_unpartitioned"""
+ spark_iceberg """
+ create table demo.${dbName}.replace_to_unpartitioned (record_key int,
metric int)
+ using iceberg partitioned by (record_key)
+ tblproperties ('format-version' = '2', 'write.format.default' =
'parquet')
+ """
+ sql """refresh database ${dbName}"""
+ sql """insert into replace_to_unpartitioned values (1, 10), (2, 20), (3,
30)"""
+ spark_iceberg """refresh table demo.${dbName}.replace_to_unpartitioned"""
+ spark_iceberg """
+ create or replace table demo.${dbName}.replace_to_unpartitioned using
iceberg
+ tblproperties ('format-version' = '2', 'write.delete.mode' =
'merge-on-read')
+ as select * from demo.${dbName}.replace_to_unpartitioned
+ """
+ sql """refresh table ${dbName}.replace_to_unpartitioned"""
+ def replacedSpecs = sql """select distinct spec_id from
replace_to_unpartitioned\$data_files"""
+ assertEquals(1, replacedSpecs.size())
+ assertTrue((replacedSpecs[0][0] as int) > 0)
+ def expectedAfterReplaceDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.replace_to_unpartitioned
+ where record_key <> 1 order by record_key
+ """
+ sql """delete from replace_to_unpartitioned where record_key = 1"""
+ checkRows("replace_to_unpartitioned", expectedAfterReplaceDelete)
+ assertEquals(replacedSpecs, sql("""
+ select distinct spec_id from replace_to_unpartitioned\$delete_files
+ """))
+ def expectedAfterRefreshDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.replace_to_unpartitioned
+ where record_key <> 2 order by record_key
+ """
+ sql """delete from replace_to_unpartitioned where record_key = 2"""
+ checkRows("replace_to_unpartitioned", expectedAfterRefreshDelete)
+
+ // Dropping the last partition field leaves both old partitioned and new
unpartitioned files live.
+ spark_iceberg """drop table if exists
demo.${dbName}.mixed_partition_specs"""
+ spark_iceberg """
+ create table demo.${dbName}.mixed_partition_specs (record_key int,
metric int)
+ using iceberg partitioned by (record_key)
+ tblproperties ('format-version' = '2', 'write.format.default' = 'orc',
+ 'write.delete.mode' = 'merge-on-read',
'write.update.mode' = 'merge-on-read')
+ """
+ spark_iceberg """insert into demo.${dbName}.mixed_partition_specs values
(1, 10), (2, 20)"""
+ spark_iceberg """alter table demo.${dbName}.mixed_partition_specs drop
partition field record_key"""
+ spark_iceberg """insert into demo.${dbName}.mixed_partition_specs values
(3, 30), (4, 40)"""
+ sql """refresh database ${dbName}"""
+ def mixedSpecs = sql """select distinct spec_id from
mixed_partition_specs\$data_files order by spec_id"""
+ assertEquals(2, mixedSpecs.size())
+ def expectedAfterMixedDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.mixed_partition_specs
+ where record_key not in (1, 3) order by record_key
+ """
+ sql """delete from mixed_partition_specs where record_key in (1, 3)"""
+ checkRows("mixed_partition_specs", expectedAfterMixedDelete)
+ assertEquals(mixedSpecs, sql("""
+ select distinct spec_id from mixed_partition_specs\$delete_files order
by spec_id
+ """))
+ def expectedAfterUpdate = spark_iceberg """
+ select record_key, metric + 100 from
demo.${dbName}.mixed_partition_specs order by record_key
+ """
+ sql """update mixed_partition_specs set metric = metric + 100"""
+ checkRows("mixed_partition_specs", expectedAfterUpdate)
+
+ // Dropping a source column makes its historical partition type unknown,
even for non-null values.
+ for (String fileFormat : ["parquet", "orc"]) {
+ String table = "dropped_partition_source_${fileFormat}"
+ spark_iceberg """drop table if exists demo.${dbName}.${table}"""
+ spark_iceberg """
+ create table demo.${dbName}.${table} (record_key int,
partition_key int, metric int)
+ using iceberg partitioned by (partition_key)
+ tblproperties ('format-version' = '2', 'write.format.default' =
'${fileFormat}',
+ 'write.delete.mode' = 'merge-on-read',
'write.update.mode' = 'merge-on-read')
+ """
+ spark_iceberg """insert into demo.${dbName}.${table} values (1, 7,
10), (2, 7, 20), (3, null, 30)"""
+ spark_iceberg """alter table demo.${dbName}.${table} drop partition
field partition_key"""
+ spark_iceberg """alter table demo.${dbName}.${table} drop column
partition_key"""
+ spark_iceberg """insert into demo.${dbName}.${table} values (4, 40)"""
+ sql """refresh database ${dbName}"""
+ def expected = spark_iceberg """
+ select record_key, metric from demo.${dbName}.${table} order by
record_key
+ """
+ checkRows(table, expected)
+ test {
+ sql """delete from ${table} where record_key = 1"""
+ exception "Cannot produce Iceberg row IDs with unsupported
partition types"
+ }
+ test {
+ sql """update ${table} set metric = metric + 100 where record_key
= 2"""
+ exception "Cannot produce Iceberg row IDs with unsupported
partition types"
+ }
+ checkRows(table, expected)
+ assertEquals([[0L]], sql("""select count(*) from
${table}\$delete_files"""))
+ // A NULL partition value needs no type, so the file of partition_key
= NULL still takes DML.
+ def expectedAfterNullPartitionDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.${table} where
record_key <> 3 order by record_key
+ """
+ sql """delete from ${table} where record_key = 3"""
+ checkRows(table, expectedAfterNullPartitionDelete)
+ }
+
+ // Historical binary partitions must retain their bytes, not become NULL
delete partitions.
+ spark_iceberg """drop table if exists demo.${dbName}.historical_binary"""
+ spark_iceberg """
+ create table demo.${dbName}.historical_binary (record_key int,
partition_key binary, metric int)
+ using iceberg partitioned by (partition_key)
+ tblproperties ('format-version' = '2', 'write.delete.mode' =
'merge-on-read')
+ """
+ spark_iceberg """insert into demo.${dbName}.historical_binary values
+ (1, unhex('00ff802f'), 10), (2, unhex('00ff802f'), 20), (3, null, 30),
(4, unhex(''), 40)"""
+ spark_iceberg """alter table demo.${dbName}.historical_binary drop
partition field partition_key"""
+ spark_iceberg """insert into demo.${dbName}.historical_binary values (5,
unhex('ff'), 50)"""
+ sql """refresh database ${dbName}"""
+ def expectedAfterBinaryDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.historical_binary
+ where record_key not in (1, 3, 4, 5) order by record_key
+ """
+ sql """delete from historical_binary where record_key in (1, 3, 4, 5)"""
+ checkRows("historical_binary", expectedAfterBinaryDelete)
+
+ // Negative fractional epochs require floor-based splitting into seconds
and nanoseconds.
+ spark_iceberg """drop table if exists
demo.${dbName}.historical_timestamp"""
+ spark_iceberg """
+ create table demo.${dbName}.historical_timestamp
+ (record_key int, partition_key timestamp_ntz, metric int)
+ using iceberg partitioned by (partition_key)
+ tblproperties ('format-version' = '2', 'write.delete.mode' =
'merge-on-read',
+ 'write.update.mode' = 'merge-on-read')
+ """
+ spark_iceberg """insert into demo.${dbName}.historical_timestamp values
+ (1, cast('1969-12-31 23:59:59.999999' as timestamp_ntz), 10),
+ (2, cast('1969-12-31 23:59:58.999999' as timestamp_ntz), 20), (3,
null, 30)"""
+ spark_iceberg """alter table demo.${dbName}.historical_timestamp drop
partition field partition_key"""
+ spark_iceberg """insert into demo.${dbName}.historical_timestamp values
+ (4, cast('1970-01-01 00:00:00.000001' as timestamp_ntz), 40)"""
+ sql """refresh database ${dbName}"""
+ def expectedAfterTimestampDelete = spark_iceberg """
+ select record_key, metric from demo.${dbName}.historical_timestamp
+ where record_key not in (1, 4) order by record_key
+ """
+ sql """delete from historical_timestamp where record_key in (1, 4)"""
+ checkRows("historical_timestamp", expectedAfterTimestampDelete)
+ def expectedAfterTimestampUpdate = spark_iceberg """
+ select record_key, metric + 100 from
demo.${dbName}.historical_timestamp order by record_key
+ """
+ sql """update historical_timestamp set metric = metric + 100"""
+ checkRows("historical_timestamp", expectedAfterTimestampUpdate)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
