Skip to content

Commit a7ef23f

Browse files
hubgeterhubgeter
committed
branch-4.1: [fix](iceberg) keep identity partition values of old specs after evolving to unpartitioned #68124
Cherry-picked from #68124. branch-4.1 has no fe-connector modules, so the same fix is applied to the legacy IcebergScanNode instead of IcebergScanPlanProvider, and it needs one branch-4.1-only addition (see below). When a table's default partition spec evolves to unpartitioned, split planning gated the per-file partition metadata on `icebergTable.spec().isPartitioned()`, so data files written under an older identity spec no longer carried their partition values. Gate on whether ANY spec of the table is partitioned instead; the values, spec id and partition data are already computed from the spec each data file was written with, which is what Iceberg's PartitionUtil.constantsMap does. That alone does not fix the read on this branch. Unlike master, which classifies the identity columns of ALL specs as partition keys, branch-4.1 keys the classification off the columns COMMON to every spec, which is empty once the default spec is unpartitioned. BE then never applies the split value: the v2 column mapper uses a partition constant only for a column FE marked as a partition key. So for scanner v2 the classification now uses the same all-specs list as master. Files that carry no value for such a column are unaffected — the v2 mapper falls back to the physical field. Scanner v1 keeps the common-columns list: it has no such per-file fallback, so widening it there would turn files that currently read the column from the file into NULLs. Behavior change, same as master: a file that has neither a split partition value nor the physical column now fails with "does not have a matching partition value" instead of reading NULL. The scanned-partition count keeps its previous meaning: partitionMapInfos now also collects the partitions of old-spec files because it doubles as the identity-value cache, so selectedPartitionNum / numApproximateSplits go through scannedPartitionNum(), which reports nothing unless the CURRENT spec is partitioned.
1 parent 9f181d9 commit a7ef23f

4 files changed

Lines changed: 148 additions & 12 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -982,6 +982,20 @@ public static Map<String, String> getPartitionInfoMap(PartitionData partitionDat
982982
return partitionInfoMap;
983983
}
984984

985+
/**
986+
* Whether <b>any</b> partition spec of the table is partitioned. Scan planning must use this instead of
987+
* the current default spec: after evolving to an unpartitioned spec, data files written under an older
988+
* partitioned spec still carry their partition metadata (identity values, spec id, partition data).
989+
*/
990+
public static boolean hasPartitionedSpec(Table table) {
991+
for (PartitionSpec spec : table.specs().values()) {
992+
if (spec.isPartitioned()) {
993+
return true;
994+
}
995+
}
996+
return false;
997+
}
998+
985999
public static List<String> getIdentityPartitionColumns(Table table) {
9861000
return getIdentityPartitionColumns(table, false, false);
9871001
}

‎fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java‎

