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 aae599269eb [fix](iceberg) Align scan schema carrier names by field id
(#69382)
aae599269eb is described below
commit aae599269eb740e9234a1586e78b363c6b3da78c
Author: daidai <[email protected]>
AuthorDate: Thu Oct 22 11:23:59 2026 +0800
[fix](iceberg) Align scan schema carrier names by field id (#69382)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
Iceberg scan carriers could use inconsistent top-level name casing or
schema generations, especially for full-schema, Top-N, and
snapshot-pinned partition scans. Build all carriers from one
authoritative scan schema and resolve partition columns by Iceberg field
id.
### Release note
Fix mixed-case and time-travel Iceberg reads by keeping scan-schema
carrier names consistent.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] 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
Regression test: `test_iceberg_scan_schema_carrier_names` passed twice,
including normal output comparison.
Unit tests: Iceberg connector suite passed (1478 tests, 0 failures, 1
skipped).
Manual test: verified mixed-case ORC, Top-N, pinned snapshot partition
rename/filter, and count pushdown with Spark, Iceberg REST, and Doris.
- 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/IcebergColumnHandle.java | 4 +-
.../connector/iceberg/IcebergPartitionUtils.java | 32 ++--
.../connector/iceberg/IcebergScanPlanProvider.java | 163 +++++++++++---------
.../doris/connector/iceberg/IcebergScanRange.java | 2 +-
.../connector/iceberg/IcebergSchemaUtils.java | 46 +++---
.../iceberg/IcebergPartitionUtilsTest.java | 57 ++++++-
.../iceberg/IcebergScanPlanProviderTest.java | 71 +++++++++
.../connector/iceberg/IcebergSchemaUtilsTest.java | 44 +++++-
.../test_iceberg_scan_schema_carrier_names.out | 37 +++++
.../test_iceberg_scan_schema_carrier_names.groovy | 166 +++++++++++++++++++++
10 files changed, 503 insertions(+), 119 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergColumnHandle.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergColumnHandle.java
index 8bd00b87fb0..6c780ccdfe5 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergColumnHandle.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergColumnHandle.java
@@ -26,13 +26,13 @@ import java.util.Set;
/**
* Iceberg {@link ConnectorColumnHandle}, mirroring the paimon connector's
{@code PaimonColumnHandle}. Carries
- * the (lowercased) column name and the iceberg field id. {@code
IcebergConnectorMetadata.getColumnHandles}
+ * the case-preserved column name and the iceberg field id. {@code
IcebergConnectorMetadata.getColumnHandles}
* produces these so {@code PluginDrivenScanNode.buildColumnHandles} can hand
the provider the pruned set of
* requested columns — which the field-id schema dictionary (T06) keys the
{@code current_schema_id = -1}
* entry off (the CI #969249 fix: the dict's top-level names == the BE
scan-slot names BY CONSTRUCTION).
*
* <p>Equality/hashCode are by name only (mirrors {@code PaimonColumnHandle}):
a handle identifies a column
- * by its (lowercased) name, the same key {@code
allHandles.get(slot.getColumn().getName())} looks it up by.
+ * by its case-preserved name, the same key {@code
allHandles.get(slot.getColumn().getName())} looks it up by.
*/
public class IcebergColumnHandle implements ConnectorColumnHandle {
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 d9035efb111..7c6d929d603 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
@@ -118,19 +118,21 @@ final class IcebergPartitionUtils {
/**
* Ordered, case-preserved, de-duplicated list of the identity partition
column names across <b>all</b>
- * partition specs of the table (mirrors legacy {@code
IcebergUtils.getIdentityPartitionColumns}). This
- * is the {@code path_partition_keys} payload: it tells FE which slots are
partition columns so they are
- * excluded from the file-decode set (the CI #968880 double-fill guard).
Non-identity transforms
- * (bucket/truncate/year/month/...) are excluded.
+ * partition specs of the table (mirrors legacy {@code
IcebergUtils.getIdentityPartitionColumns}). Names
+ * are resolved by field id against the query's authoritative {@code
scanSchema}, so current and pinned
+ * reads cannot mix schema generations. This is the {@code
path_partition_keys} payload: it tells FE which
+ * slots are partition columns so they are excluded from the file-decode
set (the CI #968880 double-fill
+ * guard). Non-identity transforms (bucket/truncate/year/month/...) are
excluded.
*/
- static List<String> getIdentityPartitionColumns(Table table) {
+ static List<String> getIdentityPartitionColumns(Table table, Schema
scanSchema) {
LinkedHashSet<String> partitionColumns = new LinkedHashSet<>();
for (PartitionSpec spec : table.specs().values()) {
for (PartitionField partitionField : spec.fields()) {
if (!partitionField.transform().isIdentity()) {
continue;
}
- String columnName =
table.schema().findColumnName(partitionField.sourceId());
+ String columnName =
+ topLevelSourceColumnName(scanSchema,
partitionField.sourceId());
if (columnName != null) {
partitionColumns.add(columnName);
}
@@ -189,12 +191,8 @@ final class IcebergPartitionUtils {
* mistaken for the top-level column {@code id}: the engine can only
materialize a top-level column.
*/
static String topLevelSourceColumnName(Schema schema, int sourceId) {
- for (NestedField column : schema.columns()) {
- if (column.fieldId() == sourceId) {
- return column.name();
- }
- }
- return null;
+ NestedField column = schema.asStruct().field(sourceId);
+ return column == null ? null : column.name();
}
/** The spec's <em>identity</em> partition over the top-level column
{@code name}, or null. */
@@ -253,11 +251,12 @@ final class IcebergPartitionUtils {
/**
* 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
- * (LinkedHashMap, spec field order). Mirrors legacy {@code
IcebergUtils.getIdentityPartitionInfoMap}.
+ * transforms and BINARY/FIXED columns (utf8 round-trip would corrupt
those). Keys are resolved by field id
+ * against {@code scanSchema}, matching {@code path_partition_keys}.
Order-preserving (LinkedHashMap, spec
+ * field order). Mirrors legacy {@code
IcebergUtils.getIdentityPartitionInfoMap}.
*/
static Map<String, String> getIdentityPartitionInfoMap(PartitionData
partitionData,
- PartitionSpec partitionSpec, Table table, ZoneId zone) {
+ PartitionSpec partitionSpec, Schema scanSchema, ZoneId zone) {
Map<String, String> partitionInfoMap = new LinkedHashMap<>();
List<NestedField> fields =
partitionData.getPartitionType().asNestedType().fields();
List<PartitionField> partitionFields = partitionSpec.fields();
@@ -275,7 +274,8 @@ final class IcebergPartitionUtils {
continue;
}
- String columnName =
table.schema().findColumnName(partitionField.sourceId());
+ String columnName =
+ topLevelSourceColumnName(scanSchema,
partitionField.sourceId());
if (columnName == null) {
continue;
}
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 2f07b5ac0a7..d3245f604d1 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
@@ -583,9 +583,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return emptySplitSource();
}
Table table = resolveTable(session, iceHandle);
+ Schema scanSchema = pinnedSchema(table, iceHandle);
TableScan scan = buildScan(table, iceHandle, filter, session);
int formatVersion = getFormatVersion(table);
- List<String> orderedPartitionKeys =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
+ List<String> orderedPartitionKeys =
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
scanSchema);
ZoneId zone = resolveSessionZone(session);
boolean partitioned = IcebergPartitionUtils.hasPartitionedSpec(table);
Map<String, String> vendedToken = context != null
@@ -597,7 +599,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
try {
CloseableIterable<FileScanTask> tasks = streamingFileScanTasks(
scan, session, table, filter, sliceSize);
- return new IcebergStreamingSplitSource(tasks, table,
formatVersion, partitioned,
+ return new IcebergStreamingSplitSource(tasks, table, scanSchema,
formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer, sliceSize,
iceHandle.getRewriteFileScope(), iceHandle,
requiresRowId(columns));
} catch (RuntimeException e) {
@@ -656,6 +658,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private final class IcebergStreamingSplitSource implements
ConnectorSplitSource {
private final CloseableIterable<FileScanTask> tasks;
private final Table table;
+ private final Schema scanSchema;
private final int formatVersion;
private final boolean partitioned;
private final List<String> orderedPartitionKeys;
@@ -677,12 +680,14 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// 1-entry cache stays O(1) memory (never accumulates), preserving the
streaming path's OOM safety.
private final PerFileScratch scratch = new PerFileScratch();
- IcebergStreamingSplitSource(CloseableIterable<FileScanTask> tasks,
Table table, int formatVersion,
+ IcebergStreamingSplitSource(CloseableIterable<FileScanTask> tasks,
Table table, Schema scanSchema,
+ int formatVersion,
boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
UnaryOperator<String> uriNormalizer, long sliceSize,
Set<String> rewriteScope,
IcebergTableHandle handle, boolean requiresRowId) {
this.tasks = tasks;
this.table = table;
+ this.scanSchema = scanSchema;
this.formatVersion = formatVersion;
this.partitioned = partitioned;
this.orderedPartitionKeys = orderedPartitionKeys;
@@ -704,8 +709,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
iterator = tasks.iterator();
}
while (iterator.hasNext()) {
- IcebergScanRange range =
buildRangeForTask(iterator.next(), table, formatVersion, partitioned,
- orderedPartitionKeys, zone, uriNormalizer,
sliceSize, rewriteScope, null, scratch);
+ IcebergScanRange range = buildRangeForTask(
+ iterator.next(), table, scanSchema, formatVersion,
partitioned,
+ orderedPartitionKeys, zone, uriNormalizer,
sliceSize, rewriteScope, null,
+ scratch);
if (range != null) {
if (requiresRowId) {
rejectRowIdRangeWithoutPartitionData(range);
@@ -764,6 +771,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return planSystemTableScan(iceHandle, columns, filter, session);
}
Table table = resolveTable(session, iceHandle);
+ Schema scanSchema = pinnedSchema(table, iceHandle);
TableScan scan = buildScan(table, iceHandle, filter, session);
// FIX-SCAN-METRICS: attach a per-scan metrics reporter so the iceberg
SDK's ScanReport (planning time,
// data/delete files, scanned vs skipped manifests) is captured into
the query profile — restores the
@@ -777,7 +785,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
int formatVersion = getFormatVersion(table);
- List<String> orderedPartitionKeys =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
+ List<String> orderedPartitionKeys =
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
scanSchema);
ZoneId zone = resolveSessionZone(session);
boolean partitioned = IcebergPartitionUtils.hasPartitionedSpec(table);
@@ -800,8 +809,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// A data-row predicate can leave partially matching files, whose
file-level recordCount is only an
// upper bound. Keep those scans on the normal path even if the engine
supplies the count signal.
if (countPushdown && filter.isEmpty()) {
- Optional<List<ConnectorScanRange>> countRanges =
planCountPushdown(table, scan, formatVersion,
- partitioned, orderedPartitionKeys, zone, uriNormalizer,
session, filter);
+ Optional<List<ConnectorScanRange>> countRanges = planCountPushdown(
+ table, scan, scanSchema, formatVersion, partitioned,
orderedPartitionKeys, zone,
+ uriNormalizer, session, filter);
if (countRanges.isPresent()) {
return countRanges.get();
}
@@ -840,9 +850,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
for (FileScanTask task : plan.tasks) {
// Shared per-task mapping (rewrite-scope skip + M-2 weight
denominator + v3 stash side-effect),
// identical to the streaming path's
IcebergStreamingSplitSource so both produce the same ranges.
- IcebergScanRange range = buildRangeForTask(task, table,
formatVersion, partitioned,
- orderedPartitionKeys, zone, uriNormalizer,
plan.targetSplitSize, rewriteScope,
- rewritableDeleteSupply, scratch);
+ IcebergScanRange range = buildRangeForTask(
+ task, table, scanSchema, formatVersion, partitioned,
orderedPartitionKeys, zone,
+ uriNormalizer, plan.targetSplitSize, rewriteScope,
rewritableDeleteSupply,
+ scratch);
if (range != null) {
ranges.add(range);
}
@@ -898,8 +909,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* side-effect. The streaming path passes {@code
rewritableDeleteSupply=null} (v3 is gated onto the eager
* path — see {@link #streamingSplitEstimate}), so the accumulation is
inert there.
*/
- private IcebergScanRange buildRangeForTask(FileScanTask task, Table table,
int formatVersion,
- boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
+ private IcebergScanRange buildRangeForTask(FileScanTask task, Table table,
Schema scanSchema,
+ int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, long targetSplitSize,
Set<String> rewriteScope,
Map<String, List<TIcebergDeleteFileDesc>> rewritableDeleteSupply,
PerFileScratch scratch) {
DataFile dataFile = task.file();
@@ -913,8 +924,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
boolean firstSliceOfFile = scratch.file != dataFile;
// targetSplitSize is the scan-level weight denominator (M-2): each
data-file range carries a
// size-proportional BE scheduling weight (selfSplitWeight computed
inside buildRange).
- IcebergScanRange range = buildRange(table, dataFile, task,
formatVersion, partitioned,
- orderedPartitionKeys, zone, uriNormalizer, -1,
targetSplitSize, scratch);
+ IcebergScanRange range = buildRange(
+ table, scanSchema, dataFile, task, formatVersion, partitioned,
orderedPartitionKeys,
+ zone, uriNormalizer, -1, targetSplitSize, scratch);
if (rewritableDeleteSupply != null && firstSliceOfFile) {
// Record this data file's non-equality delete supply keyed on its
RAW path (the exact string the BE
// matches a rewritable set against). An empty list (no old non-eq
deletes) contributes nothing.
@@ -982,7 +994,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// the ids through a Set and re-emit them in TABLE-schema order
(TypeUtil.project), breaking that
// contract; scan.project(orderedSchema) preserves the requested
order, so we build the schema by
// hand from the requested columns.
- List<String> projectedColumns = requestedLowerNames(columns);
+ List<String> projectedColumns = requestedColumnNames(columns);
if (!projectedColumns.isEmpty()) {
Schema metadataSchema = metadataTable.schema();
List<NestedField> projectedFields = new
ArrayList<>(projectedColumns.size());
@@ -1231,7 +1243,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
/** Whether the query projects the {@code partition} column (legacy read
the tuple's slots). */
private static boolean
isPositionDeletesPartitionColumnRequested(List<ConnectorColumnHandle> columns) {
- return
requestedLowerNames(columns).stream().anyMatch("partition"::equalsIgnoreCase);
+ return
requestedColumnNames(columns).stream().anyMatch("partition"::equalsIgnoreCase);
}
/**
@@ -1246,8 +1258,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
*/
private TableScan buildScan(Table table, IcebergTableHandle handle,
Optional<ConnectorExpression> filter,
ConnectorSession session) {
- Schema selectedSchema = !handle.isSystemTable() &&
handle.hasSnapshotPin()
- ? pinnedSchema(table, handle) : table.schema();
+ Schema selectedSchema = handle.isSystemTable() ? table.schema() :
pinnedSchema(table, handle);
// Keep the SDK's native Table.newScan implementation unless an actual
historical schema needs the
// metadata-only snapshot fix; catalog-specific Table wrappers may
provide their own scan behavior.
TableScan scan = !handle.isSystemTable() &&
!selectedSchema.sameSchema(table.schema())
@@ -1348,7 +1359,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* perform a normal scan.
*/
private Optional<List<ConnectorScanRange>> planCountPushdown(Table table,
TableScan scan,
- int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
+ Schema scanSchema, int formatVersion, boolean partitioned,
List<String> orderedPartitionKeys,
+ ZoneId zone,
UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter) {
Snapshot snapshot = scan.snapshot();
if (snapshot == null) {
@@ -1377,15 +1389,16 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (!visibleRows.isPresent()) {
return Optional.empty();
}
- return planManifestCountRange(table, scan,
visibleRows.getAsLong(), formatVersion,
- partitioned, orderedPartitionKeys, zone, uriNormalizer,
session, filter,
- netPositionDeletes);
+ return planManifestCountRange(
+ table, scan, scanSchema, visibleRows.getAsLong(),
formatVersion, partitioned,
+ orderedPartitionKeys, zone, uriNormalizer, session,
filter, netPositionDeletes);
}
// Older manifest lists may omit aggregate counters. Preserve
correctness by falling back to the
// bounded per-file enumeration instead of trusting snapshot summary
metadata.
- return planCountPushdownFromFileTasks(table, scan, formatVersion,
partitioned,
- orderedPartitionKeys, zone, uriNormalizer, session, filter,
netPositionDeletes,
+ return planCountPushdownFromFileTasks(
+ table, scan, scanSchema, formatVersion, partitioned,
orderedPartitionKeys, zone,
+ uriNormalizer, session, filter, netPositionDeletes,
netPositionDeletes ? positionDeleteRows.getAsLong() : 0);
}
@@ -1421,33 +1434,37 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
netPositionDeletes ? positionDeleteRows.getAsLong() :
0).isPresent();
}
- private Optional<List<ConnectorScanRange>> planManifestCountRange(Table
table, TableScan scan, long exactCount,
- int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
- UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter,
- boolean netPositionDeletes) {
+ private Optional<List<ConnectorScanRange>> planManifestCountRange(
+ Table table, TableScan scan, Schema scanSchema, long exactCount,
int formatVersion,
+ boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
+ UnaryOperator<String> uriNormalizer, ConnectorSession session,
+ Optional<ConnectorExpression> filter, boolean netPositionDeletes) {
if (!isManifestCacheEnabled()) {
- return buildManifestCountRange(table, scan.planFiles(),
exactCount, formatVersion, partitioned,
+ return buildManifestCountRange(
+ table, scanSchema, scan.planFiles(), exactCount,
formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes);
}
String statsQueryId = session != null ? session.getQueryId() : null;
try {
- return buildManifestCountRange(table,
- cacheBackedFileScanTasks(scan, session, table, filter,
statsQueryId), exactCount,
- formatVersion, partitioned, orderedPartitionKeys, zone,
uriNormalizer, netPositionDeletes);
+ return buildManifestCountRange(
+ table, scanSchema, cacheBackedFileScanTasks(scan, session,
table, filter, statsQueryId),
+ exactCount, formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer,
+ netPositionDeletes);
} catch (Exception e) {
LOG.warn("Iceberg count-pushdown representative plan with manifest
cache failed, "
+ "falling back to SDK scan: {}", e.getMessage(), e);
manifestCache.recordFailure(statsQueryId);
// The SDK retry must own a new iterable because a lazy cache
failure may leave the first one partial.
- return buildManifestCountRange(table, scan.planFiles(),
exactCount, formatVersion, partitioned,
+ return buildManifestCountRange(
+ table, scanSchema, scan.planFiles(), exactCount,
formatVersion, partitioned,
orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes);
}
}
- private Optional<List<ConnectorScanRange>> buildManifestCountRange(Table
table,
- CloseableIterable<FileScanTask> tasks, long exactCount, int
formatVersion, boolean partitioned,
- List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer,
- boolean netPositionDeletes) {
+ private Optional<List<ConnectorScanRange>> buildManifestCountRange(
+ Table table, Schema scanSchema, CloseableIterable<FileScanTask>
tasks, long exactCount,
+ int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
+ UnaryOperator<String> uriNormalizer, boolean netPositionDeletes) {
try (CloseableIterable<FileScanTask> closeableTasks = tasks) {
for (FileScanTask task : closeableTasks) {
// Data-manifest rows and the separately netted live delete
files are authoritative, but retain
@@ -1455,8 +1472,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (hasNonIgnorableTaskDeletes(task, netPositionDeletes) ||
task.file().recordCount() < 0) {
return Optional.empty();
}
- return Optional.of(Collections.singletonList(buildRange(table,
task.file(), task, formatVersion,
- partitioned, orderedPartitionKeys, zone,
uriNormalizer, exactCount, -1, null)));
+ return Optional.of(Collections.singletonList(buildRange(
+ table, scanSchema, task.file(), task, formatVersion,
partitioned,
+ orderedPartitionKeys, zone, uriNormalizer, exactCount,
-1, null)));
}
} catch (IOException e) {
throw new RuntimeException("Failed to plan iceberg count-pushdown
file, error message is:"
@@ -1465,17 +1483,20 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return exactCount == 0 ? Optional.of(Collections.emptyList()) :
Optional.empty();
}
- private Optional<List<ConnectorScanRange>>
planCountPushdownFromFileTasks(Table table, TableScan scan,
- int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
- UnaryOperator<String> uriNormalizer, ConnectorSession session,
Optional<ConnectorExpression> filter,
- boolean netPositionDeletes, long positionDeleteRows) {
+ private Optional<List<ConnectorScanRange>> planCountPushdownFromFileTasks(
+ Table table, TableScan scan, Schema scanSchema, int formatVersion,
boolean partitioned,
+ List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer,
+ ConnectorSession session, Optional<ConnectorExpression> filter,
boolean netPositionDeletes,
+ long positionDeleteRows) {
if (!isManifestCacheEnabled()) {
- return accumulateCountPushdownFileTasks(table, scan.planFiles(),
formatVersion, partitioned,
+ return accumulateCountPushdownFileTasks(
+ table, scanSchema, scan.planFiles(), formatVersion,
partitioned,
orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes, positionDeleteRows);
}
String statsQueryId = session != null ? session.getQueryId() : null;
try {
- return accumulateCountPushdownFileTasks(table,
+ return accumulateCountPushdownFileTasks(
+ table, scanSchema,
cacheBackedFileScanTasks(scan, session, table, filter,
statsQueryId), formatVersion,
partitioned, orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes,
positionDeleteRows);
@@ -1485,15 +1506,16 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
manifestCache.recordFailure(statsQueryId);
// The retry owns a fresh accumulator so rows consumed before a
lazy cache failure are never counted
// twice and the SDK fallback remains an exact restart.
- return accumulateCountPushdownFileTasks(table, scan.planFiles(),
formatVersion, partitioned,
+ return accumulateCountPushdownFileTasks(
+ table, scanSchema, scan.planFiles(), formatVersion,
partitioned,
orderedPartitionKeys, zone, uriNormalizer,
netPositionDeletes, positionDeleteRows);
}
}
- private Optional<List<ConnectorScanRange>>
accumulateCountPushdownFileTasks(Table table,
- CloseableIterable<FileScanTask> tasks, int formatVersion, boolean
partitioned,
- List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer,
- boolean netPositionDeletes, long positionDeleteRows) {
+ private Optional<List<ConnectorScanRange>>
accumulateCountPushdownFileTasks(
+ Table table, Schema scanSchema, CloseableIterable<FileScanTask>
tasks, int formatVersion,
+ boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
+ UnaryOperator<String> uriNormalizer, boolean netPositionDeletes,
long positionDeleteRows) {
FileScanTask representative = null;
long exactCount = 0;
try (CloseableIterable<FileScanTask> closeableTasks = tasks) {
@@ -1525,9 +1547,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
// targetSplitSize = -1: the count-pushdown collapse emits a single
range, so its scheduling weight is
// irrelevant and PluginDrivenSplit keeps SplitWeight.standard().
- return Optional.of(Collections.singletonList(buildRange(table,
representative.file(), representative,
- formatVersion, partitioned, orderedPartitionKeys, zone,
uriNormalizer,
- visibleRows.getAsLong(), -1, null)));
+ return Optional.of(Collections.singletonList(buildRange(
+ table, scanSchema, representative.file(), representative,
formatVersion, partitioned,
+ orderedPartitionKeys, zone, uriNormalizer,
visibleRows.getAsLong(), -1, null)));
}
private static boolean hasNonIgnorableTaskDeletes(FileScanTask task,
boolean netPositionDeletes) {
@@ -1652,13 +1674,15 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* size-proportional {@code selfSplitWeight} are per-slice. A {@code null}
scratch (the single
* count-pushdown range) computes without caching. Byte-identical to the
former per-slice recompute.</p>
*/
- private IcebergScanRange buildRange(Table table, DataFile dataFile,
FileScanTask task, int formatVersion,
- boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
+ private IcebergScanRange buildRange(Table table, Schema scanSchema,
DataFile dataFile,
+ FileScanTask task, int formatVersion, boolean partitioned,
+ List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, long pushDownRowCount, long
targetSplitSize,
PerFileScratch scratch) {
PerFileScratch file = (scratch != null && scratch.file == dataFile)
? scratch
- : computePerFileInvariants(table, dataFile, task,
formatVersion, partitioned,
+ : computePerFileInvariants(
+ table, scanSchema, dataFile, task, formatVersion,
partitioned,
orderedPartitionKeys, zone, uriNormalizer, scratch);
// M-2 size-proportional weight numerator = this split's byte length +
the byte size of every
// merge-on-read delete file applying to it, mirroring legacy
IcebergSplit.selfSplitWeight (ctor sets
@@ -1702,8 +1726,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* so the memoized range is byte-identical to the per-slice recompute. The
scratch fields are assigned only
* after the fail-loud check, so a rejected file never leaves a
half-populated scratch.
*/
- private PerFileScratch computePerFileInvariants(Table table, DataFile
dataFile, FileScanTask task,
- int formatVersion, boolean partitioned, List<String>
orderedPartitionKeys, ZoneId zone,
+ private PerFileScratch computePerFileInvariants(Table table, Schema
scanSchema, DataFile dataFile,
+ FileScanTask task, int formatVersion, boolean partitioned,
+ List<String> orderedPartitionKeys, ZoneId zone,
UnaryOperator<String> uriNormalizer, PerFileScratch scratch) {
perFileInvariantComputeCount++;
Integer partitionSpecId = null;
@@ -1726,7 +1751,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// 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 =
-
IcebergPartitionUtils.getIdentityPartitionInfoMap(partitionData, spec, table,
zone);
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ partitionData, spec, scanSchema, zone);
Map<String, String> ordered = new LinkedHashMap<>();
for (String key : orderedPartitionKeys) {
if (identityMap.containsKey(key)) {
@@ -1941,7 +1967,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* <ul>
* <li>{@code file_format_type=jni} — makes the parent default the
per-range format to {@code FORMAT_JNI},
* which each native range overrides to parquet/orc in {@code
populateRangeParams} (mirrors paimon).</li>
- * <li>{@code path_partition_keys} — the lowercased, comma-joined
identity partition columns, so FE marks
+ * <li>{@code path_partition_keys} — the case-preserved, comma-joined
identity partition columns, so FE marks
* those slots as partition columns and excludes them from the
file-decode set; without it BE
* double-fills the partition columns (decode-from-file AND
append-from-path) and DCHECKs (CI #968880).
* Emitted only when the table is partitioned (an empty value would
split into a single "" key).</li>
@@ -2018,7 +2044,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// (empty partitionKeys + null schema-dict table). resolveTable still
loads the base table here for
// the credential overlay below (the metadata table shares the base
table's FileIO).
if (!systemTable) {
- List<String> partitionKeys =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
+ List<String> partitionKeys =
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
scanSchema);
if (!partitionKeys.isEmpty()) {
props.put(ScanNodePropertyKeys.PATH_PARTITION_KEYS,
String.join(",", partitionKeys));
}
@@ -2059,7 +2086,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// pruned-by-requested-columns dict (CI #969249).
//
// Row-lineage (format-version >= 3): _row_id /
_last_updated_sequence_number are GENERATED BE scan slots
- // (they reach BE column_names) but are NOT in schema.columns(), so
requestedLowerNames — keyed off the
+ // (they reach BE column_names) but are NOT in schema.columns(), so
requestedColumnNames — keyed off the
// iceberg column handles — never carries them.
encodeSchemaEvolutionProp(appendRowLineage=true) appends
// them to the dict root so BE's StructNode children map contains
them; else the ParquetReader's
// unconditional children.at("_row_id") std::out_of_range-SIGABRTs the
whole BE.
@@ -2100,7 +2127,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
enableVarbinary, enableTimestampTz);
} else {
dict = IcebergSchemaUtils.encodeSchemaEvolutionProp(
- table, table.schema(), requestedLowerNames(columns),
+ table, table.schema(), requestedColumnNames(columns),
appendRowLineage, enableVarbinary, enableTimestampTz);
}
props.put(SCHEMA_EVOLUTION_PROP, dict);
@@ -2138,7 +2165,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
Table metadataTable =
MetadataTableUtils.createMetadataTableInstance(
table, MetadataTableType.POSITION_DELETES);
props.put(SCHEMA_EVOLUTION_PROP,
IcebergSchemaUtils.encodeSchemaEvolutionProp(
- metadataTable, metadataTable.schema(),
requestedLowerNames(columns), false,
+ metadataTable, metadataTable.schema(),
requestedColumnNames(columns), false,
enableVarbinary, enableTimestampTz));
}
// Pushed-predicate EXPLAIN prop (explain gap): serialize the iceberg
Expression form of each pushed
@@ -2173,14 +2200,14 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
/**
- * The lowercased names of the requested (pruned) columns — the
authoritative Doris scan slots the field-id
+ * The case-preserved 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), minus the synthesized
+ * {@code IcebergConnectorMetadata.getColumnHandles} produced, 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) {
+ private static List<String>
requestedColumnNames(List<ConnectorColumnHandle> columns) {
if (columns == null || columns.isEmpty()) {
return Collections.emptyList();
}
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 ef48c73ed2d..286a31fcfb5 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
@@ -69,7 +69,7 @@ public class IcebergScanRange implements ConnectorScanRange {
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
+ // Identity partition column (case-preserved) -> 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
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
index 0ea68a006c7..1be990d5554 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
@@ -109,19 +109,19 @@ public final class IcebergSchemaUtils {
}
/**
- * Orchestrator: build the schema dictionary for {@code table} keyed off
the requested (lowercased) column
- * names and serialize it for transport via the scan-node props. {@code
requestedLowerNames} is the pruned
+ * Orchestrator: build the schema dictionary for {@code table} keyed off
the requested column names and
+ * serialize it for transport via the scan-node props. {@code
requestedColumnNames} is the pruned
* scan-slot list ({@code PluginDrivenScanNode} hands the provider the
requested columns); an empty list
* (count-only scan / no column handles) falls back to all top-level
schema columns.
*/
- static String encodeSchemaEvolutionProp(Table table, List<String>
requestedLowerNames) {
- return encodeSchemaEvolutionProp(table, table.schema(),
requestedLowerNames, false, false, false);
+ static String encodeSchemaEvolutionProp(Table table, List<String>
requestedColumnNames) {
+ return encodeSchemaEvolutionProp(table, table.schema(),
requestedColumnNames, false, false, false);
}
/**
* Like {@link #encodeSchemaEvolutionProp(Table, List)} but builds the
dictionary from an explicit
* {@code dictSchema} (the latest schema for a normal read, or a
historical schema for a time-travel read —
- * T07 Option A passes the PINNED schema with an empty {@code
requestedLowerNames} so the dict covers every
+ * T07 Option A passes the PINNED schema with an empty {@code
requestedColumnNames} so the dict covers every
* BE scan slot). The name mapping is still read from {@code table} (it is
table-level, not schema-versioned).
*
* <p>{@code appendRowLineage} (set by the caller when the table
format-version >= 3) appends the iceberg
@@ -131,12 +131,12 @@ public final class IcebergSchemaUtils {
* {@code children.at("_row_id")} {@code std::out_of_range}-SIGABRTs the
whole BE. See
* {@link #appendRowLineageFields}.</p>
*/
- static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedLowerNames,
+ static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedColumnNames,
boolean appendRowLineage) {
// Thin overload: keep both optional type mappings disabled for
callers that do not thread catalog
// properties.
return encodeSchemaEvolutionProp(
- table, dictSchema, requestedLowerNames, appendRowLineage,
false, false);
+ table, dictSchema, requestedColumnNames, appendRowLineage,
false, false);
}
/**
@@ -145,19 +145,19 @@ public final class IcebergSchemaUtils {
* TIMESTAMPTZ column's iceberg initial default is serialized consistently
with how BE will read the
* column (keep the trailing offset when tz-mapping is on, drop it —
DATETIMEV2 UTC wall time — when off).
*/
- static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedLowerNames,
+ static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedColumnNames,
boolean appendRowLineage, boolean enableTimestampTz) {
return encodeSchemaEvolutionProp(
- table, dictSchema, requestedLowerNames, appendRowLineage,
false, enableTimestampTz);
+ table, dictSchema, requestedColumnNames, appendRowLineage,
false, enableTimestampTz);
}
/** Build and encode a schema dictionary with the catalog's Doris
type-mapping options. */
- static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedLowerNames,
+ static String encodeSchemaEvolutionProp(Table table, Schema dictSchema,
List<String> requestedColumnNames,
boolean appendRowLineage, boolean enableVarbinary, boolean
enableTimestampTz) {
Optional<Map<Integer, List<String>>> nameMapping =
extractNameMapping(table);
// #65784: it is the PRESENCE of the table-level mapping (not its
non-emptiness) that makes it
// authoritative for BE, so thread isPresent() through as
hasNameMapping.
- TSchema current = buildCurrentSchema(dictSchema, requestedLowerNames,
+ TSchema current = buildCurrentSchema(dictSchema, requestedColumnNames,
nameMapping.orElse(Collections.emptyMap()),
nameMapping.isPresent(), enableVarbinary,
enableTimestampTz);
if (appendRowLineage) {
@@ -267,16 +267,16 @@ public final class IcebergSchemaUtils {
* fix: the top-level names == the BE scan slots so the {@code StructNode}
DCHECK can never miss). Each
* requested name is matched case-insensitively to the iceberg schema; its
top-level {@code TField} name is
* the requested name VERBATIM (byte-matching the Doris slot name;
case-preserved post-#65094). An
- * empty/{@code null} {@code requestedLowerNames} falls back to all
top-level columns (lowercased). Fail
- * loud if a requested column is absent from the schema (a genuine
FE/connector inconsistency — not a
- * silent drop).
+ * empty/{@code null} {@code requestedColumnNames} falls back to all
top-level columns using the
+ * authoritative Iceberg field names. Fail loud if a requested column is
absent from the schema (a genuine
+ * FE/connector inconsistency — not a silent drop).
*/
- static TSchema buildCurrentSchema(Schema schema, List<String>
requestedLowerNames,
+ static TSchema buildCurrentSchema(Schema schema, List<String>
requestedColumnNames,
Map<Integer, List<String>> nameMapping) {
// Thin overload: default enableTimestampTz=false (pre-#65502
timestamp-default behaviour) and derive
// hasNameMapping from the map (a non-empty map ⇒ the table carried a
mapping — the #65784 default;
// the production path threads the precise isPresent() instead).
- return buildCurrentSchema(schema, requestedLowerNames, nameMapping,
+ return buildCurrentSchema(schema, requestedColumnNames, nameMapping,
nameMapping != null && !nameMapping.isEmpty(), false, false);
}
@@ -287,26 +287,26 @@ public final class IcebergSchemaUtils {
* AUTHORITATIVE, so every field carries an explicit (possibly empty)
per-field mapping (see
* {@link #buildField}).
*/
- static TSchema buildCurrentSchema(Schema schema, List<String>
requestedLowerNames,
+ static TSchema buildCurrentSchema(Schema schema, List<String>
requestedColumnNames,
Map<Integer, List<String>> nameMapping, boolean hasNameMapping,
boolean enableTimestampTz) {
- return buildCurrentSchema(schema, requestedLowerNames, nameMapping,
hasNameMapping, false,
+ return buildCurrentSchema(schema, requestedColumnNames, nameMapping,
hasNameMapping, false,
enableTimestampTz);
}
/** Build the dictionary using the same Doris scalar mappings as the
connector's catalog columns. */
- static TSchema buildCurrentSchema(Schema schema, List<String>
requestedLowerNames,
+ static TSchema buildCurrentSchema(Schema schema, List<String>
requestedColumnNames,
Map<Integer, List<String>> nameMapping, boolean hasNameMapping,
boolean enableVarbinary,
boolean enableTimestampTz) {
TSchema tSchema = new TSchema();
tSchema.setSchemaId(CURRENT_SCHEMA_ID);
TStructField root = new TStructField();
- if (requestedLowerNames == null || requestedLowerNames.isEmpty()) {
+ if (requestedColumnNames == null || requestedColumnNames.isEmpty()) {
for (Types.NestedField field : schema.columns()) {
- addField(root, buildField(field,
field.name().toLowerCase(Locale.ROOT), nameMapping,
+ addField(root, buildField(field, field.name(), nameMapping,
hasNameMapping, enableVarbinary, enableTimestampTz));
}
} else {
- for (String name : requestedLowerNames) {
+ for (String name : requestedColumnNames) {
Types.NestedField field =
schema.caseInsensitiveFindField(name);
if (field == null) {
throw new RuntimeException("iceberg schema-evolution:
requested column '" + name
@@ -327,7 +327,7 @@ public final class IcebergSchemaUtils {
tSchema.setSchemaId(CURRENT_SCHEMA_ID);
TStructField root = new TStructField();
for (Types.NestedField field : fields) {
- addField(root, buildField(field,
field.name().toLowerCase(Locale.ROOT), nameMapping,
+ addField(root, buildField(field, field.name(), nameMapping,
hasNameMapping, enableVarbinary, enableTimestampTz));
}
tSchema.setRootField(root);
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 a4cba8bd550..a33efc5ea2d 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
@@ -164,12 +164,13 @@ public class IcebergPartitionUtilsTest {
UUID value = UUID.fromString("00112233-4455-6677-8899-aabbccddeeff");
data.set(0, value);
Assertions.assertEquals("00112233-4455-6677-8899-aabbccddeeff",
- IcebergPartitionUtils.getIdentityPartitionInfoMap(data, spec,
table, ZoneOffset.UTC).get("key"));
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ data, spec, table.schema(),
ZoneOffset.UTC).get("key"));
Assertions.assertEquals(value.toString(),
IcebergPartitionUtils.serializePartitionValue(Types.UUIDType.get(), value,
ZoneOffset.UTC));
data.set(0, null);
- Assertions.assertNull(
- IcebergPartitionUtils.getIdentityPartitionInfoMap(data, spec,
table, ZoneOffset.UTC).get("key"));
+
Assertions.assertNull(IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ data, spec, table.schema(), ZoneOffset.UTC).get("key"));
}
@Test
@@ -258,7 +259,8 @@ public class IcebergPartitionUtilsTest {
.build();
Table table = tableWith(schema, spec);
- List<String> cols =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
+ List<String> cols =
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
table.schema());
// Only the two identity columns, CASE-PRESERVED (#65094 read-path
alignment), in spec order; the
// bucket(id) transform is excluded. MUTATION: including non-identity
transforms -> "id_bucket"/"id"
@@ -284,13 +286,31 @@ public class IcebergPartitionUtilsTest {
table.updateSpec().removeField("payload").addField("int_col").commit();
Assertions.assertEquals(2, table.specs().size());
- List<String> cols =
IcebergPartitionUtils.getIdentityPartitionColumns(table);
+ List<String> cols =
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
table.schema());
// Both specs contribute, in first-seen (spec id) order. MUTATION:
iterating table.spec() instead of
// table.specs() -> ["int_col"] -> red. MUTATION: dropping the
LinkedHashSet dedup -> duplicates -> red.
Assertions.assertEquals(Arrays.asList("payload", "int_col"), cols);
}
+ @Test
+ public void
identityPartitionColumnsUseAuthoritativeScanSchemaAfterCaseRename() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "P", Types.IntegerType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("P").build();
+ Table table = tableWith(schema, spec);
+ Schema pinnedSchema = table.schema();
+
+ table.updateSchema().renameColumn("P", "p").commit();
+
+ Assertions.assertEquals(Collections.singletonList("P"),
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
pinnedSchema));
+ Assertions.assertEquals(Collections.singletonList("p"),
+ IcebergPartitionUtils.getIdentityPartitionColumns(table,
table.schema()));
+ }
+
// ---- toPartitionFieldKeys ----
@Test
@@ -378,7 +398,8 @@ public class IcebergPartitionUtilsTest {
pd.set(1, 2); // id_bucket = 2 (non-identity -> skipped)
Map<String, String> info =
- IcebergPartitionUtils.getIdentityPartitionInfoMap(pd, spec,
table, ZoneOffset.UTC);
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ pd, spec, table.schema(), ZoneOffset.UTC);
// Only the identity column survives, key CASE-PRESERVED (#65094
read-path alignment). MUTATION:
// emitting id_bucket -> size 2 -> red. MUTATION: re-lowercasing the
key -> "p" != "P" -> red.
@@ -396,12 +417,34 @@ public class IcebergPartitionUtilsTest {
pd.set(0, null); // genuine null partition value
Map<String, String> info =
- IcebergPartitionUtils.getIdentityPartitionInfoMap(pd, spec,
table, ZoneOffset.UTC);
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ pd, spec, table.schema(), ZoneOffset.UTC);
Assertions.assertTrue(info.containsKey("p"));
Assertions.assertNull(info.get("p"));
}
+ @Test
+ public void
identityPartitionInfoMapUsesAuthoritativeScanSchemaAfterCaseRename() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "P", Types.IntegerType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("P").build();
+ Table table = tableWith(schema, spec);
+ Schema pinnedSchema = table.schema();
+
+ table.updateSchema().renameColumn("P", "p").commit();
+
+ PartitionData pd = new PartitionData(spec.partitionType());
+ pd.set(0, 7);
+ Assertions.assertEquals(Collections.singletonMap("P", "7"),
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ pd, spec, pinnedSchema, ZoneOffset.UTC));
+ Assertions.assertEquals(Collections.singletonMap("p", "7"),
+ IcebergPartitionUtils.getIdentityPartitionInfoMap(
+ pd, spec, table.schema(), ZoneOffset.UTC));
+ }
+
// ---- getPartitionDataJson ----
@Test
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 633595a680a..5cbea55cfd4 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
@@ -3020,6 +3020,77 @@ public class IcebergScanPlanProviderTest {
.getFields().get(1).getFieldPtr().getName());
}
+ @Test
+ public void
getScanNodePropertiesUnderTopnPreservesMixedCaseFullSchemaNames() throws
Exception {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "Name", Types.StringType.get()));
+ Table table = createTable("mixed_topn", schema,
PartitionSpec.unpartitioned());
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1",
"mixed_topn").withTopnLazyMaterialize(true),
+ Collections.singletonList(new IcebergColumnHandle("ID", 1)),
Optional.empty());
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+
+ List<TFieldPtr> fields =
params.getHistorySchemaInfo().get(0).getRootField().getFields();
+ Assertions.assertEquals("ID", fields.get(0).getFieldPtr().getName());
+ Assertions.assertEquals("Name", fields.get(1).getFieldPtr().getName());
+ }
+
+ @Test
+ public void pinnedMixedCasePartitionUsesOneNameAcrossAllCarriers() throws
Exception {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "City", Types.StringType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("City").build();
+ Table table = createTable("mixed_pin", schema, spec,
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend()
+ .appendFile(partitionedFile(spec,
"s3://b/db/mixed_pin/City=Beijing/a.parquet",
+ "Beijing"))
+ .commit();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ long schemaId = table.currentSnapshot().schemaId();
+ table.updateSchema().renameColumn("City", "city").commit();
+
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "mixed_pin")
+ .withSnapshot(snapshotId, null, schemaId);
+ List<ConnectorColumnHandle> columns =
+ Collections.singletonList(new IcebergColumnHandle("City", 2));
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, handle, columns, Optional.empty());
+ Assertions.assertEquals("City",
props.get(ScanNodePropertyKeys.PATH_PARTITION_KEYS));
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+ Assertions.assertEquals("City",
params.getHistorySchemaInfo().get(0).getRootField()
+ .getFields().get(1).getFieldPtr().getName());
+
+ List<ConnectorScanRange> ranges = provider.planScan(
+ new FakeScanSession("UTC", Collections.emptyMap()),
+ ConnectorScanRequest.builder(handle, columns).build());
+ Assertions.assertEquals(1, ranges.size());
+ TFileRangeDesc desc = populate(ranges.get(0));
+ Assertions.assertEquals(Collections.singletonList("City"),
desc.getColumnsFromPathKeys());
+ Assertions.assertEquals(Collections.singletonList("Beijing"),
desc.getColumnsFromPath());
+
+ ConnectorSplitSource splits = provider.streamSplits(
+ new FakeScanSession("UTC", Collections.emptyMap()),
+ ConnectorScanRequest.builder(handle, columns).build());
+ Assertions.assertTrue(splits.hasNext());
+ TFileRangeDesc streamingDesc = populate(splits.next());
+ splits.close();
+ Assertions.assertEquals(Collections.singletonList("City"),
+ streamingDesc.getColumnsFromPathKeys());
+ Assertions.assertEquals(Collections.singletonList("Beijing"),
+ streamingDesc.getColumnsFromPath());
+ }
+
@Test
public void getScanNodePropertiesPinTakesPrecedenceOverTopnLazyMat()
throws Exception {
// A time-travel pin + Top-N lazy mat must take the PIN branch (pinned
schema, full columns), not the
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
index 323ac026b8a..0e9278ab8bb 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
@@ -79,11 +79,11 @@ public class IcebergSchemaUtilsTest {
}
/** Build the dictionary for the given requested column names and return
the single (-1) entry. */
- private static TSchema dict(Table table, String... requestedLowerNames) {
+ private static TSchema dict(Table table, String... requestedColumnNames) {
// Mirror the production path (encodeSchemaEvolutionProp): thread the
PRECISE isPresent() as
// hasNameMapping (#65784), so a present-but-empty mapping is still
authoritative.
Optional<Map<Integer, List<String>>> nameMapping =
IcebergSchemaUtils.extractNameMapping(table);
- return IcebergSchemaUtils.buildCurrentSchema(table.schema(),
Arrays.asList(requestedLowerNames),
+ return IcebergSchemaUtils.buildCurrentSchema(table.schema(),
Arrays.asList(requestedColumnNames),
nameMapping.orElse(Collections.emptyMap()),
nameMapping.isPresent(), false);
}
@@ -182,6 +182,46 @@ public class IcebergSchemaUtilsTest {
Assertions.assertFalse(fields.containsKey("name"));
}
+ @Test
+ public void fullSchemaFallbackPreservesMixedCaseTopLevelNamesOnly() {
+ Schema schema = new Schema(
+ Types.NestedField.required(7, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(9, "Name", Types.StringType.get()),
+ Types.NestedField.optional(11, "Payload", Types.StructType.of(
+ Types.NestedField.optional(12, "Child",
Types.IntegerType.get()))));
+
+ Map<String, TField> fields =
topFields(IcebergSchemaUtils.buildCurrentSchema(
+ schema, Collections.emptyList(), Collections.emptyMap()));
+
+ // Snapshot-pin, Top-N lazy materialization and count-only scans have
no pruned requested list and use
+ // this fallback. Its top-level keys must equal the case-preserving
Doris scan slots.
+ Assertions.assertTrue(fields.containsKey("ID"));
+ Assertions.assertTrue(fields.containsKey("Name"));
+ Assertions.assertTrue(fields.containsKey("Payload"));
+ Assertions.assertFalse(fields.containsKey("id"));
+ Assertions.assertFalse(fields.containsKey("name"));
+ Assertions.assertFalse(fields.containsKey("payload"));
+
+ // Nested struct lookup remains lowercased by the Doris StructField
contract. The top-level fix must
+ // not leak into nested fields.
+ Assertions.assertEquals(12, childByName(fields.get("Payload"),
"child").getId());
+ }
+
+ @Test
+ public void equalityDeleteCarrierPreservesMixedCaseTopLevelNames() throws
Exception {
+ Schema schema = new Schema(
+ Types.NestedField.required(7, "ID", Types.IntegerType.get()),
+ Types.NestedField.optional(9, "Name", Types.StringType.get()));
+ Table table = createTable("mixed_equality", schema);
+
+ TFileScanRangeParams params =
decode(IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp(
+ table, schema.columns(), false, false, false));
+ List<TFieldPtr> fields =
params.getHistorySchemaInfo().get(0).getRootField().getFields();
+
+ Assertions.assertEquals("ID", fields.get(0).getFieldPtr().getName());
+ Assertions.assertEquals("Name", fields.get(1).getFieldPtr().getName());
+ }
+
@Test
public void keyedOffRequestedColumnsOnlyIncludesRequested() {
// CI #969249 invariant: the -1 entry's top-level names must equal the
BE scan slots BY CONSTRUCTION.
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.out
new file mode 100644
index 00000000000..67da718691b
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.out
@@ -0,0 +1,37 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !mixed_case_scan --
+1 one
+2 two
+3 three
+
+-- !mixed_case_topn --
+one
+two
+
+-- !mixed_case_nested --
+1 11 alpha
+2 22 beta
+3 33 gamma
+
+-- !pinned_partition_schema --
+1 Beijing
+2 Beijing
+
+-- !pinned_partition_filter --
+1 Beijing
+2 Beijing
+
+-- !pinned_partition_topn --
+1 Beijing
+
+-- !current_partition_schema --
+1 Beijing
+2 Beijing
+3 Shanghai
+
+-- !pinned_count --
+2
+
+-- !current_count --
+3
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.groovy
new file mode 100644
index 00000000000..710fa23e590
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_scan_schema_carrier_names.groovy
@@ -0,0 +1,166 @@
+// 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_scan_schema_carrier_names",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ return
+ }
+
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ sql """drop catalog if exists test_iceberg_scan_schema_carrier_names"""
+
+ spark_iceberg_multi """
+ SET spark.sql.shuffle.partitions=1;
+ CREATE DATABASE IF NOT EXISTS
demo.iceberg_scan_schema_carrier_names_db;
+ DROP TABLE IF EXISTS
demo.iceberg_scan_schema_carrier_names_db.mixed_case_orc;
+ CREATE TABLE demo.iceberg_scan_schema_carrier_names_db.mixed_case_orc (
+ ID INT,
+ Name STRING,
+ Payload STRUCT<Child: INT, Other: STRING>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.format.default'='orc'
+ );
+ INSERT INTO demo.iceberg_scan_schema_carrier_names_db.mixed_case_orc
+ SELECT /*+ COALESCE(1) */ id, name, payload
+ FROM VALUES
+ (1, 'one', named_struct('Child', 11, 'Other', 'alpha')),
+ (2, 'two', named_struct('Child', 22, 'Other', 'beta')),
+ (3, 'three', named_struct('Child', 33, 'Other', 'gamma'))
+ AS t(id, name, payload);
+
+ DROP TABLE IF EXISTS
demo.iceberg_scan_schema_carrier_names_db.mixed_case_partition_orc;
+ CREATE TABLE
demo.iceberg_scan_schema_carrier_names_db.mixed_case_partition_orc (
+ id INT,
+ City STRING
+ ) USING iceberg
+ PARTITIONED BY (City)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.format.default'='orc'
+ );
+ INSERT INTO
demo.iceberg_scan_schema_carrier_names_db.mixed_case_partition_orc
+ VALUES (1, 'Beijing'), (2, 'Beijing');
+ """
+
+ sql """
+ CREATE CATALOG test_iceberg_scan_schema_carrier_names 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',
+ 'meta.cache.iceberg.table.ttl-second'='0',
+ 'meta.cache.iceberg.schema.ttl-second'='0'
+ )
+ """
+
+ try {
+ sql """
+ ALTER TABLE
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ CREATE TAG mixed_case_initial
+ """
+
+ spark_iceberg_multi """
+ ALTER TABLE
demo.iceberg_scan_schema_carrier_names_db.mixed_case_partition_orc
+ RENAME COLUMN City TO city_name;
+ INSERT INTO
demo.iceberg_scan_schema_carrier_names_db.mixed_case_partition_orc
+ VALUES (3, 'Shanghai');
+ """
+ sql """refresh catalog test_iceberg_scan_schema_carrier_names"""
+
+ qt_mixed_case_scan """
+ SELECT ID, Name
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_orc`
+ ORDER BY ID
+ """
+
+ qt_mixed_case_topn """
+ SELECT Name
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_orc`
+ ORDER BY ID
+ LIMIT 2
+ """
+
+ qt_mixed_case_nested """
+ SELECT ID, Payload.Child, Payload.Other
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_orc`
+ ORDER BY ID
+ """
+
+ qt_pinned_partition_schema """
+ SELECT id, City
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ FOR VERSION AS OF 'mixed_case_initial'
+ ORDER BY id
+ """
+
+ qt_pinned_partition_filter """
+ SELECT id, City
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ FOR VERSION AS OF 'mixed_case_initial'
+ WHERE City = 'Beijing'
+ ORDER BY id
+ """
+
+ qt_pinned_partition_topn """
+ SELECT id, City
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ FOR VERSION AS OF 'mixed_case_initial'
+ ORDER BY id
+ LIMIT 1
+ """
+
+ qt_current_partition_schema """
+ SELECT id, city_name
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ ORDER BY id
+ """
+
+ qt_pinned_count """
+ SELECT COUNT(*)
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ FOR VERSION AS OF 'mixed_case_initial'
+ """
+
+ qt_current_count """
+ SELECT COUNT(*)
+ FROM
+
`test_iceberg_scan_schema_carrier_names`.`iceberg_scan_schema_carrier_names_db`.`mixed_case_partition_orc`
+ """
+ } finally {
+ sql """drop catalog if exists test_iceberg_scan_schema_carrier_names"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
