This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 249c1ba8e6a [fix](iceberg) keep identity partition values of old specs
after evolving to unpartitioned (#68124)
249c1ba8e6a is described below
commit 249c1ba8e6a8fedc39aea56f4396df7de7305fd9
Author: daidai <[email protected]>
AuthorDate: Sun Sep 20 10:21:56 2026 +0800
[fix](iceberg) keep identity partition values of old specs after evolving
to unpartitioned (#68124)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
After an Iceberg table's default partition spec evolves to
unpartitioned, split planning gated the per-file partition metadata on
the current spec (`table.spec().isPartitioned()`), so data files written
under an older identity spec stopped carrying their partition values.
`path_partition_keys` is the union of all specs, so BE does not read
those columns from the data file either, and an identity partition
column whose value only exists in the file's partition metadata (e.g. a
table migrated with `add_files`) was read as NULL, with both
`enable_file_scanner_v2=true` and `false`.
Gate on whether any spec of the table is partitioned instead. The
values, spec id and partition data are already derived from the spec
each data file was written with, which is what Iceberg's
`PartitionUtil.constantsMap` does.
### Release note
None
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../connector/iceberg/IcebergPartitionUtils.java | 14 +++
.../connector/iceberg/IcebergScanPlanProvider.java | 12 ++-
.../doris/connector/iceberg/IcebergScanRange.java | 22 +++-
.../iceberg/IcebergScanPlanProviderTest.java | 111 +++++++++++++++++++++
4 files changed, 156 insertions(+), 3 deletions(-)
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 9d9738e130c..e38f73f3e5e 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
@@ -137,6 +137,20 @@ final class IcebergPartitionUtils {
return new ArrayList<>(partitionColumns);
}
+ /**
+ * Whether <b>any</b> partition spec of the table is partitioned. Scan
planning must use this instead of
+ * the current default spec: after evolving to an unpartitioned spec, data
files written under an older
+ * partitioned spec still carry their partition metadata (identity values,
spec id, partition data).
+ */
+ static boolean hasPartitionedSpec(Table table) {
+ for (PartitionSpec spec : table.specs().values()) {
+ if (spec.isPartitioned()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Per-file map of identity partition column (case-preserved) to
serialized value, skipping non-identity
* transforms and BINARY/FIXED columns (utf8 round-trip would corrupt
those). Order-preserving
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 5d1f12a57fb..6b3955d38be 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
@@ -548,7 +548,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
int formatVersion = getFormatVersion(table);
List<String> orderedPartitionKeys =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
ZoneId zone = resolveSessionZone(session);
- boolean partitioned = table.spec().isPartitioned();
+ boolean partitioned = IcebergPartitionUtils.hasPartitionedSpec(table);
Map<String, String> vendedToken = context != null
? extractVendedToken(table, restVendedCredentialsEnabled()) :
Collections.emptyMap();
UnaryOperator<String> uriNormalizer = newUriNormalizer(vendedToken);
@@ -736,7 +736,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
int formatVersion = getFormatVersion(table);
List<String> orderedPartitionKeys =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
ZoneId zone = resolveSessionZone(session);
- boolean partitioned = table.spec().isPartitioned();
+ boolean partitioned = IcebergPartitionUtils.hasPartitionedSpec(table);
// Vended credentials (T09): extract the per-table REST vended token
ONCE per scan (gated on the catalog
// flag iceberg.rest.vended-credentials-enabled, mirroring legacy
IcebergVendedCredentialsProvider), then
@@ -1583,6 +1583,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private Integer partitionSpecId;
private String partitionDataJson;
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.
+ private boolean countsAsScannedPartition;
private List<IcebergScanRange.DeleteFile> deleteCarriers =
Collections.emptyList();
private String fileFormat;
private Long firstRowId;
@@ -1636,6 +1639,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
.firstRowId(file.firstRowId)
.lastUpdatedSequenceNumber(file.lastUpdatedSequenceNumber)
.partitionValues(file.partitionValues)
+ .countsAsScannedPartition(file.countsAsScannedPartition)
.deleteFiles(file.deleteCarriers)
.pushDownRowCount(pushDownRowCount)
.selfSplitWeight(selfSplitWeight)
@@ -1704,6 +1708,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
file.partitionSpecId = partitionSpecId;
file.partitionDataJson = partitionDataJson;
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
+ // start showing up in EXPLAIN partition=N/M or in a sql_block_rule
partition_num check.
+ file.countsAsScannedPartition = table.spec().isPartitioned();
file.fileFormat = fileFormat;
file.firstRowId = firstRowId;
file.lastUpdatedSequenceNumber = lastUpdatedSequenceNumber;
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 6a4a6316830..9d8fd831273 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
@@ -67,6 +67,13 @@ public class IcebergScanRange implements ConnectorScanRange {
// Identity partition column (lowercased) -> serialized value, already
ordered as the path_partition_keys
// list, filtered to keys this file carries. Drives columns-from-path.
Never null (empty when unpartitioned).
private final Map<String, String> partitionValues;
+ // Whether this range counts toward the EXPLAIN `partition=N/M` /
sql_block_rule partition_num display
+ // (see getScannedPartitionKey). False ONLY when the table's CURRENT spec
is unpartitioned: such a table
+ // reports no partitions at all (listPartitions is empty), so counting the
partitions of files written
+ // under an older spec would change `partition=N/M` and could newly trip a
partition_num block rule —
+ // display metadata the DORIS-29056 read fix must not alter. The partition
VALUES those files carry are
+ // unaffected: they are per-file read data, not display.
+ private final boolean countsAsScannedPartition;
// Merge-on-read delete files applying to this data file (T04). Never null
(empty when none / v1).
private final List<DeleteFile> deleteFiles;
// COUNT(*) pushdown precomputed row count (T05): -1 = no precomputed
count (the normal scan path);
@@ -132,6 +139,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
this.partitionValues = builder.partitionValues != null
? Collections.unmodifiableMap(builder.partitionValues)
: Collections.emptyMap();
+ this.countsAsScannedPartition = builder.countsAsScannedPartition;
this.deleteFiles = builder.deleteFiles != null
? Collections.unmodifiableList(builder.deleteFiles)
: Collections.emptyList();
@@ -226,7 +234,7 @@ public class IcebergScanRange implements ConnectorScanRange
{
* {@code IcebergScanPlanProvider} counts distinct non-null keys for
{@code selectedPartitionNum}.
*/
String getScannedPartitionKey() {
- if (partitionDataJson == null) {
+ if (partitionDataJson == null || !countsAsScannedPartition) {
return null;
}
return partitionSpecId + "|" + partitionDataJson;
@@ -467,6 +475,9 @@ public class IcebergScanRange implements ConnectorScanRange
{
private Long firstRowId;
private Long lastUpdatedSequenceNumber;
private Map<String, String> partitionValues;
+ // Default true = legacy behavior (a range carrying PartitionData
counts as a scanned partition); the
+ // data path passes the table's CURRENT spec isPartitioned().
+ private boolean countsAsScannedPartition = true;
private List<DeleteFile> deleteFiles;
private long pushDownRowCount = -1;
private String serializedSplit;
@@ -552,6 +563,15 @@ public class IcebergScanRange implements
ConnectorScanRange {
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()}.
+ */
+ public Builder countsAsScannedPartition(boolean
countsAsScannedPartition) {
+ this.countsAsScannedPartition = countsAsScannedPartition;
+ return this;
+ }
+
public Builder firstRowId(Long firstRowId) {
this.firstRowId = firstRowId;
return this;
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 27b453c9c5c..0a5a1b3ecbc 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
@@ -98,6 +98,7 @@ import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.UnaryOperator;
@@ -755,6 +756,116 @@ public class IcebergScanPlanProviderTest {
"file b's slices carry p=2 (no cross-file staleness)");
}
+ @Test
+ public void
planScanKeepsOldSpecIdentityValuesAfterEvolvingToUnpartitioned() {
+ // DORIS-29056 repro: a file written under identity(p) must still
carry p=7 after the table's default
+ // spec evolves to unpartitioned; otherwise BE fills p with NULL when
the file does not store p.
+ 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();
+ Assertions.assertTrue(table.spec().isUnpartitioned());
+
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ ConnectorScanRequest.builder(new IcebergTableHandle("db1",
"pt"), Collections.emptyList())
+ .build());
+
+ Assertions.assertEquals(1, ranges.size());
+ ConnectorScanRange range = ranges.get(0);
+ Assertions.assertEquals(Collections.singletonMap("p", "7"),
range.getPartitionValues());
+ TFileRangeDesc desc = populate(range);
+ Assertions.assertEquals(Collections.singletonList("p"),
desc.getColumnsFromPathKeys());
+ Assertions.assertEquals(Collections.singletonList("7"),
desc.getColumnsFromPath());
+ Assertions.assertEquals(0,
desc.getTableFormatParams().getIcebergParams().getPartitionSpecId());
+ Assertions.assertEquals("[\"7\"]",
desc.getTableFormatParams().getIcebergParams().getPartitionDataJson());
+ // Display parity: the table's CURRENT spec is unpartitioned, so it
still reports no scanned
+ // partitions — the read fix must not change EXPLAIN partition=N/M or
sql_block_rule partition_num.
+ Assertions.assertEquals(OptionalLong.empty(),
provider.scannedPartitionCount(ranges));
+ }
+
+ @Test
+ public void
planScanCountsScannedPartitionsWhileCurrentSpecStaysPartitioned() {
+ // The other side of the display gate: with a partitioned CURRENT
spec, files of an older spec keep
+ // counting toward selectedPartitionNum exactly as before (legacy
partitionMapInfos parity).
+ 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/a.parquet", 1024,
null, "p=7"))
+ .appendFile(dataFile(spec, "s3://b/db/pt/p=8/b.parquet", 1024,
null, "p=8"))
+ .commit();
+ Assertions.assertTrue(table.spec().isPartitioned());
+
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ ConnectorScanRequest.builder(new IcebergTableHandle("db1",
"pt"), Collections.emptyList())
+ .build());
+
+ Assertions.assertEquals(OptionalLong.of(2L),
provider.scannedPartitionCount(ranges));
+ }
+
+ @Test
+ public void
streamSplitsKeepsOldSpecIdentityValuesAfterEvolvingToUnpartitioned() throws
IOException {
+ // The lazy (batch-mode) source computes its own partitioned flag, so
the eager test above does not
+ // pin it: reverting only streamSplits' gate to the current spec would
silently bring the NULL read
+ // back for batch-mode scans. Drain the source and assert the same
per-file partition metadata.
+ 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();
+ Assertions.assertTrue(table.spec().isUnpartitioned());
+
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ List<ConnectorScanRange> ranges = new ArrayList<>();
+ try (ConnectorSplitSource source = provider.streamSplits(
+ new FakeScanSession("UTC", Collections.emptyMap()),
+ new IcebergTableHandle("db1", "pt"), Collections.emptyList(),
Optional.empty(), -1L)) {
+ while (source.hasNext()) {
+ ranges.add(source.next());
+ }
+ }
+
+ Assertions.assertEquals(1, ranges.size());
+ Assertions.assertEquals(Collections.singletonMap("p", "7"),
ranges.get(0).getPartitionValues());
+ TFileRangeDesc desc = populate(ranges.get(0));
+ Assertions.assertEquals(Collections.singletonList("p"),
desc.getColumnsFromPathKeys());
+ Assertions.assertEquals(Collections.singletonList("7"),
desc.getColumnsFromPath());
+ Assertions.assertEquals(0,
desc.getTableFormatParams().getIcebergParams().getPartitionSpecId());
+ Assertions.assertEquals("[\"7\"]",
desc.getTableFormatParams().getIcebergParams().getPartitionDataJson());
+ }
+
+ @Test
+ public void
planScanKeepsUnpartitionedSpecIdentityAfterEvolvingToPartitioned() {
+ // Guard for the DML $row_id contract: a file written before
identity(p) was added must still report
+ // spec 0 with an (empty) partition_data_json, so BE commits its
delete file under spec 0 instead of
+ // falling back to the current partitioned spec.
+ Table table = createTable("pt", PART_SCHEMA,
PartitionSpec.unpartitioned());
+ table.newAppend()
+ .appendFile(dataFile(table.spec(), "s3://b/db/pt/old.parquet",
1024, null, null))
+ .commit();
+ table.updateSpec().addField("p").commit();
+ Assertions.assertTrue(table.spec().isPartitioned());
+
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ List<ConnectorScanRange> ranges = provider.planScan(new
FakeScanSession("UTC", Collections.emptyMap()),
+ ConnectorScanRequest.builder(new IcebergTableHandle("db1",
"pt"), Collections.emptyList())
+ .build());
+
+ Assertions.assertEquals(1, ranges.size());
+ TFileRangeDesc desc = populate(ranges.get(0));
+ Assertions.assertTrue(ranges.get(0).getPartitionValues().isEmpty());
+ Assertions.assertEquals(0,
desc.getTableFormatParams().getIcebergParams().getPartitionSpecId());
+ Assertions.assertEquals("[]",
desc.getTableFormatParams().getIcebergParams().getPartitionDataJson());
+ }
+
// ── M-2: size-proportional BE scheduling weight (selfSplitWeight /
targetSplitSize) ──
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