Lines changed: 43 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -199,8 +199,15 @@ public class IcebergScanNode extends FileQueryScanNode {
199199
private List<String> orderedPartitionMetadataKeys;
200200
private boolean enableMappingVarbinaryForPartitionMetadata;
201201
private boolean enableMappingTimestampTzForPartitionMetadata;
202+
// Whether the table's CURRENT default spec is partitioned. Display/accounting only: it gates what
203+
// counts as a scanned partition (selectedPartitionNum, numApproximateSplits), preserving the legacy
204+
// numbers for a table that evolved to an unpartitioned spec.
202205
private boolean isPartitionedTable;
203206
private Boolean requiresRowId;
207+
// Whether ANY spec of the table is partitioned. This is the READ gate: data files written under an
208+
// older partitioned spec still carry their identity values, and must keep sending them even after
209+
// the default spec evolved to unpartitioned.
210+
private boolean hasPartitionedSpec;
204211
private int formatVersion;
205212
private ExecutionAuthenticator preExecutionAuthenticator;
206213
private IcebergRuntimeContext runtimeContext;
@@ -318,6 +325,7 @@ protected void doInitialize() throws UserException {
318325
partitionMapInfos = new HashMap<>();
319326
initializePartitionMetadata();
320327
isPartitionedTable = icebergTable.spec().isPartitioned();
328+
hasPartitionedSpec = IcebergUtils.hasPartitionedSpec(icebergTable);
321329
// Metadata tables (system tables) are not BaseTable instances, so we need to handle this case
322330
if (icebergTable instanceof BaseTable) {
323331
formatVersion = ((BaseTable) icebergTable).operations().current().formatVersion();
@@ -510,18 +518,26 @@ private void initializePartitionMetadata() {
510518
}
511519
enableMappingVarbinaryForPartitionMetadata = getEnableMappingVarbinary();
512520
enableMappingTimestampTzForPartitionMetadata = getEnableMappingTimestampTz();
513-
orderedPathPartitionKeys = Collections.unmodifiableList(
514-
IcebergUtils.getCommonIdentityPartitionColumns(icebergTable,
515-
enableMappingVarbinaryForPartitionMetadata,
516-
enableMappingTimestampTzForPartitionMetadata));
517521
if (sessionVariable.enableFileScannerV2) {
518-
orderedPartitionMetadataKeys = Collections.unmodifiableList(
522+
// Classify every identity column of EVERY spec as a partition key, like master does. A file
523+
// written under an older identity spec carries its value in the manifest only, and BE uses a
524+
// split partition value just for a column FE marked as a partition key, so the intersection
525+
// below would leave that value unused and the column read as NULL. Files that carry no value
526+
// for the column (a spec without it) are unaffected: the v2 column mapper falls back to the
527+
// physical field when the split has no partition value for a partition key.
528+
orderedPathPartitionKeys = Collections.unmodifiableList(
519529
IcebergUtils.getIdentityPartitionColumns(icebergTable,
520530
enableMappingVarbinaryForPartitionMetadata,
521531
enableMappingTimestampTzForPartitionMetadata));
522532
} else {
523-
orderedPartitionMetadataKeys = orderedPathPartitionKeys;
533+
// v1 has no such per-file fallback: a partition key it cannot fill from the split is not read
534+
// from the file either. Keep the columns common to all specs so those files stay readable.
535+
orderedPathPartitionKeys = Collections.unmodifiableList(
536+
IcebergUtils.getCommonIdentityPartitionColumns(icebergTable,
537+
enableMappingVarbinaryForPartitionMetadata,
538+
enableMappingTimestampTzForPartitionMetadata));
524539
}
540+
orderedPartitionMetadataKeys = orderedPathPartitionKeys;
525541
}
526542

527543
@VisibleForTesting
@@ -2611,7 +2627,10 @@ private Split createIcebergSplit(FileScanTask fileScanTask) throws UserException
26112627
split.setPartitionSpecId(specId);
26122628
PartitionData partitionData = (PartitionData) dataFile.partition();
26132629
boolean includePartitionData = requiresRowId();
2614-
if (partitionData != null && (includePartitionData || isPartitionedTable)) {
2630+
// Gate identity values on the table's spec HISTORY, not on the current default spec: a file written
2631+
// under an older identity spec keeps its partition values in the manifest, and they are the only
2632+
// source for an identity partition column that the physical file does not store.
2633+
if (partitionData != null && (includePartitionData || hasPartitionedSpec)) {
26152634
PartitionSpec partitionSpec = icebergTable.specs().get(specId);
26162635
Preconditions.checkNotNull(partitionSpec, "Partition spec with specId %s not found for table %s",
26172636
specId, icebergTable.name());
@@ -2626,7 +2645,7 @@ private Split createIcebergSplit(FileScanTask fileScanTask) throws UserException
26262645
+ specId + ". Rewrite historical data files before DELETE or UPDATE.", e);
26272646
}
26282647
}
2629-
if (isPartitionedTable) {
2648+
if (hasPartitionedSpec) {
26302649
Map<String, String> partitionInfoMap = partitionMapInfos.computeIfAbsent(
26312650
Pair.of(specId, partitionData), k -> IcebergUtils.getIdentityPartitionInfoMap(
26322651
partitionData, partitionSpec, icebergTable, sessionVariable.getTimeZone(),
@@ -2639,7 +2658,7 @@ private Split createIcebergSplit(FileScanTask fileScanTask) throws UserException
26392658
split.setIcebergPartitionValues(partitionInfoMap);
26402659
}
26412660
}
2642-
} else if (partitionData == null && isPartitionedTable) {
2661+
} else if (partitionData == null && hasPartitionedSpec) {
26432662
partitionMapInfos.put(Pair.of(specId, null), Collections.emptyMap());
26442663
}
26452664
return split;
@@ -2808,7 +2827,7 @@ private List<Split> doGetSplits(int numBackends) throws UserException {
28082827
for (FileScanTask task : customFileScanTasks) {
28092828
splits.add(createIcebergSplit(task));
28102829
}
2811-
selectedPartitionNum = partitionMapInfos.size();
2830+
selectedPartitionNum = scannedPartitionNum();
28122831
recordManifestCacheProfile();
28132832
return splits;
28142833
}
@@ -2845,7 +2864,7 @@ private List<Split> doGetSplits(int numBackends) throws UserException {
28452864
}
28462865
}
28472866

2848-
selectedPartitionNum = partitionMapInfos.size();
2867+
selectedPartitionNum = scannedPartitionNum();
28492868
recordManifestCacheProfile();
28502869
return splits;
28512870
}
@@ -3228,9 +3247,21 @@ private void assignCountToSplits(List<Split> splits, long totalCount) {
32283247
((IcebergSplit) splits.get(size - 1)).setTableLevelRowCount(countPerSplit + totalCount % size);
32293248
}
32303249

3250+
/**
3251+
* The number of scanned partitions reported to EXPLAIN ({@code partition=N/M}), the
3252+
* {@code sql_block_rule} {@code partition_num} guard and the batch-mode split estimate. A table whose
3253+
* CURRENT spec is unpartitioned reports no partitions at all, so the partitions that files of an older
3254+
* spec still carry must not start being counted — {@code partitionMapInfos} now collects them because
3255+
* it doubles as the per-(spec, partition) identity-value cache for the read path.
3256+
*/
3257+
private int scannedPartitionNum() {
3258+
return isPartitionedTable ? partitionMapInfos.size() : 0;
3259+
}
3260+
32313261
@Override
32323262
public int numApproximateSplits() {
3233-
return NUM_SPLITS_PER_PARTITION * partitionMapInfos.size() > 0 ? partitionMapInfos.size() : 1;
3263+
int partitions = scannedPartitionNum();
3264+
return NUM_SPLITS_PER_PARTITION * partitions > 0 ? partitions : 1;
32343265
}
32353266

32363267
private Optional<NotSupportedException> checkNotSupportedException(Exception e) {

‎fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/IcebergUtilsTest.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -427,6 +427,25 @@ public void testMalformedNameMappingFailsInsteadOfFallingBackToCurrentSchemaName
427427
Assert.assertTrue(exception.getMessage().contains("db.tbl"));
428428
}
429429

430+
@Test
431+
public void testHasPartitionedSpecLooksAtEverySpecNotOnlyTheCurrentOne() {
432+
Schema schema = new Schema(
433+
Types.NestedField.required(1, "id", Types.IntegerType.get()),
434+
Types.NestedField.required(2, "p", Types.IntegerType.get()));
435+
PartitionSpec identity = PartitionSpec.builderFor(schema).identity("p").withSpecId(0).build();
436+
PartitionSpec unpartitioned = PartitionSpec.unpartitioned();
437+
438+
// A table that evolved identity(p) -> unpartitioned still has files carrying p in their
439+
// partition metadata, so scan planning must keep sending those values.
440+
Table evolved = Mockito.mock(Table.class);
441+
Mockito.when(evolved.specs()).thenReturn(ImmutableMap.of(0, identity, 1, unpartitioned));
442+
Assert.assertTrue(IcebergUtils.hasPartitionedSpec(evolved));
443+
444+
Table neverPartitioned = Mockito.mock(Table.class);
445+
Mockito.when(neverPartitioned.specs()).thenReturn(ImmutableMap.of(0, unpartitioned));
446+
Assert.assertFalse(IcebergUtils.hasPartitionedSpec(neverPartitioned));
447+
}
448+
430449
@Test
431450
public void testEmptyNameMappingStillParsesAsAuthoritativeMapping() throws Exception {
432451
Table table = Mockito.mock(Table.class);

‎fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1816,6 +1816,68 @@ public void testSchemaCarrierKeepsDroppedEqualityFieldDefault() throws Exception
18161816
IcebergUtils.getSerializedInitialDefaults(fields, false).get(7));
18171817
}
18181818

