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 cb9bbe99892 [fix](iceberg) Fix historical scans after schema evolution
(#67687)
cb9bbe99892 is described below
commit cb9bbe998928ddca71989c2186c31ddb59a182ef
Author: Gabriel <[email protected]>
AuthorDate: Mon Sep 14 10:25:01 2026 +0800
[fix](iceberg) Fix historical scans after schema evolution (#67687)
### What this PR does
Ports #67479 from branch-4.1 to master.
- Upgrade Iceberg from 1.10.1 to 1.11.0 and sync the vendored
`DeleteFileIndex`.
- Preserve historical-schema field lookup for equality deletes after
columns are renamed or dropped.
- Adapt the fix to the connector-based master layout, including DLF and
HiveCatalog compatibility.
- Index evolved partition structs by their position in each spec,
because partition field IDs may contain gaps.
- Add connector unit coverage for historical predicates after
rename/drop and update Iceberg regression expectations.
Issue: [DORIS-28397](https://issues.apache.org/jira/browse/DORIS-28397)
### Tests
- Historical-schema and Iceberg 1.11 compatibility tests: 6 passed, 0
failed.
- FE Checkstyle validation: passed with 0 violations.
- Full `fe-connector-iceberg` reactor test run: 1390 passed, 5 skipped,
1 failed. The single failure
(`IcebergWritePlanProviderTest.planMergePreservesExplicitlyEmptyReadAcrossConcurrentFirstAppend`)
reproduces unchanged on the master baseline and is unrelated to this PR.
---
.../doris/iceberg/IcebergSerializationCompat.java | 84 ++++++
.../doris/iceberg/IcebergSysTableJniScanner.java | 3 +-
.../iceberg/IcebergSerializationCompatTest.java | 76 +++++
fe/check/checkstyle/suppressions.xml | 2 +
.../doris/connector/cache/ConnectorTableKey.java | 29 +-
.../cache/ConnectorMetadataCacheTest.java | 22 ++
.../fe-connector-hms-hive-shade/pom.xml | 7 +-
.../iceberg/IcebergConnectorMetadata.java | 46 ++-
.../connector/iceberg/IcebergPartitionCache.java | 26 +-
.../connector/iceberg/IcebergPartitionUtils.java | 45 ++-
.../connector/iceberg/IcebergScanPlanProvider.java | 54 +++-
.../iceberg/IcebergSystemTableSerialization.java | 90 ++++++
.../connector/iceberg/dlf/DLFTableOperations.java | 3 +-
.../java/org/apache/iceberg/DeleteFileIndex.java | 65 ++++-
.../apache/iceberg/SchemaAwareDataTableScan.java | 57 ++++
.../iceberg/IcebergConnectorCacheTest.java | 2 +-
...ergConnectorMetadataPartitionViewCacheTest.java | 62 +++-
.../iceberg/IcebergPartitionCacheTest.java | 7 +-
.../iceberg/IcebergPartitionUtilsTest.java | 108 +++++--
.../iceberg/IcebergScanPlanProviderTest.java | 316 ++++++++++++++++++++-
fe/pom.xml | 4 +-
.../iceberg_schema_change_ddl_with_branch.out | 1 -
.../iceberg/test_iceberg_sys_table.out | 2 +
.../iceberg_schema_change_ddl_with_branch.groovy | 2 +-
24 files changed, 972 insertions(+), 141 deletions(-)
diff --git
a/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSerializationCompat.java
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSerializationCompat.java
new file mode 100644
index 00000000000..dda1c43c1b2
--- /dev/null
+++
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSerializationCompat.java
@@ -0,0 +1,84 @@
+// 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.
+
+package org.apache.doris.iceberg;
+
+import org.apache.iceberg.Schema;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+final class IcebergSerializationCompat {
+ private static final long ICEBERG_1_10_1_SCHEMA_UID = 6812231194765760118L;
+ private static final long ICEBERG_1_11_0_SCHEMA_UID =
-1265875184407129845L;
+ private static final ObjectStreamClass LOCAL_SCHEMA_DESCRIPTOR =
ObjectStreamClass.lookup(Schema.class);
+
+ private IcebergSerializationCompat() {
+ }
+
+ @SuppressWarnings({"DangerousJavaDeserialization", "unchecked"})
+ static <T> T deserializeFromBase64(String base64) {
+ if (base64 == null) {
+ return null;
+ }
+ byte[] bytes =
Base64.getMimeDecoder().decode(base64.getBytes(StandardCharsets.UTF_8));
+ try (ByteArrayInputStream input = new ByteArrayInputStream(bytes);
+ ObjectInputStream objectInput = new
SchemaCompatibleObjectInputStream(input)) {
+ return (T) objectInput.readObject();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to deserialize object", e);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Could not read object ", e);
+ }
+ }
+
+ private static final class SchemaCompatibleObjectInputStream extends
ObjectInputStream {
+ private SchemaCompatibleObjectInputStream(ByteArrayInputStream input)
throws IOException {
+ super(input);
+ }
+
+ @Override
+ protected ObjectStreamClass readClassDescriptor() throws IOException,
ClassNotFoundException {
+ ObjectStreamClass descriptor = super.readClassDescriptor();
+ // Iceberg 1.11 changed Schema's generated UID without changing
its serialized fields. Accept only
+ // that known rolling-upgrade pair so unrelated or future
class-layout changes still fail closed.
+ if (Schema.class.getName().equals(descriptor.getName())
+ && descriptor.getSerialVersionUID() ==
ICEBERG_1_10_1_SCHEMA_UID
+ && LOCAL_SCHEMA_DESCRIPTOR.getSerialVersionUID() ==
ICEBERG_1_11_0_SCHEMA_UID) {
+ return LOCAL_SCHEMA_DESCRIPTOR;
+ }
+ return descriptor;
+ }
+
+ @Override
+ protected Class<?> resolveClass(ObjectStreamClass descriptor) throws
IOException, ClassNotFoundException {
+ String className = descriptor.getName();
+ if (className.indexOf('/') >= 0) {
+ // Replacing Schema's stream descriptor exposes JVM-style
array signatures with slashes; resolve
+ // them through the scanner's isolated class loader after
normalizing to Class.forName syntax.
+ return Class.forName(className.replace('/', '.'), false,
+ IcebergSerializationCompat.class.getClassLoader());
+ }
+ return super.resolveClass(descriptor);
+ }
+ }
+}
diff --git
a/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
index 5b1df1ab93a..a2b8e104afb 100644
---
a/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
+++
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
@@ -28,7 +28,6 @@ import com.google.common.base.Preconditions;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.io.CloseableIterator;
-import org.apache.iceberg.util.SerializationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,7 +54,7 @@ public class IcebergSysTableJniScanner extends JniScanner {
String serializedSplitParams = params.get("serialized_split");
Preconditions.checkArgument(serializedSplitParams != null &&
!serializedSplitParams.isEmpty(),
"serialized_split should not be empty");
- this.scanTask =
SerializationUtil.deserializeFromBase64(serializedSplitParams);
+ this.scanTask =
IcebergSerializationCompat.deserializeFromBase64(serializedSplitParams);
String requiredFieldsParam = params.get("required_fields");
Preconditions.checkArgument(requiredFieldsParam != null &&
!requiredFieldsParam.isEmpty(),
"required_fields should not be empty");
diff --git
a/fe/be-java-extensions/iceberg-metadata-scanner/src/test/java/org/apache/doris/iceberg/IcebergSerializationCompatTest.java
b/fe/be-java-extensions/iceberg-metadata-scanner/src/test/java/org/apache/doris/iceberg/IcebergSerializationCompatTest.java
new file mode 100644
index 00000000000..cc5588c4b95
--- /dev/null
+++
b/fe/be-java-extensions/iceberg-metadata-scanner/src/test/java/org/apache/doris/iceberg/IcebergSerializationCompatTest.java
@@ -0,0 +1,76 @@
+// 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.
+
+package org.apache.doris.iceberg;
+
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.util.Base64;
+
+public class IcebergSerializationCompatTest {
+ // An empty StaticDataTask serialized by Iceberg 1.10.1. It carries the
old Schema serialVersionUID while
+ // keeping all data neutral and local, so the fixture remains stable and
safe to commit.
+ private static final String ICEBERG_1_10_1_TASK =
"rO0ABXNyACFvcmcuYXBhY2hlLmljZWJlcmcuU3RhdGljRGF0YVRhc2t2PvVIpr/rlAIABEwADG1ldGFkYXRh"
+ +
"RmlsZXQAHUxvcmcvYXBhY2hlL2ljZWJlcmcvRGF0YUZpbGU7TAAPcHJvamVjdGVkU2NoZW1hdAAbTG9yZy9h"
+ +
"cGFjaGUvaWNlYmVyZy9TY2hlbWE7WwAEcm93c3QAIFtMb3JnL2FwYWNoZS9pY2ViZXJnL1N0cnVjdExpa2U7"
+ +
"TAALdGFibGVTY2hlbWFxAH4AAnhwcHNyABlvcmcuYXBhY2hlLmljZWJlcmcuU2NoZW1hXonoLcvZFnYCAARJ"
+ +
"AA5oaWdoZXN0RmllbGRJZEkACHNjaGVtYUlkWwASaWRlbnRpZmllckZpZWxkSWRzdAACW0lMAAZzdHJ1Y3R0"
+ +
"ACtMb3JnL2FwYWNoZS9pY2ViZXJnL3R5cGVzL1R5cGVzJFN0cnVjdFR5cGU7eHAAAAABAAAAAHVyAAJbSU26"
+ +
"YCZ26rKlAgAAeHAAAAAAc3IAKW9yZy5hcGFjaGUuaWNlYmVyZy50eXBlcy5UeXBlcyRTdHJ1Y3RUeXBlY2OW"
+ +
"YF+O53QCAAFbAAZmaWVsZHN0AC1bTG9yZy9hcGFjaGUvaWNlYmVyZy90eXBlcy9UeXBlcyROZXN0ZWRGaWVs"
+ +
"ZDt4cgAob3JnLmFwYWNoZS5pY2ViZXJnLnR5cGVzLlR5cGUkTmVzdGVkVHlwZWpUXj112XcBAgAAeHB1cgAt"
+ +
"W0xvcmcuYXBhY2hlLmljZWJlcmcudHlwZXMuVHlwZXMkTmVzdGVkRmllbGQ7A428r/O1h1gCAAB4cAAAAAFz"
+ +
"cgAqb3JnLmFwYWNoZS5pY2ViZXJnLnR5cGVzLlR5cGVzJE5lc3RlZEZpZWxkRnIcmOgj/wICAAdJAAJpZFoA"
+ +
"CmlzT3B0aW9uYWxMAANkb2N0ABJMamF2YS9sYW5nL1N0cmluZztMAA5pbml0aWFsRGVmYXVsdHQAKExvcmcv"
+ +
"YXBhY2hlL2ljZWJlcmcvZXhwcmVzc2lvbnMvTGl0ZXJhbDtMAARuYW1lcQB+ABJMAAR0eXBldAAfTG9yZy9h"
+ +
"cGFjaGUvaWNlYmVyZy90eXBlcy9UeXBlO0wADHdyaXRlRGVmYXVsdHEAfgATeHAAAAABAHBwdAACaWRzcgAs"
+ +
"b3JnLmFwYWNoZS5pY2ViZXJnLnR5cGVzLlByaW1pdGl2ZUxpa2VIb2xkZXLbKTPuyM89cAIAAUwADHR5cGVB"
+ +
"c1N0cmluZ3EAfgASeHB0AANpbnRwdXIAIFtMb3JnL2FwYWNoZS5pY2ViZXJnLlN0cnVjdExpa2U7MJKGbVap"
+ + "uFMCAAB4cAAAAABxAH4ACA==";
+
+ @Test
+ public void deserializesIceberg1101SystemTableTask() {
+ FileScanTask task =
IcebergSerializationCompat.deserializeFromBase64(ICEBERG_1_10_1_TASK);
+
+ Assert.assertEquals("id", task.schema().columns().get(0).name());
+ }
+
+ @Test
+ public void preservesCurrentIcebergSerialization() throws IOException {
+ Schema expected = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+
+ Schema actual = IcebergSerializationCompat.deserializeFromBase64(
+ serializeToBase64(expected));
+
+ Assert.assertTrue(expected.sameSchema(actual));
+ }
+
+ private static String serializeToBase64(Object value) throws IOException {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (ObjectOutputStream objectOutput = new ObjectOutputStream(output))
{
+ objectOutput.writeObject(value);
+ }
+ return Base64.getEncoder().encodeToString(output.toByteArray());
+ }
+}
diff --git a/fe/check/checkstyle/suppressions.xml
b/fe/check/checkstyle/suppressions.xml
index 505fcfff829..8ef6fe5c631 100644
--- a/fe/check/checkstyle/suppressions.xml
+++ b/fe/check/checkstyle/suppressions.xml
@@ -81,6 +81,8 @@ under the License.
<!-- ignore iceberg delete file index copied from
iceberg/DeleteFileIndex.java -->
<suppress files="org[\\/]apache[\\/]iceberg[\\/]DeleteFileIndex\.java"
checks="[a-zA-Z0-9]*"/>
+ <!-- Iceberg package access is required to preserve historical schema
binding in DataTableScan. -->
+ <suppress
files="org[\\/]apache[\\/]iceberg[\\/]SchemaAwareDataTableScan\.java"
checks="[a-zA-Z0-9]*"/>
<!-- ignore gensrc/thrift/ExternalTableSchema.thrift -->
<suppress files=".*thrift/schema/external/.*" checks=".*"/>
diff --git
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorTableKey.java
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorTableKey.java
index 41ac71c9930..e8893406750 100644
---
a/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorTableKey.java
+++
b/fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ConnectorTableKey.java
@@ -20,16 +20,19 @@ package org.apache.doris.connector.cache;
import java.util.Objects;
/**
- * Immutable cache key for {@link ConnectorMetadataCache}: {@code (db, table,
snapshotId, schemaId)}.
+ * Immutable cache key for {@link ConnectorMetadataCache}:
+ * {@code (db, table, snapshotId, schemaId, metadataGeneration)}.
*
* <p>Engine-agnostic (external-partition-derived-cache design doc §5, "cache
A"): a table's derived partition
* view is a pure function of its identity plus the MVCC coordinate it was
read at, so pinning that coordinate
* into the key is what makes the cache "always correct" — a new
snapshot/schema yields a new key, never a stale
* hit. Non-MVCC engines (hive) or engines without a separate schema version
pass {@code snapshotId = -1} /
- * {@code schemaId = -1}; the key still holds them, it just means
"unversioned" for that axis.
+ * {@code schemaId = -1}; the key still holds them, it just means
"unversioned" for that axis. Connectors whose
+ * derived view depends on another independently evolving metadata generation
may use
+ * {@code metadataGeneration}; other connectors leave it at {@code -1} through
the four-argument constructor.
*
* <p>{@link #matches} / {@link #matchesDb} back {@link
ConnectorMetadataCache#invalidateTable} /
- * {@link ConnectorMetadataCache#invalidateDb}, which must drop every
snapshot/schema of a (db, table) or
+ * {@link ConnectorMetadataCache#invalidateDb}, which must drop every metadata
generation of a (db, table) or
* every table of a db — mirrors the {@code matches}/{@code matchesDb} helpers
on the sibling connector caches
* ({@code MaxComputePartitionCache.PartitionKey}, {@code
HiveFileListingCache.FileListingKey}).
*/
@@ -38,12 +41,18 @@ public final class ConnectorTableKey {
private final String table;
private final long snapshotId;
private final long schemaId;
+ private final long metadataGeneration;
public ConnectorTableKey(String db, String table, long snapshotId, long
schemaId) {
+ this(db, table, snapshotId, schemaId, -1L);
+ }
+
+ public ConnectorTableKey(String db, String table, long snapshotId, long
schemaId, long metadataGeneration) {
this.db = db;
this.table = table;
this.snapshotId = snapshotId;
this.schemaId = schemaId;
+ this.metadataGeneration = metadataGeneration;
}
public String getDb() {
@@ -62,12 +71,16 @@ public final class ConnectorTableKey {
return schemaId;
}
- /** Whether this key belongs to the given (db, table), regardless of
snapshotId/schemaId. */
+ public long getMetadataGeneration() {
+ return metadataGeneration;
+ }
+
+ /** Whether this key belongs to the given (db, table), regardless of its
metadata coordinates. */
public boolean matches(String db, String table) {
return Objects.equals(this.db, db) && Objects.equals(this.table,
table);
}
- /** Whether this key belongs to the given db, regardless of
table/snapshotId/schemaId. */
+ /** Whether this key belongs to the given db, regardless of table or
metadata coordinates. */
public boolean matchesDb(String db) {
return Objects.equals(this.db, db);
}
@@ -83,18 +96,20 @@ public final class ConnectorTableKey {
ConnectorTableKey that = (ConnectorTableKey) o;
return snapshotId == that.snapshotId
&& schemaId == that.schemaId
+ && metadataGeneration == that.metadataGeneration
&& Objects.equals(db, that.db)
&& Objects.equals(table, that.table);
}
@Override
public int hashCode() {
- return Objects.hash(db, table, snapshotId, schemaId);
+ return Objects.hash(db, table, snapshotId, schemaId,
metadataGeneration);
}
@Override
public String toString() {
return "ConnectorTableKey{db=" + db + ", table=" + table
- + ", snapshotId=" + snapshotId + ", schemaId=" + schemaId +
'}';
+ + ", snapshotId=" + snapshotId + ", schemaId=" + schemaId
+ + ", metadataGeneration=" + metadataGeneration + '}';
}
}
diff --git
a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
index 8a6c1596fcc..e643e361d64 100644
---
a/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
+++
b/fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ConnectorMetadataCacheTest.java
@@ -41,6 +41,11 @@ public class ConnectorMetadataCacheTest {
return new ConnectorTableKey(db, table, snapshotId, schemaId);
}
+ private static ConnectorTableKey key(
+ String db, String table, long snapshotId, long schemaId, long
metadataGeneration) {
+ return new ConnectorTableKey(db, table, snapshotId, schemaId,
metadataGeneration);
+ }
+
private static ConnectorMetadataCache<String> newCache() {
return new ConnectorMetadataCache<>(ENGINE, "partition_view", new
HashMap<>());
}
@@ -108,6 +113,23 @@ public class ConnectorMetadataCacheTest {
Assertions.assertEquals(2, loads.get(), "distinct schemaId must
trigger a distinct load");
}
+ @Test
+ public void differentMetadataGenerationIsADistinctEntry() {
+ AtomicInteger loads = new AtomicInteger();
+ ConnectorMetadataCache<String> cache = newCache();
+
+ cache.get(key("db", "t", 1L, 1L, 10L), () -> "generation-10");
+ String second = cache.get(key("db", "t", 1L, 1L, 11L), () -> {
+ loads.incrementAndGet();
+ return "generation-11";
+ });
+
+ // Some table metadata evolves independently of both snapshot and
schema; treating this axis as part of
+ // key identity prevents a connector from serving a derived view built
for the preceding generation.
+ Assertions.assertEquals("generation-11", second);
+ Assertions.assertEquals(1, loads.get(), "a distinct metadata
generation must trigger a distinct load");
+ }
+
@Test
public void invalidateTableEvictsAllSnapshotsOfThatTableOnly() {
AtomicInteger loads = new AtomicInteger();
diff --git a/fe/fe-connector/fe-connector-hms-hive-shade/pom.xml
b/fe/fe-connector/fe-connector-hms-hive-shade/pom.xml
index 69db7fef637..a5c39c45d17 100644
--- a/fe/fe-connector/fe-connector-hms-hive-shade/pom.xml
+++ b/fe/fe-connector/fe-connector-hms-hive-shade/pom.xml
@@ -211,9 +211,9 @@ under the License.
</exclusions>
</dependency>
- <!-- iceberg-hive-metastore 1.10.1 supplies
org.apache.iceberg.hive.HiveCatalog (the hms flavor);
+ <!-- iceberg-hive-metastore supplies
org.apache.iceberg.hive.HiveCatalog (the hms flavor);
only fe-connector-iceberg loads it.
iceberg-core/api/common/bundled-guava + caffeine + slf4j
- are excluded — the iceberg plugin already ships iceberg-core
1.10.1 (which brings api/common/
+ are excluded — the iceberg plugin already ships the matching
iceberg-core version (which brings api/common/
guava/caffeine) as its own child-first jar, so re-bundling them
here would create duplicate
classes. Only the org.apache.iceberg.hive.* classes are kept, and
the relocation below rewrites
their org.apache.thrift refs to the shared private prefix so they
link against the same
@@ -221,7 +221,8 @@ under the License.
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-hive-metastore</artifactId>
- <version>1.10.1</version>
+ <!-- Keep HiveCatalog bytecode aligned with the Iceberg core
loaded by the connector. -->
+ <version>${iceberg.version}</version>
<optional>true</optional>
<exclusions>
<exclusion><groupId>org.apache.iceberg</groupId><artifactId>iceberg-core</artifactId></exclusion>
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index 1caeae8fce7..088057890c2 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -1976,21 +1976,24 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
ConnectorSession session, ConnectorTableHandle handle) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
try {
- // PERF-06 cache A: memoize the BUILT derived view keyed by (db,
table, snapshotId, schemaId) -- pure
- // function of the pinned MVCC coordinate (a new snapshot/schema
yields a new key, never a stale hit).
- // The lookup sits INSIDE executeAuthenticated so a miss runs the
loader (resolveTableForRead + the
- // remote PARTITIONS build) under the FE-injected auth scope; a
hit returns without any remote call. A
- // null cache (session=user / no-cache catalog) computes directly
every call. A resolved-empty -1
+ // PERF-06 cache A: memoize the BUILT derived view keyed by
snapshot/schema/spec generation.
+ // The lookup sits INSIDE executeAuthenticated: every lookup
resolves the live spec generation, while
+ // a miss additionally runs the remote PARTITIONS build under the
FE-injected auth scope. A null cache
+ // (session=user / no-cache catalog) computes directly every call.
A resolved-empty -1
// bypasses cache A because its numeric key is otherwise
indistinguishable from an unresolved latest
// read, even though only the former is a query-begin MVCC
boundary.
return executeAuthenticated(() -> {
if (mvccPartitionViewCache == null ||
iceHandle.isResolvedEmptySnapshot()) {
return Optional.of(buildMvccPartitionViewUncached(session,
iceHandle));
}
+ Table table = resolveTableForRead(session, iceHandle);
+ // A partition-spec commit does not create a snapshot or
schema, so the spec id is a separate
+ // cache generation; omitting it can retain a stale derived
view for the full cache TTL.
ConnectorTableKey key = new
ConnectorTableKey(iceHandle.getDbName(),
- iceHandle.getTableName(), iceHandle.getSnapshotId(),
iceHandle.getSchemaId());
+ iceHandle.getTableName(), iceHandle.getSnapshotId(),
iceHandle.getSchemaId(),
+ table.spec().specId());
return Optional.of(mvccPartitionViewCache.get(key,
- () -> buildMvccPartitionViewUncached(session,
iceHandle)));
+ () -> buildMvccPartitionView(table, iceHandle)));
});
} catch (Exception e) {
throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
@@ -2013,6 +2016,10 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
iceHandle.getResolvedEmptyPartitionStyle());
}
Table table = resolveTableForRead(session, iceHandle);
+ return buildMvccPartitionView(table, iceHandle);
+ }
+
+ private ConnectorMvccPartitionView buildMvccPartitionView(Table table,
IcebergTableHandle iceHandle) {
return IcebergPartitionUtils.buildMvccPartitionView(table,
iceHandle.getSnapshotId(),
TableIdentifier.of(iceHandle.getDbName(),
iceHandle.getTableName()), partitionCache);
}
@@ -2064,18 +2071,29 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
return Collections.emptyList();
}
try {
- // PERF-06 cache A: memoize the BUILT partition-info list keyed by
(db, table, snapshotId, schemaId).
- // The lookup sits INSIDE executeAuthenticated (a miss runs the
remote build under the auth scope; a hit
- // returns without a remote call). BYPASS the cache when the
filter is present -- that is not the
+ // PERF-06 cache A: memoize the BUILT partition-info list keyed by
snapshot/schema/spec generation.
+ // The lookup sits INSIDE executeAuthenticated (each lookup
resolves the current spec, and a miss runs
+ // the remote build under the auth scope). BYPASS the cache when
the filter is present -- that is not the
// pruning path (which always passes Optional.empty()) and is not
keyed by (snapshot, schema) alone -- or
// when the cache is null (session=user / no-cache catalog):
compute directly every call.
return executeAuthenticated(() -> {
if (listPartitionsViewCache == null || filter.isPresent()) {
return listPartitionsUncached(session, iceHandle);
}
+ Table table;
+ try {
+ table = resolveTableForRead(session, iceHandle);
+ } catch (NoSuchTableException e) {
+ LOG.warn("Iceberg table not found while listing
partitions: {}.{}",
+ iceHandle.getDbName(), iceHandle.getTableName(),
e);
+ return Collections.<ConnectorPartitionInfo>emptyList();
+ }
ConnectorTableKey key = new
ConnectorTableKey(iceHandle.getDbName(),
- iceHandle.getTableName(), iceHandle.getSnapshotId(),
iceHandle.getSchemaId());
- return listPartitionsViewCache.get(key, () ->
listPartitionsUncached(session, iceHandle));
+ iceHandle.getTableName(), iceHandle.getSnapshotId(),
iceHandle.getSchemaId(),
+ table.spec().specId());
+ // A partition-spec commit does not create a snapshot or
schema, so the spec id is a separate
+ // cache generation; omitting it can retain a stale derived
view for the full cache TTL.
+ return listPartitionsViewCache.get(key, () ->
listPartitions(table, iceHandle));
});
} catch (Exception e) {
throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
@@ -2099,6 +2117,10 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
iceHandle.getDbName(), iceHandle.getTableName(), e);
return Collections.<ConnectorPartitionInfo>emptyList();
}
+ return listPartitions(table, iceHandle);
+ }
+
+ private List<ConnectorPartitionInfo> listPartitions(Table table,
IcebergTableHandle iceHandle) {
return IcebergPartitionUtils.listPartitions(table,
TableIdentifier.of(iceHandle.getDbName(),
iceHandle.getTableName()), partitionCache);
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
index 105ca3ee506..b6a4144ea61 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionCache.java
@@ -36,18 +36,17 @@ import java.util.function.Supplier;
/**
* Per-catalog cache of an iceberg table's raw partition list (PERF-02), keyed
by {@code (TableIdentifier,
- * snapshotId)}. Restores the partition-info half of the legacy {@code
IcebergExternalMetaCache} that the SPI
+ * snapshotId, schemaId, specId)}. Restores the partition-info half of the
legacy
+ * {@code IcebergExternalMetaCache} that the SPI
* cutover dropped: the analysis-phase PARTITIONS metadata-table scan
* ({@link IcebergPartitionUtils#loadRawPartitionsUncached}, which the iceberg
SDK materializes by reading EVERY
* data+delete manifest of the snapshot) was re-run per query and re-run 4~6
times per MTMV refresh, with no
* cross-query reuse. The three consumers ({@code buildMvccPartitionView} for
the MVCC/MTMV partition view,
* {@code listPartitions} for {@code selectedPartitionNum}, {@code
listPartitionNames} for SHOW PARTITIONS) all
- * funnel through it, so they share a single scan per {@code (table,
snapshot)}.
+ * funnel through it, so they share a single scan per metadata generation.
*
- * <p><b>Snapshot-keyed, so always correct.</b> A snapshot is immutable, so
the derived partitions are a pure
- * function of the key; a new commit yields a new snapshot id (a new key ->
a live scan). Within the TTL the
- * snapshot id itself is held stable by {@link IcebergLatestSnapshotCache},
which is what makes the key stable
- * across queries and across the enumeration points of one MTMV refresh.
+ * <p><b>Metadata-generation-keyed.</b> A data commit yields a new snapshot
id, while schema/spec IDs fence
+ * metadata-only evolution that changes Iceberg's unified partition struct
without changing the snapshot.
*
* <p><b>No credential gate</b> (unlike {@link IcebergTableCache}): the cached
value is pure metadata (partition
* names, values, transforms, timestamps, snapshot ids) and carries no {@code
FileIO} / credential, so it is
@@ -62,14 +61,18 @@ import java.util.function.Supplier;
*/
final class IcebergPartitionCache {
- /** Immutable composite key: a table's partition list is distinct per
pinned snapshot id. */
+ /** Immutable composite key: partition projection depends on the snapshot
and current schema/spec metadata. */
static final class Key {
final TableIdentifier id;
final long snapshotId;
+ final int schemaId;
+ final int specId;
- Key(TableIdentifier id, long snapshotId) {
+ Key(TableIdentifier id, long snapshotId, int schemaId, int specId) {
this.id = id;
this.snapshotId = snapshotId;
+ this.schemaId = schemaId;
+ this.specId = specId;
}
@Override
@@ -81,12 +84,15 @@ final class IcebergPartitionCache {
return false;
}
Key that = (Key) o;
- return snapshotId == that.snapshotId && Objects.equals(id,
that.id);
+ return snapshotId == that.snapshotId
+ && schemaId == that.schemaId
+ && specId == that.specId
+ && Objects.equals(id, that.id);
}
@Override
public int hashCode() {
- return Objects.hash(id, snapshotId);
+ return Objects.hash(id, snapshotId, schemaId, specId);
}
}
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 ddc8cbde915..9d9738e130c 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
@@ -33,6 +33,7 @@ import org.apache.iceberg.MetadataTableUtils;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Partitioning;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
@@ -42,6 +43,7 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Type.TypeID;
import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.types.Types.TimestampType;
import org.apache.iceberg.util.JsonUtil;
import org.apache.iceberg.util.StructProjection;
@@ -502,8 +504,6 @@ final class IcebergPartitionUtils {
private static final String DAY = "day";
private static final String HOUR = "hour";
- // Iceberg partition field id starts at PARTITION_DATA_ID_START
(org.apache.iceberg.PartitionSpec).
- private static final int PARTITION_DATA_ID_START = 1000;
// Master IcebergUtils.UNKNOWN_SNAPSHOT_ID: an empty table / a null
last_updated_snapshot_id row.
private static final long UNKNOWN_SNAPSHOT_ID = -1;
@@ -758,30 +758,45 @@ final class IcebergPartitionUtils {
/**
* The cross-query PARTITIONS-scan de-duplication seam (PERF-02): when
{@code cache} is non-null the raw
* partition list is served from / populated into the per-catalog {@link
IcebergPartitionCache} keyed by
- * {@code (id, snapshotId)} — a snapshot is immutable, so the derived
partitions are a pure function of that
- * key and safe to reuse across queries (restoring the legacy
IcebergExternalMetaCache partition-info cache).
+ * {@code (id, snapshotId, schemaId, specId)}. The schema/spec generation
is required because metadata-only
+ * evolution can change the unified partition projection without creating
a snapshot.
* A {@code null} cache (offline unit tests / the no-cache catalog) reads
live every call. The cached list is
- * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; the loader's exception (e.g. the
- * dropped-partition-source-column {@link ValidationException}) propagates
verbatim so callers keep their own
- * degradation, and a failed scan is not cached.
+ * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; loader exceptions propagate
+ * verbatim so callers keep their own degradation, and a failed scan is
not cached.
*/
private static List<IcebergRawPartition> loadRawPartitions(TableIdentifier
id, Table table, long snapshotId,
IcebergPartitionCache cache) {
if (cache == null) {
return loadRawPartitionsUncached(table, snapshotId);
}
- return cache.getOrLoad(new IcebergPartitionCache.Key(id, snapshotId),
+ return cache.getOrLoad(new IcebergPartitionCache.Key(
+ id, snapshotId, table.schema().schemaId(),
table.spec().specId()),
() ->
Collections.unmodifiableList(loadRawPartitionsUncached(table, snapshotId)));
}
private static List<IcebergRawPartition> loadRawPartitionsUncached(Table
table, long snapshotId) {
+ StructType unifiedPartitionType = Partitioning.partitionType(table);
+ Map<Integer, Integer> partitionFieldOrdinals = new HashMap<>();
+ for (int i = 0; i < unifiedPartitionType.fields().size(); i++) {
+
partitionFieldOrdinals.put(unifiedPartitionType.fields().get(i).fieldId(), i);
+ }
Table partitionsTable =
MetadataTableUtils.createMetadataTableInstance(table,
MetadataTableType.PARTITIONS);
List<IcebergRawPartition> partitions = new ArrayList<>();
try (CloseableIterable<FileScanTask> tasks =
partitionsTable.newScan().useSnapshot(snapshotId).planFiles()) {
for (FileScanTask task : tasks) {
CloseableIterable<StructLike> rows = task.asDataTask().rows();
for (StructLike row : rows) {
- partitions.add(generateRawPartition(table, row));
+ PartitionSpec liveSpec = table.specs().get(row.get(1,
Integer.class));
+ boolean hasUnrepresentableField =
liveSpec.fields().stream()
+ .anyMatch(field ->
!partitionFieldOrdinals.containsKey(field.fieldId()));
+ if (hasUnrepresentableField) {
+ // Only specs represented by live partition rows
affect this snapshot. Rejecting an
+ // orphaned but file-free historical spec would hide
otherwise valid current partitions.
+ LOG.warn("Cannot represent a live historical partition
spec for iceberg table {}; "
+ + "reporting an empty partition display.",
table.name());
+ return Collections.emptyList();
+ }
+ partitions.add(generateRawPartition(table, row,
partitionFieldOrdinals));
}
}
} catch (IOException e) {
@@ -790,7 +805,8 @@ final class IcebergPartitionUtils {
return partitions;
}
- private static IcebergRawPartition generateRawPartition(Table table,
StructLike row) {
+ private static IcebergRawPartition generateRawPartition(
+ Table table, StructLike row, Map<Integer, Integer>
partitionFieldOrdinals) {
// PARTITIONS row layout: 0 partitionData, 1 spec_id, 2 record_count,
3 file_count,
// 4 total_data_file_size_in_bytes, 5..8 position/equality delete
stats, 9 last_updated_at,
// 10 last_updated_snapshot_id. Only 0/1/9/10 are needed by the MTMV
partition view.
@@ -805,11 +821,10 @@ final class IcebergPartitionUtils {
for (int i = 0; i < partitionSpec.fields().size(); ++i) {
PartitionField partitionField = partitionSpec.fields().get(i);
Class<?> fieldClass = partitionSpec.javaClasses()[i];
- int fieldId = partitionField.fieldId();
- // Iceberg partition field id starts at PARTITION_DATA_ID_START,
so the index into partitionData is
- // fieldId - PARTITION_DATA_ID_START.
- int index = fieldId - PARTITION_DATA_ID_START;
- Object o = partitionData.get(index, fieldClass);
+ // Iceberg 1.11 projects every metadata row into the table-wide
unified partition struct; a spec-local
+ // position can therefore point at a different evolved field, so
resolve the ordinal by field ID.
+ int ordinal = partitionFieldOrdinals.get(partitionField.fieldId());
+ Object o = partitionData.get(ordinal, fieldClass);
String fieldValue = o == null ? null : o.toString();
sb.append(partitionField.name()).append("=").append(fieldValue).append("/");
// Resolve the partition field's SOURCE column name
(case-preserved), matching the generic
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 b04b5ec82e0..3fb1472114a 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
@@ -66,9 +66,11 @@ import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.PositionDeletesScanTask;
import org.apache.iceberg.ScanTask;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaAwareDataTableScan;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SplittableScanTask;
+import org.apache.iceberg.SupportsDistributedScanPlanning;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
@@ -92,7 +94,6 @@ import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.ScanTaskUtil;
-import org.apache.iceberg.util.SerializationUtil;
import org.apache.iceberg.util.TableScanUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -493,7 +494,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
long threshold = sessionLong(session, NUM_FILES_IN_BATCH_MODE,
DEFAULT_NUM_FILES_IN_BATCH_MODE);
long fileCount = 0;
try (CloseableIterable<ManifestFile> matching = getMatchingManifest(
- snapshot.dataManifests(table.io()), table.specs(),
scan.filter())) {
+ snapshot.dataManifests(table.io()),
+ SchemaAwareDataTableScan.specsFor(table, scan.schema()),
scan.filter())) {
for (ManifestFile manifest : matching) {
// Manifest metadata counts (cheap — no per-file read). Null
guard for ancient manifests that
// omit the counts (legacy summed them unguarded; 0 is the
safe under-count, never over-streams).
@@ -888,7 +890,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* resolve the metadata table ({@link #resolveSysTable}), apply the
time-travel pin + predicate through the
* shared {@link #buildScan} (legacy {@code createTableScan} honors {@code
useSnapshot}/{@code useRef} on the
* metadata-table scan too — iceberg system tables are legal time-travel
targets), then serialize each
- * metadata {@code FileScanTask} ({@code
SerializationUtil.serializeToBase64}) into a JNI split carrying ONLY
+ * metadata {@code FileScanTask} into a rolling-upgrade-compatible JNI
split carrying ONLY
* {@code serialized_split} + {@code FORMAT_JNI} (see {@link
IcebergScanRange#populateRangeParams}). COUNT(*)
* pushdown does not apply (a metadata table has no snapshot-summary
count). The serialized {@code
* FileScanTask} bytes are consumed verbatim by BE's {@code
IcebergSysTableJniScanner}
@@ -960,7 +962,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
for (FileScanTask task : tasks) {
ranges.add(new IcebergScanRange.Builder()
.path(SYS_TABLE_DUMMY_PATH)
-
.serializedSplit(SerializationUtil.serializeToBase64(task))
+
.serializedSplit(IcebergSystemTableSerialization.serializeToBase64(task))
.build());
}
} catch (IOException e) {
@@ -1201,7 +1203,12 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
*/
private TableScan buildScan(Table table, IcebergTableHandle handle,
Optional<ConnectorExpression> filter,
ConnectorSession session) {
- TableScan scan = table.newScan();
+ Schema selectedSchema = !handle.isSystemTable() &&
handle.hasSnapshotPin()
+ ? pinnedSchema(table, handle) : table.schema();
+ // 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())
+ ? SchemaAwareDataTableScan.newScan(table) : table.newScan();
// MVCC / time-travel pin: a tag/branch pins by REF (so a later commit
to the ref is honored, legacy
// parity), else by snapshot id (legacy createTableScan: useRef when
info.getRef()!=null else useSnapshot).
if (handle.hasSnapshotPin() && supportsSnapshotSelection(handle)) {
@@ -1211,10 +1218,16 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
scan = scan.useSnapshot(handle.getSnapshotId());
}
}
+ // A latest MVCC pin may pair the current schema with the preceding
snapshot after a schema-only commit.
+ // Preserve that logical schema explicitly so cache metrics use new
field IDs and their initial defaults.
+ if (!handle.isSystemTable() &&
!scan.schema().sameSchema(selectedSchema)) {
+ scan = scan.project(selectedSchema);
+ }
if (filter.isPresent()) {
// Historical predicates must resolve names to the field ids of
the generation used for binding;
// using the current schema drops renamed predicates or can bind a
later reused name incorrectly.
- Schema predicateSchema = handle.hasSnapshotPin() ?
pinnedSchema(table, handle) : table.schema();
+ Schema predicateSchema = handle.isSystemTable() &&
handle.hasSnapshotPin()
+ ? pinnedSchema(table, handle) : selectedSchema;
List<Expression> predicates =
new IcebergPredicateConverter(predicateSchema,
resolveSessionZone(session)).convert(filter.get());
for (Expression predicate : predicates) {
@@ -2754,16 +2767,17 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (snapshot == null) {
return CloseableIterable.withNoopClose(Collections.emptyList());
}
- Expression filterExpr = combineFilter(filter, table, session);
- Map<Integer, PartitionSpec> specsById = table.specs();
+ Schema scanSchema = scan.schema();
+ Expression filterExpr = combineFilter(filter, scanSchema, session);
+ Map<Integer, PartitionSpec> specsById =
SchemaAwareDataTableScan.specsFor(table, scanSchema);
boolean caseSensitive = true;
Map<Integer, ResidualEvaluator> residualEvaluators = new HashMap<>();
specsById.forEach((id, spec) -> residualEvaluators.put(id,
ResidualEvaluator.of(spec, filterExpr, caseSensitive)));
InclusiveMetricsEvaluator metricsEvaluator =
- new InclusiveMetricsEvaluator(table.schema(), filterExpr,
caseSensitive);
- String schemaJson = SchemaParser.toJson(table.schema());
+ new InclusiveMetricsEvaluator(scanSchema, filterExpr,
caseSensitive);
+ String schemaJson = SchemaParser.toJson(scanSchema);
// Phase 1 (eager): partition-prune + cache-load delete manifests into
the delete-file index.
List<DeleteFile> deleteFiles = new ArrayList<>();
@@ -2781,6 +2795,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
deleteFiles.addAll(manifestCacheGet(manifest, table,
statsQueryId).getDeleteFiles());
}
DeleteFileIndex deleteIndex = DeleteFileIndex.builderFor(deleteFiles)
+ .schemasById(table.schemas())
.specsById(specsById)
.caseSensitive(caseSensitive)
.build();
@@ -2908,15 +2923,15 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
/**
* Combine the pushed predicate into one iceberg {@link Expression} for
manifest-level pruning, mirroring
* legacy {@code
conjuncts.stream().map(convertToIcebergExpr).filter(nonNull).reduce(alwaysTrue,
and)}. Reuses
- * the T02 {@link IcebergPredicateConverter} on the table's CURRENT
schema; an absent filter is
+ * the T02 {@link IcebergPredicateConverter} on the scan-bound schema; an
absent filter is
* {@code alwaysTrue()} (scan everything).
*/
- private Expression combineFilter(Optional<ConnectorExpression> filter,
Table table, ConnectorSession session) {
+ private Expression combineFilter(Optional<ConnectorExpression> filter,
Schema schema, ConnectorSession session) {
if (!filter.isPresent()) {
return Expressions.alwaysTrue();
}
List<Expression> predicates =
- new IcebergPredicateConverter(table.schema(),
resolveSessionZone(session)).convert(filter.get());
+ new IcebergPredicateConverter(schema,
resolveSessionZone(session)).convert(filter.get());
Expression combined = Expressions.alwaysTrue();
for (Expression predicate : predicates) {
combined = Expressions.and(combined, predicate);
@@ -3092,9 +3107,21 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
() -> tableCache.borrow(
TableIdentifier.of(handle.getDbName(),
handle.getTableName()), directLoader),
directLoader);
+ rejectServerSideScanPlanning(raw, handle);
return wrapTableForScan(raw);
}
+ private static void rejectServerSideScanPlanning(Table table,
IcebergTableHandle handle) {
+ if (table instanceof SupportsDistributedScanPlanning
+ && !((SupportsDistributedScanPlanning)
table).allowDistributedPlanning()) {
+ // Iceberg 1.11 marks REST server-planned tables this way. Doris
reads manifests and table.io()
+ // before planFiles(), when REST scan-scoped credentials do not
exist, so fail before any local I/O.
+ throw new DorisConnectorException("Iceberg server-side scan
planning is not supported for table "
+ + handle.getDbName() + "." + handle.getTableName()
+ + "; configure the REST catalog to use client-side scan
planning");
+ }
+ }
+
/**
* Loads the RAW iceberg table for {@code handle} through the cross-query
{@link IcebergTableCache} when
* enabled (the connector disables it for credential-dependent catalogs),
else a direct remote
@@ -3160,6 +3187,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
() ->
tableCache.borrow(TableIdentifier.of(handle.getDbName(), handle.getTableName()),
() -> ops.loadTable(handle.getDbName(),
handle.getTableName())),
() -> ops.loadTable(handle.getDbName(),
handle.getTableName()));
+ rejectServerSideScanPlanning(base, handle);
return MetadataTableUtils.createMetadataTableInstance(
base,
MetadataTableType.from(handle.getSysTableName()));
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSystemTableSerialization.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSystemTableSerialization.java
new file mode 100644
index 00000000000..97c702a4901
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSystemTableSerialization.java
@@ -0,0 +1,90 @@
+// 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.
+
+package org.apache.doris.connector.iceberg;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.util.SerializationUtil;
+
+import java.io.ObjectStreamClass;
+import java.io.ObjectStreamConstants;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+/** Keeps the Java-serialized Iceberg system-table task wire compatible with
Iceberg 1.10.1. */
+final class IcebergSystemTableSerialization {
+ static final long ICEBERG_1_10_1_SCHEMA_UID = 6812231194765760118L;
+ private static final long ICEBERG_1_11_0_SCHEMA_UID =
-1265875184407129845L;
+ private static final long LOCAL_SCHEMA_UID =
ObjectStreamClass.lookup(Schema.class).getSerialVersionUID();
+ private static final byte[] SCHEMA_CLASS_NAME =
Schema.class.getName().getBytes(StandardCharsets.UTF_8);
+
+ private IcebergSystemTableSerialization() {
+ }
+
+ static String serializeToBase64(Object value) {
+ byte[] bytes = SerializationUtil.serializeToBytes(value);
+ rewriteSchemaUid(bytes, ICEBERG_1_11_0_SCHEMA_UID,
ICEBERG_1_10_1_SCHEMA_UID);
+ return new String(Base64.getMimeEncoder().encode(bytes),
StandardCharsets.UTF_8);
+ }
+
+ static <T> T deserializeFromBase64(String base64) {
+ byte[] bytes =
Base64.getMimeDecoder().decode(base64.getBytes(StandardCharsets.UTF_8));
+ rewriteSchemaUid(bytes, ICEBERG_1_10_1_SCHEMA_UID,
ICEBERG_1_11_0_SCHEMA_UID);
+ return SerializationUtil.deserializeFromBytes(bytes);
+ }
+
+ static long schemaUid(String base64) {
+ byte[] bytes =
Base64.getMimeDecoder().decode(base64.getBytes(StandardCharsets.UTF_8));
+ int offset = schemaUidOffset(bytes);
+ return offset < 0 ? Long.MIN_VALUE : ByteBuffer.wrap(bytes, offset,
Long.BYTES).getLong();
+ }
+
+ private static void rewriteSchemaUid(byte[] bytes, long fromUid, long
toUid) {
+ // The 1.10 and 1.11 Schema classes have identical serialized fields.
Pinning only this known descriptor
+ // to the legacy UID lets old and new BE readers share one task wire
while future layouts fail closed.
+ if (LOCAL_SCHEMA_UID != ICEBERG_1_11_0_SCHEMA_UID) {
+ throw new IllegalStateException("Unsupported Iceberg Schema
serialVersionUID: " + LOCAL_SCHEMA_UID);
+ }
+ int offset = schemaUidOffset(bytes);
+ if (offset >= 0 && ByteBuffer.wrap(bytes, offset,
Long.BYTES).getLong() == fromUid) {
+ ByteBuffer.wrap(bytes, offset, Long.BYTES).putLong(toUid);
+ }
+ }
+
+ private static int schemaUidOffset(byte[] bytes) {
+ int descriptorSize = 1 + Short.BYTES + SCHEMA_CLASS_NAME.length +
Long.BYTES;
+ for (int i = 0; i <= bytes.length - descriptorSize; i++) {
+ if (bytes[i] != ObjectStreamConstants.TC_CLASSDESC
+ || bytes[i + 1] != (byte) (SCHEMA_CLASS_NAME.length >>> 8)
+ || bytes[i + 2] != (byte) SCHEMA_CLASS_NAME.length) {
+ continue;
+ }
+ boolean matches = true;
+ for (int j = 0; j < SCHEMA_CLASS_NAME.length; j++) {
+ if (bytes[i + 3 + j] != SCHEMA_CLASS_NAME[j]) {
+ matches = false;
+ break;
+ }
+ }
+ if (matches) {
+ return i + 3 + SCHEMA_CLASS_NAME.length;
+ }
+ }
+ return -1;
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/DLFTableOperations.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/DLFTableOperations.java
index d0a894430f3..17dd2242e75 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/DLFTableOperations.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/dlf/DLFTableOperations.java
@@ -33,6 +33,7 @@ public class DLFTableOperations extends HiveTableOperations {
String catalogName,
String database,
String table) {
- super(conf, metaClients, fileIO, catalogName, database, table);
+ // DLF does not configure an Iceberg KMS client; null preserves the
existing unencrypted behavior.
+ super(conf, metaClients, fileIO, null, catalogName, database, table);
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/DeleteFileIndex.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/DeleteFileIndex.java
index 5f997bdfadd..4c45cfc1b3e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/DeleteFileIndex.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/DeleteFileIndex.java
@@ -32,6 +32,9 @@ import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
@@ -64,7 +67,7 @@ import org.apache.iceberg.util.Tasks;
* DataFile)} or {@link #forEntry(ManifestEntry)} to get the delete files to
apply to a given data
* file.
*
- * Copied from
https://github.com/apache/iceberg/blob/apache-iceberg-1.9.1/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
+ * Copied from
https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
* Change DeleteFileIndex and some methods to public.
*
* <p>P6.2-T08 (catalog-spi): VENDORED into the iceberg connector module
(mirrors the identical fe-core copy)
@@ -381,6 +384,7 @@ public class DeleteFileIndex {
private final Iterable<DeleteFile> deleteFiles;
private long minSequenceNumber = 0L;
private Map<Integer, PartitionSpec> specsById = null;
+ private Map<Integer, Schema> schemasById = null;
private Expression dataFilter = Expressions.alwaysTrue();
private Expression partitionFilter = Expressions.alwaysTrue();
private PartitionSet partitionSet = null;
@@ -406,6 +410,13 @@ public class DeleteFileIndex {
return this;
}
+ // Doris' cache-backed planner builds this index outside
org.apache.iceberg and must preserve equality-delete
+ // fields from every historical schema, including fields absent from the
current table schema.
+ public Builder schemasById(Map<Integer, Schema> newSchemasById) {
+ this.schemasById = newSchemasById;
+ return this;
+ }
+
public Builder specsById(Map<Integer, PartitionSpec> newSpecsById) {
this.specsById = newSpecsById;
return this;
@@ -469,8 +480,14 @@ public class DeleteFileIndex {
try (CloseableIterable<ManifestEntry<DeleteFile>> reader =
deleteFile) {
for (ManifestEntry<DeleteFile> entry : reader) {
if (entry.dataSequenceNumber() > minSequenceNumber) {
+ DeleteFile file = entry.file();
+ // keep minimum stats to avoid memory pressure
+ Set<Integer> columns =
+ file.content() == FileContent.POSITION_DELETES
+ ?
Collections.singleton(MetadataColumns.DELETE_FILE_PATH.fieldId())
+ : Sets.newHashSet(file.equalityFieldIds());
// copy with stats for better filtering against data
file stats
- files.add(entry.file().copy());
+ files.add(ContentFileUtil.copy(file, true, columns));
}
}
} catch (IOException e) {
@@ -480,10 +497,21 @@ public class DeleteFileIndex {
return files;
}
+ private Collection<Schema> schemas() {
+ if (schemasById != null) {
+ return schemasById.values();
+ } else {
+ return
specsById.values().stream().map(PartitionSpec::schema).collect(Collectors.toList());
+ }
+ }
+
public DeleteFileIndex build() {
+ // Equality deletes may reference fields from historical schemas, so
index every known field ID.
+ Map<Integer, Types.NestedField> fieldsById =
Schema.indexFields(schemas());
+ Function<Integer, Types.NestedField> fieldLookup = fieldsById::get;
Iterable<DeleteFile> files = deleteFiles != null ? filterDeleteFiles() :
loadDeleteFiles();
- EqualityDeletes globalDeletes = new EqualityDeletes();
+ EqualityDeletes globalDeletes = new EqualityDeletes(fieldLookup);
PartitionMap<EqualityDeletes> eqDeletesByPartition =
PartitionMap.create(specsById);
PartitionMap<PositionDeletes> posDeletesByPartition =
PartitionMap.create(specsById);
Map<String, PositionDeletes> posDeletesByPath = Maps.newHashMap();
@@ -499,7 +527,7 @@ public class DeleteFileIndex {
}
break;
case EQUALITY_DELETES:
- add(globalDeletes, eqDeletesByPartition, file);
+ add(globalDeletes, eqDeletesByPartition, file, fieldLookup);
break;
default:
throw new UnsupportedOperationException("Unsupported content: " +
file.content());
@@ -546,7 +574,8 @@ public class DeleteFileIndex {
private void add(
EqualityDeletes globalDeletes,
PartitionMap<EqualityDeletes> deletesByPartition,
- DeleteFile file) {
+ DeleteFile file,
+ Function<Integer, Types.NestedField> fieldLookup) {
PartitionSpec spec = specsById.get(file.specId());
EqualityDeletes deletes;
@@ -555,10 +584,11 @@ public class DeleteFileIndex {
} else {
int specId = spec.specId();
StructLike partition = file.partition();
- deletes = deletesByPartition.computeIfAbsent(specId, partition,
EqualityDeletes::new);
+ Supplier<EqualityDeletes> initEqDeletes = () -> new
EqualityDeletes(fieldLookup);
+ deletes = deletesByPartition.computeIfAbsent(specId, partition,
initEqDeletes);
}
- deletes.add(spec, file);
+ deletes.add(file);
}
private Iterable<CloseableIterable<ManifestEntry<DeleteFile>>>
deleteManifestReaders() {
@@ -735,6 +765,8 @@ public class DeleteFileIndex {
Comparator.comparingLong(EqualityDeleteFile::applySequenceNumber);
private static final EqualityDeleteFile[] EMPTY_EQUALITY_DELETES = new
EqualityDeleteFile[0];
+ private final Function<Integer, Types.NestedField> fieldLookup;
+
// indexed state
private long[] seqs = null;
private EqualityDeleteFile[] files = null;
@@ -742,9 +774,13 @@ public class DeleteFileIndex {
// a buffer that is used to hold files before indexing
private volatile List<EqualityDeleteFile> buffer = Lists.newArrayList();
- public void add(PartitionSpec spec, DeleteFile file) {
+ EqualityDeletes(Function<Integer, Types.NestedField> fieldLookup) {
+ this.fieldLookup = fieldLookup;
+ }
+
+ public void add(DeleteFile file) {
Preconditions.checkState(buffer != null, "Can't add files upon
indexing");
- buffer.add(new EqualityDeleteFile(spec, file));
+ buffer.add(new EqualityDeleteFile(fieldLookup, file));
}
public DeleteFile[] filter(long seq, DataFile dataFile) {
@@ -810,15 +846,15 @@ public class DeleteFileIndex {
// an equality delete file wrapper that caches the converted boundaries for
faster boundary checks
// this class is not meant to be exposed beyond the delete file index
private static class EqualityDeleteFile {
- private final PartitionSpec spec;
+ private final Function<Integer, Types.NestedField> fieldLookup;
private final DeleteFile wrapped;
private final long applySequenceNumber;
private volatile List<Types.NestedField> equalityFields = null;
private volatile Map<Integer, Object> convertedLowerBounds = null;
private volatile Map<Integer, Object> convertedUpperBounds = null;
- EqualityDeleteFile(PartitionSpec spec, DeleteFile file) {
- this.spec = spec;
+ EqualityDeleteFile(Function<Integer, Types.NestedField> fieldLookup,
DeleteFile file) {
+ this.fieldLookup = fieldLookup;
this.wrapped = file;
this.applySequenceNumber = wrapped.dataSequenceNumber() - 1;
}
@@ -837,7 +873,8 @@ public class DeleteFileIndex {
if (equalityFields == null) {
List<Types.NestedField> fields = Lists.newArrayList();
for (int id : wrapped.equalityFieldIds()) {
- Types.NestedField field = spec.schema().findField(id);
+ Types.NestedField field = fieldLookup.apply(id);
+ Preconditions.checkArgument(field != null, "Cannot find field
for ID %s", id);
fields.add(field);
}
this.equalityFields = fields;
@@ -900,7 +937,7 @@ public class DeleteFileIndex {
if (bounds != null) {
for (Types.NestedField field : equalityFields()) {
int id = field.fieldId();
- Type type = spec.schema().findField(id).type();
+ Type type = field.type();
if (type.isPrimitiveType()) {
ByteBuffer bound = bounds.get(id);
if (bound != null) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
new file mode 100644
index 00000000000..dbef7b6c44c
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
@@ -0,0 +1,57 @@
+// 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.
+
+package org.apache.iceberg;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** A data-table scan that keeps partition specs bound to the scan's schema. */
+public final class SchemaAwareDataTableScan extends DataTableScan {
+
+ private SchemaAwareDataTableScan(Table table, Schema schema,
TableScanContext context) {
+ super(table, schema, context);
+ }
+
+ public static TableScan newScan(Table table) {
+ return new SchemaAwareDataTableScan(table, table.schema(),
TableScanContext.empty());
+ }
+
+ /** Returns every table spec rebound to {@code schema}. */
+ public static Map<Integer, PartitionSpec> specsFor(Table table, Schema
schema) {
+ if (schema.sameSchema(table.schema())) {
+ return table.specs();
+ }
+
+ Map<Integer, PartitionSpec> specs = new LinkedHashMap<>();
+ table.specs().forEach((id, spec) -> specs.put(id,
spec.toUnbound().bind(schema, true)));
+ return Collections.unmodifiableMap(specs);
+ }
+
+ @Override
+ protected Map<Integer, PartitionSpec> specs() {
+ // A metadata-only schema commit preserves the current snapshot ID, so
schema identity—not snapshot
+ // identity—must decide whether historical partition specs need
rebinding.
+ return specsFor(table(), tableSchema());
+ }
+
+ @Override
+ protected TableScan newRefinedScan(Table table, Schema schema,
TableScanContext context) {
+ return new SchemaAwareDataTableScan(table, schema, context);
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
index 97f4054640a..1b8a92ced7c 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorCacheTest.java
@@ -403,7 +403,7 @@ public class IcebergConnectorCacheTest {
// ============ PERF-02: partition-view cache (session=user gated) +
invalidation ============
private static IcebergPartitionCache.Key partKey(String db, String tbl,
long snapshotId) {
- return new IcebergPartitionCache.Key(TableIdentifier.of(db, tbl),
snapshotId);
+ return new IcebergPartitionCache.Key(TableIdentifier.of(db, tbl),
snapshotId, 0, 0);
}
@Test
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataPartitionViewCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataPartitionViewCacheTest.java
index 337a23c50ad..204266c1536 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataPartitionViewCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataPartitionViewCacheTest.java
@@ -50,11 +50,11 @@ import java.util.stream.Collectors;
* {@link ConnectorMetadataCache}) wired into {@link
IcebergConnectorMetadata#getMvccPartitionView} /
* {@link IcebergConnectorMetadata#listPartitions}. Uses the real {@link
InMemoryCatalog} +
* {@link RecordingIcebergCatalogOps} harness (no Mockito, no docker): the
cache sits ABOVE the per-query build,
- * whose first step is {@code resolveTableForRead -> catalogOps.loadTable}
(logged as {@code loadTable:db1.t1}), so
- * a cache HIT skips the whole loader and the {@code loadTable} count is the
enumeration counter — the SAME proxy
- * the sibling cache tests use ({@code
IcebergConnectorMetadataMvccTest.beginQuerySnapshot*Cache*}). The raw
- * partition cache (PERF-02) is passed null throughout, so the {@code
loadTable} count isolates cache A's effect.
- * The partition math/merge parity itself is covered by {@link
IcebergPartitionUtilsTest}.
+ * whose first step is {@code resolveTableForRead -> catalogOps.loadTable}
(logged as {@code loadTable:db1.t1}).
+ * The spec generation must be read before each lookup because an external
spec-only commit changes neither the
+ * snapshot nor schema id; object identity therefore proves a derived-cache
hit while load counts prove generation
+ * checks still happen. The raw partition cache (PERF-02) is passed null
throughout, so object identity comes
+ * directly from cache A. The partition math/merge parity itself is covered by
{@link IcebergPartitionUtilsTest}.
*/
public class IcebergConnectorMetadataPartitionViewCacheTest {
@@ -191,9 +191,9 @@ public class IcebergConnectorMetadataPartitionViewCacheTest
{
@Test
public void getMvccPartitionViewCachesDerivedViewAcrossQueries() {
- // WHY: cache A must memoize the BUILT MVCC view keyed by (db, table,
snapshotId, schemaId), so a repeated
- // query on the same pin skips the derived rebuild AND the underlying
loadTable/scan. MUTATION: not
- // consulting the cache (compute directly every call) -> loadTable
runs twice -> red.
+ // WHY: cache A must memoize the BUILT MVCC view keyed by
snapshot/schema/spec generation, so a repeated
+ // query on the same generation skips the derived rebuild. Resolving
the live table is still required to
+ // detect spec-only commits that do not change snapshot/schema ids.
TwoSnap f = twoSnapshotTable();
RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
ops.table = f.table;
@@ -205,7 +205,9 @@ public class IcebergConnectorMetadataPartitionViewCacheTest
{
Assertions.assertEquals(java.util.Arrays.asList("ts_day=100",
"ts_day=101"), mvccNames(first));
Assertions.assertEquals(mvccNames(first), mvccNames(second), "the
cached view is returned verbatim");
- Assertions.assertEquals(1, loadCount(ops), "a cache hit must not
re-enumerate (loadTable once)");
+ Assertions.assertSame(first.orElseThrow(), second.orElseThrow(),
+ "a cache hit must return the same derived view instance");
+ Assertions.assertEquals(2, loadCount(ops), "each lookup must resolve
the live spec generation");
}
@Test
@@ -259,6 +261,28 @@ public class
IcebergConnectorMetadataPartitionViewCacheTest {
Assertions.assertEquals(2, loadCount(ops), "a null (disabled) cache
must re-enumerate every call");
}
+ @Test
+ public void getMvccPartitionViewSpecOnlyCommitDoesNotReuseDerivedView() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ PartitionSpec daySpec =
PartitionSpec.builderFor(PARTITIONED_SCHEMA).day("ts").build();
+ Table table = catalog.createTable(TableIdentifier.of("db1", "t1"),
PARTITIONED_SCHEMA, daySpec,
+ Collections.singletonMap("format-version", "2"));
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ IcebergConnectorMetadata md = metadataWithMvccCache(ops, mvccCache());
+
+ ConnectorMvccPartitionView before = md.getMvccPartitionView(null,
handle()).orElseThrow();
+ table.updateSpec().removeField("ts_day").addField("id").commit();
+ ops.table = catalog.loadTable(TableIdentifier.of("db1", "t1"));
+ ConnectorMvccPartitionView after = md.getMvccPartitionView(null,
handle()).orElseThrow();
+
+ Assertions.assertEquals(ConnectorMvccPartitionView.Style.RANGE,
before.getStyle());
+
Assertions.assertEquals(ConnectorMvccPartitionView.Style.UNPARTITIONED,
after.getStyle(),
+ "a spec-only commit must not reuse the preceding derived
partition view");
+ }
+
@Test
public void resolvedEmptyPartitionViewIgnoresConcurrentFirstAppend() {
InMemoryCatalog catalog = new InMemoryCatalog();
@@ -347,8 +371,8 @@ public class IcebergConnectorMetadataPartitionViewCacheTest
{
@Test
public void listPartitionsCachesDerivedListAcrossQueries() {
- // WHY: the empty-filter pruning path must memoize the built
partition-info list. MUTATION: not consulting
- // the cache -> loadTable twice -> red.
+ // WHY: the empty-filter pruning path must memoize the built
partition-info list while resolving the live
+ // table on every lookup to detect spec-only commits.
TwoSnap f = twoSnapshotTable();
RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
ops.table = f.table;
@@ -362,7 +386,8 @@ public class IcebergConnectorMetadataPartitionViewCacheTest
{
Assertions.assertEquals(java.util.Arrays.asList("ts_day=100",
"ts_day=101"), names);
Assertions.assertEquals(names,
second.stream().map(ConnectorPartitionInfo::getPartitionName).collect(Collectors.toList()));
- Assertions.assertEquals(1, loadCount(ops), "a cache hit must not
re-enumerate (loadTable once)");
+ Assertions.assertSame(first, second, "a cache hit must return the same
derived list instance");
+ Assertions.assertEquals(2, loadCount(ops), "each lookup must resolve
the live spec generation");
}
@Test
@@ -401,4 +426,17 @@ public class
IcebergConnectorMetadataPartitionViewCacheTest {
md.listPartitions(null, handle(), Optional.empty());
Assertions.assertEquals(2, loadCount(ops), "invalidateAll must force a
re-enumeration");
}
+
+ @Test
+ public void listPartitionsSpecOnlyCommitDoesNotReuseDerivedList() {
+ TwoSnap f = twoSnapshotTable();
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = f.table;
+ IcebergConnectorMetadata md = metadataWithListCache(ops, listCache());
+
+ Assertions.assertEquals(2, md.listPartitions(null, handle(),
Optional.empty()).size());
+ f.table.updateSpec().removeField("ts_day").commit();
+ Assertions.assertTrue(md.listPartitions(null, handle(),
Optional.empty()).isEmpty(),
+ "a spec-only commit must not reuse a list built for the
preceding partition spec");
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionCacheTest.java
index 1cc83596399..315acb132b4 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionCacheTest.java
@@ -32,14 +32,13 @@ import java.util.concurrent.atomic.AtomicInteger;
/**
* Unit tests for {@link IcebergPartitionCache} (PERF-02). Mirrors {@link
IcebergTableCacheTest} but keys by
- * {@code (TableIdentifier, snapshotId)} and stores the raw partition list.
Covers within-TTL stability, the
- * {@code ttl <= 0} disable, invalidation, and the exception-propagation
guarantee that {@code listPartitions}'
- * dropped-partition-source-column degradation depends on.
+ * {@code (TableIdentifier, snapshotId, schemaId, specId)} and stores the raw
partition list. Covers within-TTL
+ * stability, the {@code ttl <= 0} disable, invalidation, and exception
propagation.
*/
public class IcebergPartitionCacheTest {
private static IcebergPartitionCache.Key key(String db, String tbl, long
snapshotId) {
- return new IcebergPartitionCache.Key(TableIdentifier.of(db, tbl),
snapshotId);
+ return new IcebergPartitionCache.Key(TableIdentifier.of(db, tbl),
snapshotId, 0, 0);
}
/** A raw partition list of the given size, distinguishable by size. */
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 6afffa80d48..5e6d2c744be 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
@@ -24,18 +24,13 @@ import
org.apache.doris.connector.spi.mvcc.ConnectorMvccPartitionView;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.FileScanTask;
-import org.apache.iceberg.MetadataTableType;
-import org.apache.iceberg.MetadataTableUtils;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.inmemory.InMemoryCatalog;
-import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -1099,13 +1094,11 @@ public class IcebergPartitionUtilsTest {
}
@Test
- public void
listPartitionsDegradesToEmptyWhenPartitionSourceColumnDropped() {
+ public void
listPartitionsDegradesWhenDroppedPartitionSourceCannotBeRepresented() {
// Partition-evolution regression
(external_table_p0/iceberg/test_iceberg_partition_evolution): a
- // HISTORICAL spec references a source column that was later DROPPED,
while the CURRENT spec stays
- // partitioned on a surviving column. Building the PARTITIONS metadata
table unifies the partition type
- // across ALL specs, so iceberg throws ValidationException ("Cannot
find source column for partition
- // field: ...") for the orphaned field. listPartitions is
display/enforcement metadata only (never the
- // read set), so it must degrade to an empty (UNPARTITIONED) list
instead of failing the whole query.
+ // HISTORICAL spec references a source column that is later DROPPED,
while the CURRENT spec stays
+ // partitioned on a surviving column. Iceberg 1.11 removes the
historical field from the unified
+ // partition type, so distinct old buckets become unrepresentable and
must not be reported as null.
InMemoryCatalog catalog = new InMemoryCatalog();
catalog.initialize("test", Collections.emptyMap());
catalog.createNamespace(Namespace.of("db1"));
@@ -1119,37 +1112,94 @@ public class IcebergPartitionUtilsTest {
Collections.singletonMap("format-version", "2"));
// A data file under the original (bucket-on-region) spec so the
PARTITIONS metadata scan has a row whose
// spec must be unified.
- table.newAppend().appendFile(DataFiles.builder(table.spec())
-
.withPath("s3://b/db1/t/f0.parquet").withFileSizeInBytes(100).withRecordCount(1)
-
.withPartitionPath("region_bucket=1").withFormat(FileFormat.PARQUET).build()).commit();
+ table.newAppend()
+ .appendFile(DataFiles.builder(table.spec())
+
.withPath("s3://b/db1/t/f0.parquet").withFileSizeInBytes(100).withRecordCount(1)
+
.withPartitionPath("region_bucket=1").withFormat(FileFormat.PARQUET).build())
+ .appendFile(DataFiles.builder(table.spec())
+
.withPath("s3://b/db1/t/f1.parquet").withFileSizeInBytes(100).withRecordCount(1)
+
.withPartitionPath("region_bucket=2").withFormat(FileFormat.PARQUET).build())
+ .commit();
// Evolve: drop the bucket(region) partition field and add
identity(id) — the CURRENT spec stays
// PARTITIONED (on the surviving id column), so listPartitions passes
the isUnpartitioned() early-return
// and genuinely reaches the metadata scan (guards this test against a
vacuous unpartitioned pass).
table.updateSpec().removeField("region_bucket").addField("id").commit();
table.newAppend().appendFile(DataFiles.builder(table.spec())
-
.withPath("s3://b/db1/t/f1.parquet").withFileSizeInBytes(100).withRecordCount(1)
+
.withPath("s3://b/db1/t/f2.parquet").withFileSizeInBytes(100).withRecordCount(1)
.withPartitionPath("id=5").withFormat(FileFormat.PARQUET).build()).commit();
+ IcebergPartitionCache cache = new IcebergPartitionCache(100, 1000);
+ List<String> beforeDrop =
IcebergPartitionUtils.listPartitions(catalog.loadTable(id), id, cache).stream()
+
.map(ConnectorPartitionInfo::getPartitionName).sorted().collect(Collectors.toList());
+ Assertions.assertEquals(Arrays.asList("id=5", "region_bucket=1",
"region_bucket=2"), beforeDrop);
+ long snapshotBeforeDrop = table.currentSnapshot().snapshotId();
// Drop the source column referenced only by the historical spec,
leaving that spec dangling.
table.updateSchema().deleteColumn("region").commit();
Table evolved = catalog.loadTable(id);
+ Assertions.assertEquals(snapshotBeforeDrop,
evolved.currentSnapshot().snapshotId(),
+ "schema-only evolution must keep the same snapshot and
exercise the cache-key boundary");
Assertions.assertTrue(evolved.spec().isPartitioned(),
"current spec must stay partitioned so listPartitions reaches
the metadata scan, not the "
+ "unpartitioned early-return (otherwise this test
would pass vacuously)");
- // Precondition — prove the raw iceberg partition-metadata scan
genuinely throws ValidationException here
- // (guards against a future iceberg that tolerates the dangling spec,
which would make this test vacuous).
- Assertions.assertThrows(ValidationException.class, () -> {
- Table partitionsTable =
MetadataTableUtils.createMetadataTableInstance(
- evolved, MetadataTableType.PARTITIONS);
- try (CloseableIterable<FileScanTask> tasks =
partitionsTable.newScan().planFiles()) {
- tasks.forEach(t -> { });
- }
- });
-
- // The fix: listPartitions swallows exactly that failure and reports
UNPARTITIONED (empty), so a
- // full-table select on such a table is not blocked by uncomputable
display metadata. MUTATION:
- // rethrowing (or removing the catch) -> this throws instead of
returning empty -> red.
-
Assertions.assertTrue(IcebergPartitionUtils.listPartitions(evolved).isEmpty());
+ Assertions.assertTrue(IcebergPartitionUtils.listPartitions(evolved,
id, cache).isEmpty(),
+ "unrepresentable historical buckets must degrade to an empty
partition display");
+ Assertions.assertEquals(2, cache.loadCountForTest(),
+ "a schema-only commit must not reuse partition metadata cached
under the preceding schema");
+ }
+
+ @Test
+ public void listPartitionsIgnoresUnrepresentableSpecWithoutLiveFiles() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()));
+ TableIdentifier id = TableIdentifier.of("db1", "live_specs_only");
+ Table table = catalog.createTable(id, schema,
+ PartitionSpec.builderFor(schema).bucket("region", 8).build(),
+ Collections.singletonMap("format-version", "2"));
+
table.updateSpec().removeField("region_bucket").addField("id").commit();
+ table.newAppend().appendFile(DataFiles.builder(table.spec())
+ .withPath("s3://b/db1/live_specs_only/f.parquet")
+ .withFileSizeInBytes(100).withRecordCount(1)
+
.withPartitionPath("id=5").withFormat(FileFormat.PARQUET).build()).commit();
+ table.updateSchema().deleteColumn("region").commit();
+
+ List<ConnectorPartitionInfo> partitions =
IcebergPartitionUtils.listPartitions(
+ catalog.loadTable(id), id, null);
+
+ Assertions.assertEquals(1, partitions.size());
+ Assertions.assertEquals("id=5", partitions.get(0).getPartitionName(),
+ "an unrepresentable historical spec with no live files must
not hide live partitions");
+ }
+
+ @Test
+ public void listPartitionsReadsValuesFromUnifiedPartitionStructByFieldId()
{
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "region",
Types.StringType.get()));
+ TableIdentifier id = TableIdentifier.of("db1",
"unified_partition_values");
+ Table table = catalog.createTable(id, schema,
+ PartitionSpec.builderFor(schema).bucket("region", 8).build(),
+ Collections.singletonMap("format-version", "2"));
+ table.newAppend().appendFile(DataFiles.builder(table.spec())
+
.withPath("s3://b/db1/t/f0.parquet").withFileSizeInBytes(100).withRecordCount(1)
+
.withPartitionPath("region_bucket=1").withFormat(FileFormat.PARQUET).build()).commit();
+
table.updateSpec().removeField("region_bucket").addField("id").commit();
+ table.newAppend().appendFile(DataFiles.builder(table.spec())
+
.withPath("s3://b/db1/t/f1.parquet").withFileSizeInBytes(100).withRecordCount(1)
+
.withPartitionPath("id=5").withFormat(FileFormat.PARQUET).build()).commit();
+
+ List<ConnectorPartitionInfo> partitions =
IcebergPartitionUtils.listPartitions(
+ catalog.loadTable(id), id, null);
+ List<String> names =
partitions.stream().map(ConnectorPartitionInfo::getPartitionName)
+ .sorted().collect(java.util.stream.Collectors.toList());
+
+ Assertions.assertEquals(Arrays.asList("id=5", "region_bucket=1"),
names);
}
@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 512c2ffa338..27b453c9c5c 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
@@ -45,6 +45,7 @@ import org.apache.doris.thrift.schema.external.TFieldPtr;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
@@ -58,11 +59,14 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.SupportsDistributedScanPlanning;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
@@ -73,15 +77,16 @@ import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
-import org.apache.iceberg.util.SerializationUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
import java.nio.ByteBuffer;
+import java.nio.file.Path;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
@@ -107,6 +112,9 @@ import java.util.function.UnaryOperator;
*/
public class IcebergScanPlanProviderTest {
+ @TempDir
+ Path tempDir;
+
private static final Schema SCHEMA = new Schema(
Types.NestedField.required(1, "id", Types.IntegerType.get()),
Types.NestedField.optional(2, "name", Types.StringType.get()));
@@ -128,6 +136,15 @@ public class IcebergScanPlanProviderTest {
return catalog.createTable(TableIdentifier.of("db1", name), schema,
spec, null, props);
}
+ private Table createPersistedTable(String name, Schema schema,
PartitionSpec spec, Map<String, String> props) {
+ return new HadoopTables(new Configuration()).create(
+ schema, spec, props, tempDir.resolve(name).toUri().toString());
+ }
+
+ private static Table reloadPersistedTable(Table table) {
+ return new HadoopTables(new Configuration()).load(table.location());
+ }
+
private static DataFile dataFile(PartitionSpec spec, String path, long
sizeBytes, List<Long> splitOffsets,
String partitionPath) {
return dataFile(spec, path, sizeBytes, splitOffsets, partitionPath,
FileFormat.PARQUET);
@@ -159,6 +176,20 @@ public class IcebergScanPlanProviderTest {
(proxy, method, args) -> method.getName().equals("io") ?
fileIO : invoke(method, table, args));
}
+ private static Table serverPlannedTable(Table table, AtomicBoolean
localMetadataAccessed) {
+ return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
+ new Class<?>[] {Table.class,
SupportsDistributedScanPlanning.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("allowDistributedPlanning")) {
+ return false;
+ }
+ if (method.getName().equals("newScan") ||
method.getName().equals("io")) {
+ localMetadataAccessed.set(true);
+ }
+ return invoke(method, table, args);
+ });
+ }
+
private static Table tableWithMissingManifestRowsAndIo(Table table, FileIO
fileIO) {
return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
new Class<?>[] {Table.class},
(proxy, method, args) -> {
@@ -298,6 +329,58 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals("t1", ops.lastLoadTable);
}
+ @Test
+ public void planScanRejectsServerPlanningBeforeReplacingHistoricalScan() {
+ Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(
+ table.spec(), "s3://b/db1/t1/f.parquet", 100, null,
null)).commit();
+ long historicalSnapshotId = table.currentSnapshot().snapshotId();
+ int historicalSchemaId = table.schema().schemaId();
+ table.updateSchema().renameColumn("name", "renamed_name").commit();
+ AtomicBoolean localMetadataAccessed = new AtomicBoolean();
+ IcebergScanPlanProvider provider =
providerOver(serverPlannedTable(table, localMetadataAccessed));
+ IcebergTableHandle handle = new IcebergTableHandle("db1", "t1")
+ .withSnapshot(historicalSnapshotId, null, historicalSchemaId);
+
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.planScan(emptySession(),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList()).build()));
+
+ Assertions.assertTrue(failure.getMessage().contains("server-side scan
planning"), failure.getMessage());
+ Assertions.assertFalse(localMetadataAccessed.get(),
+ "rejection must happen before replacing the native scan or
opening local metadata");
+ }
+
+ @Test
+ public void
streamingEstimateRejectsServerPlanningBeforeReadingLocalManifests() {
+ Table table = threeFileTable();
+ AtomicBoolean localMetadataAccessed = new AtomicBoolean();
+ IcebergScanPlanProvider provider =
providerOver(serverPlannedTable(table, localMetadataAccessed));
+
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.streamingSplitEstimate(batchSession(1, true),
+ new IcebergTableHandle("db1", "t1"), Optional.empty(),
false));
+
+ Assertions.assertTrue(failure.getMessage().contains("server-side scan
planning"), failure.getMessage());
+ Assertions.assertFalse(localMetadataAccessed.get(),
+ "streaming dispatch must reject before table.io() reads local
manifests");
+ }
+
+ @Test
+ public void
scanPropertiesRejectServerPlanningBeforeReadingTableCredentials() {
+ Table table = threeFileTable();
+ AtomicBoolean localMetadataAccessed = new AtomicBoolean();
+ IcebergScanPlanProvider provider =
providerOver(serverPlannedTable(table, localMetadataAccessed));
+
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.getScanNodeProperties(emptySession(), new
IcebergTableHandle("db1", "t1"),
+ Collections.emptyList(), Optional.empty()));
+
+ Assertions.assertTrue(failure.getMessage().contains("server-side scan
planning"), failure.getMessage());
+ Assertions.assertFalse(localMetadataAccessed.get(),
+ "scan properties must reject before reading table.io() instead
of planned-scan credentials");
+ }
+
@Test
public void planScanMissingMetadataFileHasStableTableError() {
RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
@@ -315,11 +398,13 @@ public class IcebergScanPlanProviderTest {
@Test
public void planScanMissingManifestListHasStableTableError() {
- Table table = createTable("missing_manifest_list", SCHEMA,
PartitionSpec.unpartitioned());
+ Table table = createPersistedTable(
+ "missing_manifest_list", SCHEMA,
PartitionSpec.unpartitioned(), Collections.emptyMap());
table.newAppend().appendFile(dataFile(table.spec(),
"s3://b/db/missing_manifest_list/f1.parquet", 1024, null,
null)).commit();
table.io().deleteFile(table.currentSnapshot().manifestListLocation());
- IcebergScanPlanProvider provider = providerOver(table);
+ // Iceberg 1.11 caches parsed manifests on Snapshot; reload from
metadata to exercise the missing file.
+ IcebergScanPlanProvider provider =
providerOver(reloadPersistedTable(table));
DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
() -> provider.planScan(emptySession(),
ConnectorScanRequest.builder(
@@ -330,11 +415,13 @@ public class IcebergScanPlanProviderTest {
@Test
public void streamingMissingManifestListHasStableTableError() {
- Table table = createTable("missing_stream_manifest", SCHEMA,
PartitionSpec.unpartitioned());
+ Table table = createPersistedTable(
+ "missing_stream_manifest", SCHEMA,
PartitionSpec.unpartitioned(), Collections.emptyMap());
table.newAppend().appendFile(dataFile(table.spec(),
"s3://b/db/missing_stream_manifest/f1.parquet", 1024, null,
null)).commit();
table.io().deleteFile(table.currentSnapshot().manifestListLocation());
- IcebergScanPlanProvider provider = providerOver(table);
+ // Iceberg 1.11 caches parsed manifests on Snapshot; reload from
metadata to exercise the missing file.
+ IcebergScanPlanProvider provider =
providerOver(reloadPersistedTable(table));
IcebergTableHandle handle = new IcebergTableHandle("db1",
"missing_stream_manifest");
DorisConnectorException estimate =
Assertions.assertThrows(DorisConnectorException.class,
@@ -1539,7 +1626,7 @@ public class IcebergScanPlanProviderTest {
.build());
Assertions.assertEquals(1, ranges.size(), "one data file -> one
metadata split");
- FileScanTask task = SerializationUtil.deserializeFromBase64(
+ FileScanTask task =
IcebergSystemTableSerialization.deserializeFromBase64(
((IcebergScanRange) ranges.get(0)).getSerializedSplit());
Assertions.assertNotNull(task.schema().findField("file_size_in_bytes"),
"the requested column must survive in the projected task
schema");
@@ -1582,7 +1669,7 @@ public class IcebergScanPlanProviderTest {
.build());
Assertions.assertEquals(1, ranges.size(), "one commit -> one
$snapshots metadata split");
- FileScanTask task = SerializationUtil.deserializeFromBase64(
+ FileScanTask task =
IcebergSystemTableSerialization.deserializeFromBase64(
((IcebergScanRange) ranges.get(0)).getSerializedSplit());
try (CloseableIterable<StructLike> rows = task.asDataTask().rows()) {
Iterator<StructLike> it = rows.iterator();
@@ -1648,11 +1735,18 @@ public class IcebergScanPlanProviderTest {
.withFileSizeInBytes(100)
.withRecordCount(1)
.build();
- Table table = tableWithPositionDelete(deleteFile);
+ Table table = createPersistedTable(
+ "missing_position_delete_manifest", SCHEMA,
PartitionSpec.unpartitioned(), Collections.emptyMap());
+ table.newAppend()
+ .appendFile(dataFile(table.spec(), "s3://b/db/t1/f1.parquet",
512, null, null))
+ .commit();
+ table.newRowDelta().addDeletes(deleteFile).commit();
table.io().deleteFile(table.currentSnapshot().manifestListLocation());
+ // Iceberg 1.11 caches parsed manifests on Snapshot; reload from
metadata to exercise the missing file.
+ Table reloaded = reloadPersistedTable(table);
DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
- () -> planPositionDeletes(table, Collections.emptyList()));
+ () -> planPositionDeletes(reloaded, Collections.emptyList()));
Assertions.assertTrue(ex.getMessage().contains(
"Metadata not found in metadata location for table db1.t1"),
ex.getMessage());
}
@@ -1764,6 +1858,18 @@ public class IcebergScanPlanProviderTest {
"a metadata table is still not base-spec partitioned -> no
path_partition_keys");
}
+ @Test
+ public void jniSystemTableNeedsNoRollingUpgradeFence() {
+ Table table = createTable("sys_upgrade", SCHEMA,
PartitionSpec.unpartitioned());
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, IcebergTableHandle.forSystemTable("db1", "sys_upgrade",
"snapshots", -1L, null, -1L),
+ Collections.emptyList(), Optional.empty());
+
+
Assertions.assertFalse(props.containsKey(ScanNodePropertyKeys.REQUIRED_CURRENT_BACKEND_SEMANTICS));
+ }
+
@Test
public void
getScanNodePropertiesForPositionDeletesRowRequiresCurrentBackendSemantics() {
Table table = tableWithPositionDelete(
@@ -1832,6 +1938,97 @@ public class IcebergScanPlanProviderTest {
Assertions.assertTrue(pinned.get(0).getPath().get().endsWith("f1.parquet"));
}
+ @Test
+ public void planScanHistoricalPredicateSurvivesColumnRename() {
+ assertHistoricalPredicatePlansAfterSchemaEvolution(false);
+ }
+
+ @Test
+ public void planScanHistoricalPredicateSurvivesColumnDrop() {
+ assertHistoricalPredicatePlansAfterSchemaEvolution(true);
+ }
+
+ private void assertHistoricalPredicatePlansAfterSchemaEvolution(boolean
dropColumn) {
+ Schema historicalSchema = new Schema(
+ Types.NestedField.optional(1, "x", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "y", Types.IntegerType.get()),
+ Types.NestedField.optional(3, "part",
Types.IntegerType.get()));
+ Table table = createTable(
+ "historical_predicate_after_" + (dropColumn ? "drop" :
"rename"),
+ historicalSchema, PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newFastAppend()
+ .appendFile(dataFile(table.spec(),
"s3://b/db/historical.parquet", 1024, null, null))
+ .commit();
+ long historicalSnapshotId = table.currentSnapshot().snapshotId();
+ int historicalSchemaId = table.currentSnapshot().schemaId();
+
+ if (dropColumn) {
+ table.updateSchema().deleteColumn("x").commit();
+ } else {
+ table.updateSchema().renameColumn("x", "renamed_x").commit();
+ }
+
+ assertHistoricalPredicatePlans(table, historicalSnapshotId,
historicalSchemaId);
+
+ table.newFastAppend()
+ .appendFile(dataFile(table.spec(),
"s3://b/db/current.parquet", 1024, null, null))
+ .commit();
+
+ assertHistoricalPredicatePlans(table, historicalSnapshotId,
historicalSchemaId);
+ }
+
+ private static void assertHistoricalPredicatePlans(
+ Table table, long historicalSnapshotId, int historicalSchemaId) {
+ IcebergTableHandle historicalHandle = new IcebergTableHandle("db1",
"t1")
+ .withSnapshot(historicalSnapshotId, null, historicalSchemaId);
+ List<ConnectorScanRange> ranges = providerOver(table).planScan(
+ emptySession(), ConnectorScanRequest.builder(historicalHandle,
Collections.emptyList())
+ .filter(Optional.of(eqInt("x", 1)))
+ .build());
+
+ // Historical predicates must remain bound to the snapshot schema
after later schema evolution.
+ Assertions.assertEquals(1, ranges.size());
+
Assertions.assertTrue(ranges.get(0).getPath().get().endsWith("historical.parquet"));
+ }
+
+ @Test
+ public void
streamingDispatchRebindsPartitionSpecsForHistoricalSchemaBeforeAndAfterSnapshotAdvance()
+ throws IOException {
+ Schema historicalSchema = new Schema(
+ Types.NestedField.optional(1, "x", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "value",
Types.StringType.get()));
+ PartitionSpec historicalSpec =
PartitionSpec.builderFor(historicalSchema).identity("x").build();
+ Table table = createTable("historical_batch_estimate",
historicalSchema, historicalSpec,
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/historical_batch_estimate/old.parquet", 1024, null,
"x=1")).commit();
+ long historicalSnapshotId = table.currentSnapshot().snapshotId();
+ int historicalSchemaId = table.currentSnapshot().schemaId();
+ table.updateSchema().renameColumn("x", "renamed_x").commit();
+
+ assertHistoricalStreamingDispatch(table, historicalSnapshotId,
historicalSchemaId);
+
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/historical_batch_estimate/new.parquet", 1024, null,
"x=2")).commit();
+ assertHistoricalStreamingDispatch(table, historicalSnapshotId,
historicalSchemaId);
+ }
+
+ private static void assertHistoricalStreamingDispatch(
+ Table table, long historicalSnapshotId, int historicalSchemaId)
throws IOException {
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"historical_batch_estimate")
+ .withSnapshot(historicalSnapshotId, null, historicalSchemaId);
+ IcebergScanPlanProvider provider = providerOver(table);
+ ConnectorSession session = batchSession(1, true);
+ Optional<ConnectorExpression> filter = Optional.of(eqInt("x", 1));
+
+ Assertions.assertEquals(1L, provider.streamingSplitEstimate(session,
handle, filter, false));
+ List<ConnectorScanRange> ranges = drain(provider.streamSplits(
+ session, handle, Collections.emptyList(), filter, -1L));
+ Assertions.assertEquals(1, ranges.size());
+
Assertions.assertTrue(ranges.get(0).getPath().get().endsWith("old.parquet"));
+ }
+
@Test
public void planScanResolvedEmptySnapshotIgnoresConcurrentFirstAppend() {
// MERGE may resolve its read snapshot while the target has no
snapshots, then race with the first append.
@@ -1945,6 +2142,29 @@ public class IcebergScanPlanProviderTest {
Assertions.assertTrue(pinned.get(0).getPath().get().endsWith("f1.parquet"));
}
+ @Test
+ public void planScanPinnedToBranchBindsPredicateToCurrentSchema() {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "score",
Types.IntegerType.get()));
+ Table table = createTable("branch_predicate", schema,
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(
+ table.spec(), "s3://b/db/branch_predicate/f1.parquet", 1024,
null, null)).commit();
+ long branchSnapshotId = table.currentSnapshot().snapshotId();
+ table.manageSnapshots().createBranch("b1", branchSnapshotId).commit();
+ table.updateSchema().renameColumn("score", "grade").commit();
+ IcebergTableHandle branchHandle = new IcebergTableHandle("db1",
"branch_predicate")
+ .withSnapshot(branchSnapshotId, "b1",
table.schema().schemaId());
+
+ List<ConnectorScanRange> ranges = providerOver(table).planScan(
+ emptySession(), ConnectorScanRequest.builder(branchHandle,
Collections.emptyList())
+ .filter(Optional.of(eqInt("grade", 1)))
+ .build());
+
+ Assertions.assertEquals(1, ranges.size());
+
Assertions.assertTrue(ranges.get(0).getPath().get().endsWith("f1.parquet"));
+ }
+
@Test
public void countPushdownFollowsTheSnapshotPin() {
// f1=1000/100=10 records (S1); + f2=2000/100=20 -> latest
total-records 30. Pinned to S1 the count is
@@ -3274,6 +3494,68 @@ public class IcebergScanPlanProviderTest {
Assertions.assertTrue(cache.size() >= 2, "the data + delete manifests
must both be cached");
}
+ @Test
+ public void
streamSplitsManifestCacheResolvesDroppedEqualityDeleteFieldAfterReload() throws
IOException {
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "old_key",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "value",
Types.StringType.get()));
+ Table table = createPersistedTable("dropped_cache_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/dropped_cache_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/dropped_cache_key/eq.parquet", FileFormat.PARQUET,
1)).commit();
+ table.updateSchema().deleteColumn("old_key").commit();
+ Table reloaded = reloadPersistedTable(table);
+ IcebergManifestCache cache = new IcebergManifestCache();
+
+ List<ConnectorScanRange> ranges =
drain(manifestProvider(manifestCacheProps(), reloaded, cache)
+ .streamSplits(emptySession(), new IcebergTableHandle("db1",
"dropped_cache_key"),
+ Collections.emptyList(), Optional.empty(), -1L));
+
+ Assertions.assertEquals(1, ranges.size());
+ Assertions.assertEquals(1, deleteCount(ranges.get(0)));
+ Assertions.assertEquals(0L, cache.takeStats("q")[2],
+ "historical equality-delete keys must not force the cache path
to fail");
+ }
+
+ @Test
+ public void manifestCacheUsesLatestSchemaForSchemaOnlyMvccPin() throws
IOException {
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "k", Types.IntegerType.get()));
+ Table table = createPersistedTable("latest_schema_cache", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ Map<Integer, ByteBuffer> oldBounds = Collections.singletonMap(
+ 2, Conversions.toByteBuffer(Types.IntegerType.get(), 1));
+ DataFile oldFile = DataFiles.builder(table.spec())
+ .withPath(table.location() + "/old.parquet")
+ .withFileSizeInBytes(100)
+ .withRecordCount(1)
+ .withMetrics(new Metrics(1L, null, null, null, null,
oldBounds, oldBounds))
+ .withFormat(FileFormat.PARQUET)
+ .build();
+ table.newAppend().appendFile(oldFile).commit();
+ long pinnedSnapshotId = table.currentSnapshot().snapshotId();
+
table.updateSchema().allowIncompatibleChanges().deleteColumn("k").commit();
+ table.updateSchema().addRequiredColumn(
+ "k", Types.IntegerType.get(), Literal.of(7)).commit();
+ Table reloaded = reloadPersistedTable(table);
+ IcebergTableHandle latestPin = new IcebergTableHandle("db1",
"latest_schema_cache")
+ .withSnapshot(pinnedSnapshotId, null,
reloaded.schema().schemaId());
+
+ IcebergManifestCache cache = new IcebergManifestCache();
+ List<ConnectorScanRange> ranges = manifestProvider(
+ manifestCacheProps(), reloaded, cache).planScan(
+ emptySession(),
ConnectorScanRequest.builder(latestPin, Collections.emptyList())
+ .filter(Optional.of(eqInt("k", 7))).build());
+
+ Assertions.assertEquals(1, ranges.size(),
+ "the new field's initial default matches old files and must
not be pruned by retired-field stats");
+ Assertions.assertEquals(1, cache.size(), "the assertion must exercise
the manifest-cache path");
+ Assertions.assertEquals(0L, cache.takeStats("q")[2], "the cache path
must not fall back to the SDK");
+ }
+
// --- T09: vended credentials (extractVendedToken + static/vended
location.* + URI threading) ---
@Test
@@ -3842,7 +4124,7 @@ public class IcebergScanPlanProviderTest {
@Test
public void planScanForSystemTableSerializesEachFileScanTaskAsJniSplit() {
// A $snapshots handle plans through the metadata table
(MetadataTableUtils.createMetadataTableInstance):
- // each metadata FileScanTask is serialized
(SerializationUtil.serializeToBase64) and emitted as a JNI
+ // each metadata FileScanTask is serialized with the Iceberg 1.10.1
Schema UID and emitted as a JNI
// split carrying ONLY serialized_split + FORMAT_JNI +
table_level_row_count=-1, mirroring legacy
// IcebergScanNode.doGetSystemTableSplits + setIcebergParams.
MUTATION: routing the sys handle through
// the normal data-file path (resolveTable + buildRange) -> the range
carries the f1.parquet path and no
@@ -3862,6 +4144,8 @@ public class IcebergScanPlanProviderTest {
String serialized = ((IcebergScanRange)
range).getSerializedSplit();
Assertions.assertNotNull(serialized, "every sys split must carry a
serialized FileScanTask");
Assertions.assertFalse(serialized.isEmpty());
+
Assertions.assertEquals(IcebergSystemTableSerialization.ICEBERG_1_10_1_SCHEMA_UID,
+ IcebergSystemTableSerialization.schemaUid(serialized));
TFileRangeDesc rangeDesc = populate(range);
Assertions.assertEquals(TFileFormatType.FORMAT_JNI,
rangeDesc.getFormatType());
Assertions.assertEquals(serialized,
@@ -3874,7 +4158,8 @@ public class IcebergScanPlanProviderTest {
public void
planScanForSystemTableSplitDeserializesThroughTheBeJniReaderPath() throws
Exception {
// The strongest FE-reachable byte-shape parity check: the
serialized_split must be consumable EXACTLY
// as BE's IcebergSysTableJniScanner consumes it —
- // SerializationUtil.deserializeFromBase64(...).asDataTask().rows() —
and must carry the METADATA-table
+ //
IcebergSystemTableSerialization.deserializeFromBase64(...).asDataTask().rows()
— and must carry the
+ // METADATA-table
// schema ($snapshots), not the base table's. (Cross-version /
classloader interop is P6.8 docker e2e.)
// MUTATION: serializing anything other than the FileScanTask (e.g.
the DataFile) -> deserialize /
// asDataTask() fails or yields the wrong schema -> red.
@@ -3891,7 +4176,8 @@ public class IcebergScanPlanProviderTest {
long snapshotRows = 0;
for (ConnectorScanRange range : ranges) {
FileScanTask task =
-
SerializationUtil.deserializeFromBase64(((IcebergScanRange)
range).getSerializedSplit());
+ IcebergSystemTableSerialization.deserializeFromBase64(
+ ((IcebergScanRange) range).getSerializedSplit());
// the deserialized task exposes the $snapshots metadata schema,
not the base table's columns.
Assertions.assertNotNull(task.schema().findField("snapshot_id"),
"the serialized split must carry the metadata-table
($snapshots) schema");
@@ -4049,7 +4335,8 @@ public class IcebergScanPlanProviderTest {
private static String firstSysSplitResidual(List<ConnectorScanRange>
ranges) throws Exception {
Assertions.assertFalse(ranges.isEmpty(), "the metadata table must plan
at least one split");
FileScanTask task =
- SerializationUtil.deserializeFromBase64(((IcebergScanRange)
ranges.get(0)).getSerializedSplit());
+ IcebergSystemTableSerialization.deserializeFromBase64(
+ ((IcebergScanRange)
ranges.get(0)).getSerializedSplit());
return task.residual().toString();
}
@@ -4176,7 +4463,8 @@ public class IcebergScanPlanProviderTest {
long rows = 0;
for (ConnectorScanRange range : ranges) {
FileScanTask task =
-
SerializationUtil.deserializeFromBase64(((IcebergScanRange)
range).getSerializedSplit());
+ IcebergSystemTableSerialization.deserializeFromBase64(
+ ((IcebergScanRange) range).getSerializedSplit());
try (CloseableIterable<StructLike> closeable =
task.asDataTask().rows()) {
Iterator<StructLike> it = closeable.iterator();
while (it.hasNext()) {
diff --git a/fe/pom.xml b/fe/pom.xml
index 593653f1052..bcc40349bb2 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -246,7 +246,7 @@ under the License.
<module>fe-grpc</module>
</modules>
<properties>
-
<doris.hive.catalog.shade.version>3.1.1</doris.hive.catalog.shade.version>
+
<doris.hive.catalog.shade.version>3.1.3</doris.hive.catalog.shade.version>
<!-- iceberg 1.9.1 depends avro on 1.12 -->
<avro.version>1.12.1</avro.version>
<parquet.version>1.17.0</parquet.version>
@@ -361,7 +361,7 @@ under the License.
<!-- ATTN: avro version must be consistent with Iceberg version -->
<!-- Please modify iceberg.version and avro.version together,
you can find avro version info in iceberg mvn repository -->
- <iceberg.version>1.10.1</iceberg.version>
+ <iceberg.version>1.11.0</iceberg.version>
<!-- 0.56.1 has bug that "SplitMode" in query response may not be
set-->
<maxcompute.version>0.53.2-public</maxcompute.version>
<!-- FE-only POC: validate whether Arrow Java 19.0.0 can compile
before touching BE/C++ thirdparty. -->
diff --git
a/regression-test/data/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.out
b/regression-test/data/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.out
index 7546005e2f2..cf3ccac5bc8 100644
---
a/regression-test/data/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.out
+++
b/regression-test/data/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.out
@@ -329,4 +329,3 @@ phone text Yes true \N
5 Eve 91.3 [email protected] \N
6 Frank 88.7 [email protected] \N
7 Grace 93.2 [email protected] 555-0123
-
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_sys_table.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_sys_table.out
index 6a5298b356f..3692b06a99e 100644
--- a/regression-test/data/external_table_p0/iceberg/test_iceberg_sys_table.out
+++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_sys_table.out
@@ -478,6 +478,7 @@ deleted_data_files_count int Yes true \N
NONE
deleted_delete_files_count int Yes true \N NONE
existing_data_files_count int Yes true \N NONE
existing_delete_files_count int Yes true \N NONE
+key_metadata text Yes true \N NONE
length bigint Yes true \N NONE
partition_spec_id int Yes true \N NONE
partition_summaries array<struct<contains_null:boolean not
null,contains_nan:boolean not null,lower_bound:text,upper_bound:text>> Yes
true \N NONE
@@ -986,6 +987,7 @@ deleted_data_files_count int Yes true \N
NONE
deleted_delete_files_count int Yes true \N NONE
existing_data_files_count int Yes true \N NONE
existing_delete_files_count int Yes true \N NONE
+key_metadata text Yes true \N NONE
length bigint Yes true \N NONE
partition_spec_id int Yes true \N NONE
partition_summaries array<struct<contains_null:boolean not
null,contains_nan:boolean not null,lower_bound:text,upper_bound:text>> Yes
true \N NONE
diff --git
a/regression-test/suites/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.groovy
b/regression-test/suites/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.groovy
index b3f05f1d606..017f2bf498d 100644
---
a/regression-test/suites/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/iceberg_schema_change_ddl_with_branch.groovy
@@ -198,7 +198,7 @@ suite("iceberg_schema_change_ddl_with_branch",
"p0,external") {
// All branches expose the current table columns: id, name, grade, email,
phone.
- // Verify all branches have the latest columns
+ // Branches bind predicates against the current table schema even when
their head snapshot is historical.
qt_all_branches_have_grade """ SELECT id, grade FROM
${branch_table_name}@branch(branch1) WHERE grade > 0 ORDER BY id """
qt_all_branches_have_email """ SELECT id, email FROM
${branch_table_name}@branch(branch2) WHERE email IS NOT NULL ORDER BY id """
qt_all_branches_have_phone """ SELECT id, phone FROM
${branch_table_name}@branch(branch3) WHERE phone IS NOT NULL ORDER BY id """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