1819+
@Test
1820+
public void testSplitKeepsOldSpecIdentityValuesAfterEvolvingToUnpartitioned() throws Exception {
1821+
// A file written under identity(p) keeps p=7 only in its manifest partition metadata. Gating on
1822+
// the CURRENT spec drops that value once the default spec evolves to unpartitioned, and keying
1823+
// the partition-key classification off the columns common to all specs leaves BE reading p from
1824+
// a file that does not store it, so the column comes back NULL in both cases.
1825+
Schema schema = new Schema(
1826+
Types.NestedField.required(1, "id", Types.IntegerType.get()),
1827+
Types.NestedField.required(2, "p", Types.IntegerType.get()));
1828+
PartitionSpec spec = PartitionSpec.builderFor(schema).identity("p").build();
1829+
String location = temporaryFolder.newFolder("identity_evolved_table").toURI().toString();
1830+
Table table = new HadoopTables(new Configuration()).create(schema, spec, location);
1831+
table.newFastAppend()
1832+
.appendFile(DataFiles.builder(table.spec())
1833+
.withPath(location + "data/p=7/old.parquet")
1834+
.withFileSizeInBytes(1024L)
1835+
.withRecordCount(2L)
1836+
.withFormat(FileFormat.PARQUET)
1837+
.withPartitionPath("p=7")
1838+
.build())
1839+
.commit();
1840+
table.updateSpec().removeField("p").commit();
1841+
Assert.assertTrue(table.spec().isUnpartitioned());
1842+
1843+
SessionVariable sessionVariable = new SessionVariable();
1844+
sessionVariable.enableFileScannerV2 = true;
1845+
TestIcebergScanNode node = new TestIcebergScanNode(sessionVariable);
1846+
setIcebergTable(node, table);
1847+
setPrivateField(node, "formatVersion", 2);
1848+
setPrivateField(node, "storagePropertiesMap", Collections.emptyMap());
1849+
setPrivateField(node, "partitionMapInfos", new HashMap<>());
1850+
// Exactly what doInitialize computes for this table.
1851+
setPrivateField(node, "isPartitionedTable", table.spec().isPartitioned());
1852+
setPrivateField(node, "hasPartitionedSpec", IcebergUtils.hasPartitionedSpec(table));
1853+
1854+
// p must stay a partition key even though the current spec no longer mentions it; otherwise BE
1855+
// ignores the split value and reads the (absent) physical column.
1856+
Assert.assertEquals(ImmutableList.of("p"), orderedPathPartitionKeys(node));
1857+
1858+
FileScanTask task;
1859+
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
1860+
Iterator<FileScanTask> iterator = tasks.iterator();
1861+
Assert.assertTrue(iterator.hasNext());
1862+
task = iterator.next();
1863+
}
1864+
1865+
IcebergSplit split = createIcebergSplit(node, task);
1866+
Assert.assertEquals(ImmutableMap.of("p", "7"), split.getIcebergPartitionValues());
1867+
Assert.assertEquals(Integer.valueOf(0), split.getPartitionSpecId());
1868+
Assert.assertEquals("[\"7\"]", split.getPartitionDataJson());
1869+
1870+
// What BE actually receives: p=7 as a columns-from-path constant for this range.
1871+
TFileRangeDesc rangeDesc = new TFileRangeDesc();
1872+
setIcebergParams(node, rangeDesc, split);
1873+
Assert.assertEquals(ImmutableList.of("p"), rangeDesc.getColumnsFromPathKeys());
1874+
Assert.assertEquals(ImmutableList.of("7"), rangeDesc.getColumnsFromPath());
1875+
1876+
// Display parity: the CURRENT spec is unpartitioned, so the table still reports no scanned
1877+
// partitions and the batch-mode estimate stays at the unpartitioned default.
1878+
Assert.assertEquals(1, node.numApproximateSplits());
1879+
}
1880+
18191881
@Test
18201882
public void testBatchSplitCarriesDroppedEqualitySchemaThroughThrift() throws Exception {
18211883
Types.NestedField id = Types.NestedField.required(1, "id", Types.LongType.get());
@@ -1950,6 +2012,7 @@ public void testReadSplitsSkipPartitionJson() throws Exception {
19502012
node.addSlot(0, new Column("record_key", Type.INT));
19512013
setIcebergTable(node, table);
19522014
setPrivateField(node, "isPartitionedTable", keepPartitioned);
2015+
setPrivateField(node, "hasPartitionedSpec", IcebergUtils.hasPartitionedSpec(table));
19532016
setPrivateField(node, "storagePropertiesMap", Collections.emptyMap());
19542017
setPrivateField(node, "formatVersion", 2);
19552018
setPrivateField(node, "partitionMapInfos", new HashMap<>());
@@ -2035,6 +2098,7 @@ private void assertSplitAfterDroppingPartitionSourceColumn(boolean keepPartition
20352098
TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable());
20362099
setIcebergTable(node, table);
20372100
setPrivateField(node, "isPartitionedTable", keepPartitioned);
2101+
setPrivateField(node, "hasPartitionedSpec", IcebergUtils.hasPartitionedSpec(table));
20382102
setPrivateField(node, "storagePropertiesMap", Collections.emptyMap());
20392103
setPrivateField(node, "formatVersion", 2);
20402104
setPrivateField(node, "partitionMapInfos", new HashMap<>());
@@ -2173,6 +2237,7 @@ private void assertDeleteSplitPartitionMetadata(Table table, SessionVariable ses
21732237
node.addSlot(0, IcebergRowId.createHiddenColumn());
21742238
setIcebergTable(node, table);
21752239
setPrivateField(node, "isPartitionedTable", table.spec().isPartitioned());
2240+
setPrivateField(node, "hasPartitionedSpec", IcebergUtils.hasPartitionedSpec(table));
21762241
setPrivateField(node, "storagePropertiesMap", Collections.emptyMap());
21772242
setPrivateField(node, "formatVersion", 2);
21782243
setPrivateField(node, "partitionMapInfos", new HashMap<>());
@@ -4013,6 +4078,13 @@ private static FileScanTask fileScanTask(String path, DeleteFile deleteFile) {
40134078
return task;
40144079
}
40154080

4081+
@SuppressWarnings("unchecked")
4082+
private static List<String> orderedPathPartitionKeys(IcebergScanNode node) throws Exception {
4083+
Method method = IcebergScanNode.class.getDeclaredMethod("getOrderedPathPartitionKeys");
4084+
method.setAccessible(true);
4085+
return (List<String>) method.invoke(node);
4086+
}
4087+
40164088
private static IcebergSplit createIcebergSplit(IcebergScanNode node, FileScanTask task)
40174089
throws Exception {
40184090
Method method = IcebergScanNode.class.getDeclaredMethod("createIcebergSplit", FileScanTask.class);

0 commit comments

Comments
 (0)