This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new fd268c27f5c [improvement](fe) Upgrade iceberg to 1.12.0 (#69031)
fd268c27f5c is described below
commit fd268c27f5c7e42175a91bd20b4723291b0ebdc1
Author: Andrei Tserakhau <[email protected]>
AuthorDate: Thu Oct 15 11:18:34 2026 +0200
[improvement](fe) Upgrade iceberg to 1.12.0 (#69031)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Upgrade the Iceberg Java SDK from 1.11.0 to 1.12.0.
1.12.0. This change only upgrades the dependency and adapts Doris to it.
- Bump `iceberg.version` 1.11.0 -> 1.12.0, and `avro.version` 1.12.1 ->
1.12.2 to stay consistent with what iceberg 1.12.0 is built against (the
pom requires the two to move together).
- Bump the explicit caffeine pin in fe-connector-iceberg 2.9.3 -> 3.2.4.
Iceberg 1.12.0 moved to caffeine 3. The pin is a direct dependency, so
it wins Maven mediation; leaving it at 2.9.3 would package an older
caffeine into the iceberg plugin than iceberg-core is built against.
- Re-vendor `org.apache.iceberg.DeleteFileIndex` from 1.12.0 and
re-apply the Doris changes (public class/builder methods, schemasById
comment, equality-field-id sets). The vendored class shadows the SDK
class in the same package, and 1.12.0 changed it (an extra `specsById`
constructor argument and delete/data file partition-match validation),
so keeping the 1.11.0 copy would break the SDK's own callers at runtime.
Since Doris passes `specsById` to the builder, the cached scan planning
path now also rejects delete files whose partition spec or tuple does
not match the data file, the same as upstream iceberg.
- Pass `table.specs()` to `ManifestFiles.read` in
`IcebergManifestCache`. Iceberg 1.12.0 removed the deprecated
`read(ManifestFile, FileIO)` overload, which read the partition specs
from the manifest file's own metadata (only possible for Avro
manifests). This matches the delete manifest path next to it
(`readDeleteManifest(..., table.specs())`) and
`IcebergRemoveOrphanFilesAction`.
- Keep the AWS SDK at 2.29.52 and default iceberg-aws's HTTP client to
`urlconnection`. In 1.12.0, iceberg-aws's default `http-client.type`
(`apache`) builds an `Apache5HttpClient` from
`software.amazon.awssdk:apache5-client`, which first shipped in AWS SDK
2.32. Doris pins 2.29.52 on purpose (#58566: newer SDKs add default S3
integrity checksums that break S3-compatible services), so with the
default every S3FileIO, Glue and STS client iceberg builds failed with
`NoClassDefFoundError:
software/amazon/awssdk/http/apache5/Apache5HttpClient$Builder`.
`IcebergCatalogFactory.buildBaseCatalogProperties`, the base of every
catalog flavor, now sets `http-client.type=urlconnection` unless the
catalog sets it. S3FileIO keeps the setting when it is serialized, so
the BE iceberg metadata scanner, which rebuilds the FileIO from it, uses
the same client; `url-connection-client` is added to that scanner, which
only bundled `apache-client` and `netty-nio-client`. An explicit
`http-client.type=apache` is rejected at CREATE/ALTER CATALOG instead of
failing on the first S3 or Glue call.
`SchemaAwareDataTableScan` (the other vendored class) needs no change:
`DataTableScan`, `BaseTableScan`, `SnapshotScan` and `TableScanContext`
are identical between 1.11.0 and 1.12.0.
### Release note
Upgrade the Iceberg Java SDK to 1.12.0. Iceberg's AWS clients (S3FileIO,
Glue, STS) now use the URLConnection HTTP client by default, and the
`http-client.type=apache` catalog property is no longer accepted.
Iceberg 1.12.0 no longer reads the S3 remote-signing properties
`s3.signer.uri` and `s3.signer.endpoint` and no longer defaults the
signer endpoint to `v1/aws/s3/sign`, so an Iceberg REST server that
enables S3 remote signing must send `signer.uri` / `signer.endpoint` or
serve the table-scoped sign endpoint. Iceberg system tables
(`$snapshots`, `$files`, ...) cannot be queried while FE and BE are on
different Iceberg versions (1.11 and 1.12), e.g. during a rolling
upgrade; they work again once every node is upgraded.
### Check List (For Author)
- Test: Unit Test
- Added
IcebergCatalogFactoryTest#buildBaseDefaultsIcebergAwsHttpClientToUrlConnection
(builds a real S3 client builder with the default) and
#buildBaseKeepsAnExplicitHttpClientType, and
IcebergCatalogPropertiesTest#createTimeRulesRejectTheApacheHttpClientType;
IcebergCatalogOptionsSnapshotTest snapshots now include the default.
- `build.sh --fe` passes. Ran every iceberg FE unit test with
run-fe-ut.sh (fe-connector-iceberg, fe-connector-metastore-iceberg,
be-java-extensions/iceberg-metadata-scanner): 1630 tests, 0 failures, 5
skipped.
- Iceberg regression suites not run.
- Behavior changed: Yes.
- Iceberg's AWS clients use the URLConnection HTTP client by default, so
`http-client.apache.*` properties no longer apply, and
`http-client.type=apache` is rejected at CREATE/ALTER CATALOG.
- Iceberg scans whose delete files have a partition spec or partition
tuple that does not match the data file now fail with a
ValidationException (iceberg 1.12.0 behavior) instead of applying the
deletes.
- S3 remote signing: iceberg 1.12.0 no longer reads `s3.signer.uri` /
`s3.signer.endpoint` and no longer defaults the signer endpoint to
`v1/aws/s3/sign`. A REST catalog whose server turns on remote signing
but sends only those legacy keys, or relies on that default, fails to
read manifests. No compatibility shim is added: Doris does not use
remote signing to read data (BE uses static or vended credentials), and
`signer.uri` / `signer.endpoint` are deprecated in 1.12.0 and scheduled
for removal in 1.13.0.
- Iceberg system tables cannot be queried while FE and BE are on
different Iceberg versions (1.11 and 1.12), e.g. during a rolling
upgrade: iceberg 1.12.0 changed the serialized form of `BaseFile` and
`PartitionData`, which system table tasks carry. BE reports the version
mismatch instead of a bare `InvalidClassException`. Normal data scans
are not affected.
- Does this need documentation: No
Issue Number: related #68916
Co-authored-by: Isaac <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../iceberg-metadata-scanner/pom.xml | 6 ++
.../doris/iceberg/IcebergSerializationCompat.java | 11 ++-
.../iceberg/IcebergSerializationCompatTest.java | 83 ++++++++++++++++++++++
fe/fe-connector/fe-connector-iceberg/pom.xml | 4 +-
.../iceberg/IcebergCacheSizeEstimator.java | 5 +-
.../connector/iceberg/IcebergCatalogFactory.java | 26 +++++--
.../iceberg/IcebergCatalogProperties.java | 10 +++
.../doris/connector/iceberg/IcebergConnector.java | 18 +++--
.../connector/iceberg/IcebergManifestCache.java | 45 ++++++++----
.../connector/iceberg/IcebergManifestEntryKey.java | 24 +++++--
.../connector/iceberg/IcebergScanPlanProvider.java | 11 +--
.../iceberg/IcebergSystemTableSerialization.java | 11 ++-
.../java/org/apache/iceberg/DeleteFileIndex.java | 48 +++++++++++--
.../iceberg/IcebergCatalogFactoryTest.java | 41 +++++++++++
.../iceberg/IcebergCatalogOptionsSnapshotTest.java | 20 ++++++
.../iceberg/IcebergCatalogPropertiesTest.java | 19 +++++
.../iceberg/IcebergConnectorCacheTest.java | 2 +-
.../IcebergConnectorTestConnectionTest.java | 63 +++++++++++++++-
.../iceberg/IcebergManifestCacheTest.java | 43 ++++++++---
.../iceberg/IcebergManifestEntryKeyTest.java | 33 ++++++---
.../iceberg/IcebergScanPlanProviderTest.java | 54 ++++++++++++++
fe/pom.xml | 6 +-
22 files changed, 520 insertions(+), 63 deletions(-)
diff --git a/fe/be-java-extensions/iceberg-metadata-scanner/pom.xml
b/fe/be-java-extensions/iceberg-metadata-scanner/pom.xml
index b6d275327ae..8b63a2428d4 100644
--- a/fe/be-java-extensions/iceberg-metadata-scanner/pom.xml
+++ b/fe/be-java-extensions/iceberg-metadata-scanner/pom.xml
@@ -101,6 +101,12 @@ under the License.
<groupId>software.amazon.awssdk</groupId>
<artifactId>auth</artifactId>
</dependency>
+ <!-- The HTTP client the FE sets for iceberg-aws
(http-client.type=urlconnection); the serialized S3FileIO
+ carries that setting here. iceberg's default "apache" type needs
a newer AWS SDK than Doris ships. -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>url-connection-client</artifactId>
+ </dependency>
<!--
Hadoop: a task's FileIO is whatever the table was created with, and
for every non-S3
catalog that is HadoopFileIO (directly or behind ResolvingFileIO).
It used to be provided
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
index dda1c43c1b2..14ebf8c8cb2 100644
---
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
@@ -17,10 +17,12 @@
package org.apache.doris.iceberg;
+import org.apache.iceberg.IcebergBuild;
import org.apache.iceberg.Schema;
import java.io.ByteArrayInputStream;
import java.io.IOException;
+import java.io.InvalidClassException;
import java.io.ObjectInputStream;
import java.io.ObjectStreamClass;
import java.io.UncheckedIOException;
@@ -44,10 +46,15 @@ final class IcebergSerializationCompat {
try (ByteArrayInputStream input = new ByteArrayInputStream(bytes);
ObjectInputStream objectInput = new
SchemaCompatibleObjectInputStream(input)) {
return (T) objectInput.readObject();
+ } catch (InvalidClassException | ClassNotFoundException e) {
+ // A class in the task differs from, or is missing in, this BE's
Iceberg: the split was serialized by an
+ // FE on another Iceberg version. Only Schema is shimmed; Iceberg
1.12 changed BaseFile, PartitionData and
+ // their anonymous classes, which every task carries, so the task
cannot be read here.
+ throw new IllegalStateException("Cannot read an Iceberg system
table split serialized with a different"
+ + " Iceberg version than this BE's (" +
IcebergBuild.version() + "). Iceberg system tables are"
+ + " unavailable while FE and BE run different Doris
versions, e.g. during a rolling upgrade.", e);
} catch (IOException e) {
throw new UncheckedIOException("Failed to deserialize object", e);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException("Could not read object ", e);
}
}
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
index c23715d4adf..3d28b18146f 100644
---
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
@@ -17,7 +17,10 @@
package org.apache.doris.iceberg;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -25,10 +28,18 @@ import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InvalidClassException;
import java.io.ObjectOutputStream;
+import java.io.ObjectStreamConstants;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.Base64;
public class IcebergSerializationCompatTest {
+ // BaseFile's serialVersionUID in Iceberg 1.11.0; 1.12.0 changed it (and
added a field).
+ private static final long ICEBERG_1_11_0_BASE_FILE_UID =
-8856745894322453272L;
+
// 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"
@@ -66,6 +77,78 @@ public class IcebergSerializationCompatTest {
Assertions.assertTrue(expected.sameSchema(actual));
}
+ @Test
+ public void reportsATaskFromAnotherIcebergVersion() throws IOException {
+ // WHY: every system-table task carries a DataFile (StaticDataTask's
metadata file is built like this one),
+ // and an FE on Iceberg 1.11 writes BaseFile with its 1.11 UID.
MUTATION: dropping the InvalidClassException
+ // mapping -> a bare UncheckedIOException with no hint at the version
skew -> red.
+ byte[] bytes =
serialize(DataFiles.builder(PartitionSpec.unpartitioned())
+
.withPath("/dummyPath").withFileSizeInBytes(0L).withRecordCount(1L)
+ .withFormat(FileFormat.METADATA).build());
+ rewriteUid(bytes, "org.apache.iceberg.BaseFile",
ICEBERG_1_11_0_BASE_FILE_UID);
+
+ IllegalStateException e =
Assertions.assertThrows(IllegalStateException.class,
+ () ->
IcebergSerializationCompat.deserializeFromBase64(Base64.getEncoder().encodeToString(bytes)));
+ Assertions.assertTrue(e.getMessage().contains("different Iceberg
version"), e.getMessage());
+ Assertions.assertInstanceOf(InvalidClassException.class, e.getCause());
+ }
+
+ @Test
+ public void reportsAClassThisIcebergDoesNotHave() throws IOException {
+ // WHY: Iceberg 1.12 renumbered anonymous classes a task carries (e.g.
BaseFile$1), so a task from another
+ // version can name a class that does not exist here. MUTATION:
mapping only InvalidClassException -> red.
+ byte[] bytes = serialize(new Probe());
+ String name = Probe.class.getName();
+ replaceAll(bytes, name.getBytes(StandardCharsets.UTF_8),
+ (name.substring(0, name.length() - 1) +
"X").getBytes(StandardCharsets.UTF_8));
+
+ IllegalStateException e =
Assertions.assertThrows(IllegalStateException.class,
+ () ->
IcebergSerializationCompat.deserializeFromBase64(Base64.getEncoder().encodeToString(bytes)));
+ Assertions.assertInstanceOf(ClassNotFoundException.class,
e.getCause());
+ }
+
+ private static final class Probe implements Serializable {
+ private static final long serialVersionUID = 1L;
+ }
+
+ private static byte[] serialize(Object value) throws IOException {
+ ByteArrayOutputStream output = new ByteArrayOutputStream();
+ try (ObjectOutputStream objectOutput = new ObjectOutputStream(output))
{
+ objectOutput.writeObject(value);
+ }
+ return output.toByteArray();
+ }
+
+ /** Overwrites the serialVersionUID that follows the class descriptor
named {@code className}. */
+ private static void rewriteUid(byte[] bytes, String className, long uid) {
+ byte[] name = className.getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i + 3 + name.length + Long.BYTES <= bytes.length; i++)
{
+ if (bytes[i] == ObjectStreamConstants.TC_CLASSDESC && bytes[i + 1]
== (byte) (name.length >>> 8)
+ && bytes[i + 2] == (byte) name.length && matchesAt(bytes,
i + 3, name)) {
+ ByteBuffer.wrap(bytes, i + 3 + name.length,
Long.BYTES).putLong(uid);
+ return;
+ }
+ }
+ Assertions.fail("no class descriptor for " + className);
+ }
+
+ private static void replaceAll(byte[] bytes, byte[] from, byte[] to) {
+ for (int i = 0; i + from.length <= bytes.length; i++) {
+ if (matchesAt(bytes, i, from)) {
+ System.arraycopy(to, 0, bytes, i, to.length);
+ }
+ }
+ }
+
+ private static boolean matchesAt(byte[] bytes, int offset, byte[] pattern)
{
+ for (int j = 0; j < pattern.length; j++) {
+ if (bytes[offset + j] != pattern[j]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
private static String serializeToBase64(Object value) throws IOException {
ByteArrayOutputStream output = new ByteArrayOutputStream();
try (ObjectOutputStream objectOutput = new ObjectOutputStream(output))
{
diff --git a/fe/fe-connector/fe-connector-iceberg/pom.xml
b/fe/fe-connector/fe-connector-iceberg/pom.xml
index 856594ff29f..282499f2988 100644
--- a/fe/fe-connector/fe-connector-iceberg/pom.xml
+++ b/fe/fe-connector/fe-connector-iceberg/pom.xml
@@ -115,13 +115,13 @@ under the License.
<!-- Caffeine: needed on the COMPILE classpath for the vendored
split-package
org.apache.iceberg.DeleteFileIndex (T08). iceberg-core 1.10.1
made DeleteFileIndex package-private,
so the connector ships its own public copy (mirrors fe-core) to
drive manifest-level scan planning;
- that copy imports com.github.benmanes.caffeine. iceberg-core
already brings caffeine 2.9.3 at runtime
+ that copy imports com.github.benmanes.caffeine. iceberg-core
already brings caffeine 3.2.4 at runtime
scope (so it is in the plugin-zip), but runtime-scoped
transitives are not on the compile classpath
— hence this explicit compile-scope declaration (same version,
deduped). -->
<dependency>
<groupId>com.github.ben-manes.caffeine</groupId>
<artifactId>caffeine</artifactId>
- <version>2.9.3</version>
+ <version>3.2.4</version>
</dependency>
<!-- iceberg-bundled-guava: needed on the COMPILE classpath for the
vendored split-package
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
index fdffb7dc184..bbd05df2982 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCacheSizeEstimator.java
@@ -246,7 +246,10 @@ final class IcebergCacheSizeEstimator {
}
static long estimateManifestKey(IcebergManifestEntryKey key) {
- return add(MANIFEST_KEY_SHALLOW_BYTES,
JvmSizeUtils.stringSize(key.getManifestPath()));
+ long bytes = add(MANIFEST_KEY_SHALLOW_BYTES,
JvmSizeUtils.stringSize(key.getManifestPath()));
+ // The key retains its bound partition type. A non-empty value's files
share it, but an empty value (a delete
+ // manifest whose entries are all DELETED) holds nothing that would
account for it.
+ return add(bytes, estimateIcebergType(key.getPartitionType()));
}
static long estimateManifestValue(ManifestCacheValue value) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
index 6374647c60f..9620d6fec20 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java
@@ -34,6 +34,7 @@ import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.aws.AssumeRoleAwsClientFactory;
import org.apache.iceberg.aws.AwsClientProperties;
import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.HttpClientProperties;
import org.apache.iceberg.aws.s3.S3FileIOProperties;
import org.apache.iceberg.rest.auth.OAuth2Properties;
@@ -141,10 +142,11 @@ public final class IcebergCatalogFactory {
* mirroring the legacy {@code
AbstractIcebergProperties.initializeCatalog} base: (1) seed from ALL
* raw props (legacy {@code getOrigProps()} copy-all — arbitrary
user/iceberg keys pass through to
* the SDK; the per-flavor appenders + the connector add the derived keys
on top), (2) map
- * {@code warehouse} to {@link CatalogProperties#WAREHOUSE_LOCATION}, (3)
add manifest-cache keys.
- * PURE: depends only on its arguments; {@code metaCacheWeightLimited} is
- * {@code IcebergConnector#hasMetaCacheWeightLimit}. The flavor's {@code
catalog-impl} and the {@code type}
- * removal are applied by the caller (the connector / per-flavor path).
+ * {@code warehouse} to {@link CatalogProperties#WAREHOUSE_LOCATION}, (3)
add manifest-cache keys,
+ * (4) default iceberg-aws's HTTP client to {@code urlconnection}. PURE:
depends only on its arguments;
+ * {@code metaCacheWeightLimited} is {@code
IcebergConnector#hasMetaCacheWeightLimit}. The flavor's
+ * {@code catalog-impl} and the {@code type} removal are applied by the
caller (the connector / per-flavor
+ * path).
*/
public static Map<String, String> buildBaseCatalogProperties(Map<String,
String> props,
boolean metaCacheWeightLimited) {
@@ -154,9 +156,25 @@ public final class IcebergCatalogFactory {
opts.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
}
appendManifestCacheProperties(props, opts, metaCacheWeightLimited);
+ applyDefaultAwsHttpClient(opts);
return opts;
}
+ /**
+ * Defaults iceberg-aws's HTTP client to {@code urlconnection} unless
{@code opts} sets one. A blank value counts as
+ * unset: iceberg-aws turns an empty type back into its own "apache"
default. iceberg-aws's default
+ * "apache" client type is Apache5HttpClient since iceberg 1.12, which
needs an AWS SDK newer than the one Doris
+ * pins (2.29.x, see #58566), so every S3FileIO / Glue / STS client
iceberg builds from a map without this key fails
+ * with NoClassDefFoundError; url-connection-client ships with that SDK.
Every map that ends up building an
+ * iceberg AWS client must go through here. The key travels with
S3FileIO's properties, so the BE metadata
+ * scanner, which rebuilds the FileIO from them, uses the same client.
+ */
+ static void applyDefaultAwsHttpClient(Map<String, String> opts) {
+ if (StringUtils.isBlank(opts.get(HttpClientProperties.CLIENT_TYPE))) {
+ opts.put(HttpClientProperties.CLIENT_TYPE,
HttpClientProperties.CLIENT_TYPE_URLCONNECTION);
+ }
+ }
+
/**
* Mirrors legacy {@code
AbstractIcebergProperties.addManifestCacheProperties}: pass through any
* explicitly-set {@code io.manifest.cache.*} keys, then — only when the
user did NOT set
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
index fe9844a82c3..b0981af2789 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogProperties.java
@@ -22,6 +22,8 @@ import
org.apache.doris.connector.metastore.spi.MetaStoreProviders;
import org.apache.doris.foundation.property.ConnectorPropertiesUtils;
import org.apache.doris.foundation.property.ConnectorProperty;
+import org.apache.iceberg.aws.HttpClientProperties;
+
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Locale;
@@ -144,6 +146,14 @@ public final class IcebergCatalogProperties {
if (TYPE_DLF.equals(flavor) && externalCatalogName != null) {
throw new IllegalArgumentException("external_catalog.name is not
supported for Iceberg DLF catalogs");
}
+ // iceberg-aws's "apache" client is Apache5HttpClient, which the AWS
SDK Doris ships does not contain;
+ // fail here rather than with a NoClassDefFoundError on the first S3
or Glue call.
+ if
(HttpClientProperties.CLIENT_TYPE_APACHE.equalsIgnoreCase(raw.get(HttpClientProperties.CLIENT_TYPE)))
{
+ throw new
IllegalArgumentException(HttpClientProperties.CLIENT_TYPE + "="
+ + HttpClientProperties.CLIENT_TYPE_APACHE + " is not
supported: it requires an AWS SDK newer than"
+ + " the one Doris ships. Use " +
HttpClientProperties.CLIENT_TYPE_URLCONNECTION
+ + " (the default) instead.");
+ }
// Selects the backend by flavor and enforces its fail-fast rules --
REST (security/creds enums,
// OAuth2, signing, AK/SK), Glue (AK/SK-together, endpoint https,
at-least-one-credential), JDBC
// (uri/catalog_name/warehouse), the shared HMS connection checks;
hadoop/s3tables are no-op (their
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
index 8e910135238..afc79d39544 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java
@@ -173,6 +173,8 @@ public class IcebergConnector implements Connector {
/** The key BE looks up (case-sensitively) to learn which location to
probe. */
private static final String BE_TEST_LOCATION = "test_location";
+ /** Prefix of iceberg-aws's HTTP client settings ({@code
http-client.type}, proxy, timeouts). */
+ private static final String HTTP_CLIENT_PROPERTY_PREFIX = "http-client.";
private final Map<String, String> properties;
// The bound connector-level facts (flavor, type-mapping switches,
namespace level), the single
@@ -469,12 +471,12 @@ public class IcebergConnector implements Connector {
// Map Doris s3.* keys to Iceberg S3FileIO keys and force static
credentials (disable
// remote/vended signing) so the probe validates exactly what the user
configured.
- Map<String, String> ioProps = buildStorageProbeProperties(catalogType,
chosenStorage.get());
+ Map<String, String> ioProps = buildStorageProbeProperties(catalogType,
chosenStorage.get(), properties);
// Load S3FileIO reflectively via CatalogUtil so this module needs no
compile-time AWS SDK
// dependency; the AWS SDK is resolved from the shared runtime
classpath at execution time.
// Pin the TCCL to the plugin loader for the probe: iceberg-aws builds
its S3 client lazily and
- // resolves org.apache.iceberg.aws.ApacheHttpClientConfigurations via
DynMethods (default loader =
+ // resolves its org.apache.iceberg.aws.*HttpClientConfigurations class
via DynMethods (default loader =
// TCCL). This CREATE-CATALOG thread runs under the default 'app'
TCCL, which would return the
// fe-core copy of the class and ClassCast against the child-loaded
plugin copy the rest of the
// iceberg-aws stack uses — the SAME split-brain
TcclPinningConnectorContext guards on the commit
@@ -510,8 +512,8 @@ public class IcebergConnector implements Connector {
return probeStorageFromBackend(location);
}
- static Map<String, String> buildStorageProbeProperties(
- String catalogType, S3CompatibleFileSystemProperties
storageProperties) {
+ static Map<String, String> buildStorageProbeProperties(String catalogType,
+ S3CompatibleFileSystemProperties storageProperties, Map<String,
String> catalogProperties) {
String endpoint = storageProperties.getEndpoint();
if (IcebergCatalogProperties.TYPE_DLF.equalsIgnoreCase(catalogType)
&& "OSS".equals(storageProperties.providerName())) {
@@ -530,6 +532,14 @@ public class IcebergConnector implements Connector {
if (!isBlank(storageProperties.getRegion())) {
ioProps.put("client.region", storageProperties.getRegion());
}
+ // The probe builds its own S3FileIO, so it must reach the store the
way the catalog does: same HTTP client
+ // and proxy settings, and the same default client type as every other
iceberg AWS client.
+ for (Map.Entry<String, String> e : catalogProperties.entrySet()) {
+ if (e.getKey().startsWith(HTTP_CLIENT_PROPERTY_PREFIX)) {
+ ioProps.put(e.getKey(), e.getValue());
+ }
+ }
+ IcebergCatalogFactory.applyDefaultAwsHttpClient(ioProps);
return ioProps;
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
index 96084f19db2..ca54be70761 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java
@@ -24,12 +24,14 @@ import org.apache.doris.connector.cache.MetaCacheDefinition;
import org.apache.doris.connector.cache.MetaCacheSizeEstimators;
import org.apache.doris.connector.cache.ScopePath;
+import com.google.common.base.Preconditions;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.ManifestContent;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Table;
import java.io.IOException;
@@ -200,20 +202,22 @@ final class IcebergManifestCache {
* loader runs OUTSIDE Caffeine's compute lock (manual miss-load;
single-flight per key), so a same-key
* miss parses at most once.
*/
- ManifestCacheValue getManifestCacheValue(ManifestFile manifest, Table
table) {
- IcebergManifestEntryKey key = IcebergManifestEntryKey.of(manifest);
+ ManifestCacheValue getManifestCacheValue(ManifestFile manifest, Table
table,
+ Map<Integer, PartitionSpec> specsById) {
+ IcebergManifestEntryKey key = keyOf(manifest, specsById);
return entry.get(key, k -> loadManifestCacheValue(
- manifest, table, k.getContent(), entry.isWeightBounded()));
+ manifest, table, specsById, k.getContent(),
entry.isWeightBounded()));
}
/**
- * Same as {@link #getManifestCacheValue(ManifestFile, Table)} but tallies
this access as a hit or miss under
+ * Same as {@link #getManifestCacheValue(ManifestFile, Table, Map)} but
tallies this access as a hit or miss under
* {@code queryId} (a hit == the entry was already cached BEFORE this
load, matching the legacy
* {@code getIfPresent(key) != null} probe). Within one statement the
manifest-level plan processes manifests
* sequentially, so the per-query counters need no synchronization. A
blank queryId is not recorded.
*/
- ManifestCacheValue getManifestCacheValue(ManifestFile manifest, Table
table, String queryId) {
- IcebergManifestEntryKey key = IcebergManifestEntryKey.of(manifest);
+ ManifestCacheValue getManifestCacheValue(ManifestFile manifest, Table
table,
+ Map<Integer, PartitionSpec> specsById, String queryId) {
+ IcebergManifestEntryKey key = keyOf(manifest, specsById);
boolean hit = entry.getIfPresent(key) != null;
if (queryId != null && !queryId.isEmpty()) {
ScanStats stats = touch(queryId);
@@ -224,7 +228,7 @@ final class IcebergManifestCache {
}
}
return entry.get(key, k -> loadManifestCacheValue(
- manifest, table, k.getContent(), entry.isWeightBounded()));
+ manifest, table, specsById, k.getContent(),
entry.isWeightBounded()));
}
/**
@@ -270,22 +274,34 @@ final class IcebergManifestCache {
statsByQuery.entrySet().removeIf(e -> nowNanos -
e.getValue().lastTouchNanos >= statsTtlNanos);
}
+ /**
+ * {@code specsById} is the scan's spec map (bound to the scan's schema),
the same one its residuals and delete
+ * index use: the reader converts partition values to these types, so they
must match what the scan evaluates.
+ */
+ private static IcebergManifestEntryKey keyOf(ManifestFile manifest,
Map<Integer, PartitionSpec> specsById) {
+ PartitionSpec boundSpec = specsById.get(manifest.partitionSpecId());
+ Preconditions.checkArgument(boundSpec != null, "No bound partition
spec %s for iceberg manifest %s",
+ manifest.partitionSpecId(), manifest.path());
+ return IcebergManifestEntryKey.of(manifest, boundSpec);
+ }
+
private static ManifestCacheValue loadManifestCacheValue(ManifestFile
manifest, Table table,
- ManifestContent content, boolean estimateWeight) {
+ Map<Integer, PartitionSpec> specsById, ManifestContent content,
boolean estimateWeight) {
try {
if (content == ManifestContent.DELETES) {
- return
ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table),
estimateWeight);
+ return
ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table, specsById),
estimateWeight);
}
- return ManifestCacheValue.forDataFiles(loadDataFiles(manifest,
table), estimateWeight);
+ return ManifestCacheValue.forDataFiles(loadDataFiles(manifest,
table, specsById), estimateWeight);
} catch (IOException e) {
throw new RuntimeException("Failed to read iceberg manifest " +
manifest.path(), e);
}
}
- private static List<DataFile> loadDataFiles(ManifestFile manifest, Table
table) throws IOException {
+ private static List<DataFile> loadDataFiles(ManifestFile manifest, Table
table,
+ Map<Integer, PartitionSpec> specsById) throws IOException {
List<DataFile> dataFiles = new ArrayList<>();
// .copy() is mandatory: the ManifestReader iterator reuses the same
object across iterations.
- try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest,
table.io())) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest,
table.io(), specsById)) {
for (DataFile dataFile : reader) {
dataFiles.add(dataFile.copy());
}
@@ -293,10 +309,11 @@ final class IcebergManifestCache {
return dataFiles;
}
- private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest,
Table table) throws IOException {
+ private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest,
Table table,
+ Map<Integer, PartitionSpec> specsById) throws IOException {
List<DeleteFile> deleteFiles = new ArrayList<>();
try (ManifestReader<DeleteFile> reader =
- ManifestFiles.readDeleteManifest(manifest, table.io(),
table.specs())) {
+ ManifestFiles.readDeleteManifest(manifest, table.io(),
specsById)) {
for (DeleteFile deleteFile : reader) {
deleteFiles.add(deleteFile.copy());
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
index 90552bef6d5..9e40896d52b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKey.java
@@ -19,6 +19,8 @@ package org.apache.doris.connector.iceberg;
import org.apache.iceberg.ManifestContent;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.types.Types;
import java.util.Objects;
@@ -31,18 +33,25 @@ import java.util.Objects;
* cached payload, and a table-level invalidation (REFRESH TABLE)
intentionally does NOT drop it (legacy
* {@code testInvalidateTableKeepsManifestCache} parity). Runtime loader
context (the manifest/table instances)
* must not be stored here.
+ *
+ * <p>The key also carries the partition type the manifest is decoded with:
the reader converts partition values to
+ * the bound spec's types, and after a type promotion (e.g. INT to BIGINT) a
scan pinned to the old schema and a scan
+ * on the current one bind the same manifest to different types. Each binding
needs its own decoded copy.
*/
public class IcebergManifestEntryKey {
private final String manifestPath;
private final ManifestContent content;
+ private final Types.StructType partitionType;
- public IcebergManifestEntryKey(String manifestPath, ManifestContent
content) {
+ public IcebergManifestEntryKey(String manifestPath, ManifestContent
content, Types.StructType partitionType) {
this.manifestPath = Objects.requireNonNull(manifestPath, "manifestPath
can not be null");
this.content = Objects.requireNonNull(content, "content can not be
null");
+ this.partitionType = Objects.requireNonNull(partitionType,
"partitionType can not be null");
}
- public static IcebergManifestEntryKey of(ManifestFile manifest) {
- return new IcebergManifestEntryKey(manifest.path(),
manifest.content());
+ /** The key for {@code manifest} decoded with {@code boundSpec}, the
scan's spec for its partition spec id. */
+ public static IcebergManifestEntryKey of(ManifestFile manifest,
PartitionSpec boundSpec) {
+ return new IcebergManifestEntryKey(manifest.path(),
manifest.content(), boundSpec.partitionType());
}
public String getManifestPath() {
@@ -53,6 +62,10 @@ public class IcebergManifestEntryKey {
return content;
}
+ public Types.StructType getPartitionType() {
+ return partitionType;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -63,11 +76,12 @@ public class IcebergManifestEntryKey {
}
IcebergManifestEntryKey that = (IcebergManifestEntryKey) o;
return Objects.equals(manifestPath, that.manifestPath)
- && content == that.content;
+ && content == that.content
+ && partitionType.equals(that.partitionType);
}
@Override
public int hashCode() {
- return Objects.hash(manifestPath, content);
+ return Objects.hash(manifestPath, content, partitionType);
}
}
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 b612be4f42d..73beceb47ed 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
@@ -2838,7 +2838,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (!ManifestEvaluator.forPartitionFilter(filterExpr, spec,
caseSensitive).eval(manifest)) {
continue;
}
- deleteFiles.addAll(manifestCacheGet(manifest, table,
statsQueryId).getDeleteFiles());
+ deleteFiles.addAll(manifestCacheGet(manifest, table, specsById,
statsQueryId).getDeleteFiles());
}
DeleteFileIndex deleteIndex = DeleteFileIndex.builderFor(deleteFiles)
.schemasById(table.schemas())
@@ -2864,10 +2864,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
/** Dispatch a manifest read to the stats-tallying or no-stats cache
overload (see cacheBackedFileScanTasks). */
- private ManifestCacheValue manifestCacheGet(ManifestFile manifest, Table
table, String statsQueryId) {
+ private ManifestCacheValue manifestCacheGet(ManifestFile manifest, Table
table,
+ Map<Integer, PartitionSpec> specsById, String statsQueryId) {
return statsQueryId != null
- ? manifestCache.getManifestCacheValue(manifest, table,
statsQueryId)
- : manifestCache.getManifestCacheValue(manifest, table);
+ ? manifestCache.getManifestCacheValue(manifest, table,
specsById, statsQueryId)
+ : manifestCache.getManifestCacheValue(manifest, table,
specsById);
}
/**
@@ -2956,7 +2957,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
currentResidual = residual;
currentSpecJson = PartitionSpecParser.toJson(spec);
- dataFileIt = manifestCacheGet(manifest, table,
statsQueryId).getDataFiles().iterator();
+ dataFileIt = manifestCacheGet(manifest, table, specsById,
statsQueryId).getDataFiles().iterator();
}
}
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
index 97c702a4901..a66abd229af 100644
---
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
@@ -26,7 +26,14 @@ 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. */
+/**
+ * Keeps the Java-serialized Iceberg system-table task wire compatible with
Iceberg 1.10.1.
+ *
+ * <p>That only covers Schema, the one class whose UID 1.11 changed. Iceberg
1.12 changed the serialized form of
+ * classes every task carries (BaseFile, PartitionData and their anonymous
inner classes), so a task cannot cross
+ * between 1.11 and 1.12: Iceberg system tables are unavailable while FE and
BE run different Iceberg versions,
+ * and the BE reader reports that instead of a bare InvalidClassException.
+ */
final class IcebergSystemTableSerialization {
static final long ICEBERG_1_10_1_SCHEMA_UID = 6812231194765760118L;
private static final long ICEBERG_1_11_0_SCHEMA_UID =
-1265875184407129845L;
@@ -56,7 +63,7 @@ final class IcebergSystemTableSerialization {
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.
+ // to the legacy UID lets 1.10 and 1.11 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);
}
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 4c45cfc1b3e..bddcb5a237d 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
@@ -67,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.11.0/core/src/main/java/org/apache/iceberg/DeleteFileIndex.java
+ * Copied from
https://github.com/apache/iceberg/blob/apache-iceberg-1.12.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)
@@ -88,6 +88,7 @@ public class DeleteFileIndex {
private final PartitionMap<PositionDeletes> posDeletesByPartition;
private final Map<String, PositionDeletes> posDeletesByPath;
private final Map<String, DeleteFile> dvByPath;
+ private final LoadingCache<Integer, Comparator<StructLike>>
partitionComparatorsBySpecId;
private final boolean hasEqDeletes;
private final boolean hasPosDeletes;
private final boolean isEmpty;
@@ -97,12 +98,18 @@ public class DeleteFileIndex {
PartitionMap<EqualityDeletes> eqDeletesByPartition,
PartitionMap<PositionDeletes> posDeletesByPartition,
Map<String, PositionDeletes> posDeletesByPath,
- Map<String, DeleteFile> dvByPath) {
+ Map<String, DeleteFile> dvByPath,
+ Map<Integer, PartitionSpec> specsById) {
this.globalDeletes = globalDeletes;
this.eqDeletesByPartition = eqDeletesByPartition;
this.posDeletesByPartition = posDeletesByPartition;
this.posDeletesByPath = posDeletesByPath;
this.dvByPath = dvByPath;
+ this.partitionComparatorsBySpecId =
+ specsById == null
+ ? null
+ : Caffeine.newBuilder()
+ .build(specId ->
Comparators.forType(specsById.get(specId).partitionType()));
this.hasEqDeletes = globalDeletes != null || eqDeletesByPartition != null;
this.hasPosDeletes =
posDeletesByPartition != null || posDeletesByPath != null || dvByPath
!= null;
@@ -209,7 +216,15 @@ public class DeleteFileIndex {
}
PositionDeletes deletes = posDeletesByPath.get(dataFile.location());
- return deletes == null ? EMPTY_DELETES : deletes.filter(seq);
+ if (deletes == null) {
+ return EMPTY_DELETES;
+ }
+
+ DeleteFile[] matchingDeletes = deletes.filter(seq);
+ for (DeleteFile deleteFile : matchingDeletes) {
+ validatePartitionMatch(deleteFile, dataFile);
+ }
+ return matchingDeletes;
}
private DeleteFile findDV(long seq, DataFile dataFile) {
@@ -224,10 +239,34 @@ public class DeleteFileIndex {
"DV data sequence number (%s) must be greater than or equal to data
file sequence number (%s)",
dv.dataSequenceNumber(),
seq);
+ validatePartitionMatch(dv, dataFile);
}
return dv;
}
+ private void validatePartitionMatch(DeleteFile deleteFile, DataFile
dataFile) {
+ ValidationException.check(
+ deleteFile.specId() == dataFile.specId(),
+ "Mismatched partition specs (%s, %s) for delete file %s and data file
%s:"
+ + " metadata is corrupted",
+ deleteFile.specId(),
+ dataFile.specId(),
+ deleteFile.location(),
+ dataFile.location());
+ if (partitionComparatorsBySpecId != null) {
+ Comparator<StructLike> partitionComparator =
+ partitionComparatorsBySpecId.get(deleteFile.specId());
+ ValidationException.check(
+ partitionComparator.compare(deleteFile.partition(),
dataFile.partition()) == 0,
+ "Mismatched partition tuples (%s, %s) for delete file %s and data
file %s:"
+ + " metadata is corrupted",
+ deleteFile.partition(),
+ dataFile.partition(),
+ deleteFile.location(),
+ dataFile.location());
+ }
+ }
+
@SuppressWarnings("checkstyle:CyclomaticComplexity")
private static boolean canContainEqDeletesForFile(
DataFile dataFile, EqualityDeleteFile deleteFile) {
@@ -540,7 +579,8 @@ public class DeleteFileIndex {
eqDeletesByPartition.isEmpty() ? null : eqDeletesByPartition,
posDeletesByPartition.isEmpty() ? null : posDeletesByPartition,
posDeletesByPath.isEmpty() ? null : posDeletesByPath,
- dvByPath.isEmpty() ? null : dvByPath);
+ dvByPath.isEmpty() ? null : dvByPath,
+ specsById.isEmpty() ? null : specsById);
}
private void add(Map<String, DeleteFile> dvByPath, DeleteFile dv) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
index 6a3f31dbb1f..3fc0f6c3d99 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogFactoryTest.java
@@ -27,6 +27,7 @@ import
org.apache.doris.filesystem.properties.StorageProperties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.HttpClientProperties;
import org.apache.iceberg.rest.responses.ConfigResponse;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -37,6 +38,7 @@ import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
import
software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider;
import
software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider;
+import software.amazon.awssdk.services.s3.S3Client;
import java.util.Arrays;
import java.util.Collections;
@@ -215,6 +217,45 @@ public class IcebergCatalogFactoryTest {
Assertions.assertEquals("false",
opts.get("io.manifest.cache-enabled"));
}
+ @Test
+ public void buildBaseDefaultsIcebergAwsHttpClientToUrlConnection() {
+ // WHY: iceberg 1.12's default "apache" client type is
Apache5HttpClient, absent from the AWS SDK Doris
+ // pins, so without this default every S3FileIO / Glue / STS client
fails with NoClassDefFoundError.
+ // Building a client here runs on the plugin's own SDK, so it fails
exactly as production would.
+ // MUTATION: dropping the default -> "apache" -> NoClassDefFoundError
-> red.
+ Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
+ props("iceberg.catalog.type", "rest"), false);
+ Assertions.assertEquals(HttpClientProperties.CLIENT_TYPE_URLCONNECTION,
+ opts.get(HttpClientProperties.CLIENT_TYPE));
+ Assertions.assertDoesNotThrow(
+ () -> new
HttpClientProperties(opts).applyHttpClientConfigurations(S3Client.builder()));
+ }
+
+ @Test
+ public void buildBaseTreatsABlankHttpClientTypeAsUnset() {
+ // WHY: iceberg-aws maps an empty http-client.type back to its
"apache" default (Apache5HttpClient), so an
+ // empty value set at CREATE or by ALTER must get the supported
default too. MUTATION: putIfAbsent keeps "" ->
+ // NoClassDefFoundError on the client build -> red.
+ for (String blank : new String[] {"", " "}) {
+ Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
+ props("iceberg.catalog.type", "rest",
HttpClientProperties.CLIENT_TYPE, blank), false);
+
Assertions.assertEquals(HttpClientProperties.CLIENT_TYPE_URLCONNECTION,
+ opts.get(HttpClientProperties.CLIENT_TYPE));
+ Assertions.assertDoesNotThrow(
+ () -> new
HttpClientProperties(opts).applyHttpClientConfigurations(S3Client.builder()));
+ }
+ }
+
+ @Test
+ public void buildBaseKeepsAnExplicitHttpClientType() {
+ // WHY: the default only fills a missing key; a user's choice is
passed through (and "apache" is rejected
+ // at CREATE/ALTER by IcebergCatalogProperties, not silently rewritten
here). MUTATION: put instead of
+ // putIfAbsent -> the user's value is overwritten -> red.
+ Map<String, String> opts =
IcebergCatalogFactory.buildBaseCatalogProperties(
+ props("iceberg.catalog.type", "rest",
HttpClientProperties.CLIENT_TYPE, "apache"), false);
+ Assertions.assertEquals("apache",
opts.get(HttpClientProperties.CLIENT_TYPE));
+ }
+
@Test
public void buildBaseMetaCacheEnabledButZeroTtlDoesNotEnable() {
// WHY: legacy isCacheEnabled = enable && ttl != 0 && capacity != 0;
an explicit ttl-second=0
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
index cd3b22b0d4f..6c24483f98f 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogOptionsSnapshotTest.java
@@ -87,6 +87,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restMinimalSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest.example/api";,
"warehouse", "s3://bucket/wh",
"catalog-impl", "org.apache.iceberg.rest.RESTCatalog",
@@ -106,6 +107,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restOAuth2CredentialFlowSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"iceberg.rest.uri", "https://rest/api";,
"uri", "https://rest/api";,
"iceberg.rest.prefix", "main",
@@ -148,6 +150,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restOAuth2TokenFlowSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest/api";,
"iceberg.rest.security.type", "oauth2",
"iceberg.rest.oauth2.token", "tkn-123",
@@ -174,6 +177,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restTrimsSurroundingWhitespaceSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"iceberg.rest.uri", " https://rest/api ",
"iceberg.rest.security.type", "oauth2",
"iceberg.rest.oauth2.token", " tkn-123 ",
@@ -196,6 +200,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restSigningGlueWithBoundS3Snapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest/api";,
"iceberg.rest.signing-name", "glue",
"iceberg.rest.signing-region", "us-east-1",
@@ -233,6 +238,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restSigningGlueAssumeRoleSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest/api";,
"iceberg.rest.signing-name", "glue",
"iceberg.rest.signing-region", "us-east-1",
@@ -266,6 +272,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restSigningOtherNameWithRestCredentialsSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest/api";,
"iceberg.rest.signing-name", "custom",
"iceberg.rest.signing-region", "cn-north-1",
@@ -301,6 +308,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void restSigningProviderChainSnapshot() {
assertOptions(
props("iceberg.catalog.type", "rest",
+ "http-client.type", "urlconnection",
"uri", "https://rest/api";,
"iceberg.rest.signing-name", "custom",
"iceberg.rest.credentials_provider_type",
"instance-profile",
@@ -331,6 +339,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void glueStaticCredentialsSnapshot() {
assertOptions(
props("iceberg.catalog.type", "glue",
+ "http-client.type", "urlconnection",
"warehouse", "s3://bucket/wh",
"glue.endpoint",
"https://glue.us-east-1.amazonaws.com";,
"glue.access_key", "GAK",
@@ -366,6 +375,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void glueAssumeRoleWithoutWarehouseSnapshot() {
assertOptions(
props("iceberg.catalog.type", "glue",
+ "http-client.type", "urlconnection",
"glue.endpoint",
"https://glue.eu-west-1.amazonaws.com";,
"glue.role_arn", "arn:aws:iam::1:role/r",
"glue.external_id", "eid",
@@ -388,6 +398,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void glueExplicitRegionAliasSnapshot() {
assertOptions(
props("iceberg.catalog.type", "glue",
+ "http-client.type", "urlconnection",
"warehouse", "s3://bucket/wh",
"glue.endpoint",
"https://glue.us-east-1.amazonaws.com";,
"aws.glue.region", "ap-southeast-1",
@@ -415,6 +426,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void glueRegionFallsBackToUsEast1Snapshot() {
assertOptions(
props("iceberg.catalog.type", "glue",
+ "http-client.type", "urlconnection",
"warehouse", "s3://bucket/wh",
"glue.endpoint",
"https://vpce-1234.glue.privatelink.aws";,
"glue.role_arn", "arn:aws:iam::1:role/r",
@@ -443,6 +455,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void jdbcWithoutUriEmitsEmptyUriSnapshot() {
assertOptions(
props("iceberg.catalog.type", "jdbc",
+ "http-client.type", "urlconnection",
"warehouse", "s3://bucket/wh",
"uri", "",
"catalog-impl", "org.apache.iceberg.jdbc.JdbcCatalog"),
@@ -460,6 +473,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void jdbcFullSnapshot() {
assertOptions(
props("iceberg.catalog.type", "jdbc",
+ "http-client.type", "urlconnection",
"uri", "jdbc:postgresql://h/db",
"warehouse", "s3://bucket/wh",
"iceberg.jdbc.user", "u",
@@ -495,6 +509,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void jdbcUriAliasOrderSnapshot() {
assertOptions(
props("iceberg.catalog.type", "jdbc",
+ "http-client.type", "urlconnection",
"uri", "jdbc:postgresql://plain/db",
"iceberg.jdbc.uri", "jdbc:postgresql://prefixed/db",
"warehouse", "s3://bucket/wh",
@@ -519,6 +534,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void hmsSnapshot() {
assertOptions(
props("iceberg.catalog.type", "hms",
+ "http-client.type", "urlconnection",
"hive.metastore.uris", "thrift://h:9083",
"warehouse", "s3://bucket/wh",
"catalog-impl", DorisHiveCatalog.class.getName()),
@@ -536,6 +552,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void hadoopWithNonGenericStoreSnapshot() {
assertOptions(
props("iceberg.catalog.type", "hadoop",
+ "http-client.type", "urlconnection",
"warehouse", "oss://bucket/wh",
"catalog-impl",
"org.apache.iceberg.hadoop.HadoopCatalog",
"s3.endpoint", "https://oss";,
@@ -556,6 +573,7 @@ public class IcebergCatalogOptionsSnapshotTest {
public void manifestCacheDerivationSnapshot() {
assertOptions(
props("iceberg.catalog.type", "hadoop",
+ "http-client.type", "urlconnection",
"warehouse", "s3://bucket/wh",
"meta.cache.iceberg.manifest.enable", "true",
"meta.cache.iceberg.manifest.ttl-second", "3600",
@@ -576,6 +594,7 @@ public class IcebergCatalogOptionsSnapshotTest {
@Test
public void s3TablesWithStaticCredentialsSnapshot() {
Map<String, String> expected = props("iceberg.catalog.type",
"s3tables",
+ "http-client.type", "urlconnection",
"warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b",
"s3.role_arn", "arn:aws:iam::1:role/ignored",
"client.region", "us-east-1",
@@ -594,6 +613,7 @@ public class IcebergCatalogOptionsSnapshotTest {
@Test
public void s3TablesWithoutBoundStorageSnapshot() {
Map<String, String> expected = props("iceberg.catalog.type",
"s3tables",
+ "http-client.type", "urlconnection",
"warehouse", "arn:aws:s3tables:us-east-1:1:bucket/b",
"s3.region", "us-east-1",
"client.region", "us-east-1");
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogPropertiesTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogPropertiesTest.java
index 29c18e33677..99e13b5ad20 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogPropertiesTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCatalogPropertiesTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector.iceberg;
+import org.apache.iceberg.aws.HttpClientProperties;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -135,6 +136,24 @@ public class IcebergCatalogPropertiesTest {
Assertions.assertTrue(on.isEnableMappingTimestampTz(), "parsing must
stay case-insensitive");
}
+ @Test
+ public void createTimeRulesRejectTheApacheHttpClientType() {
+ // WHY: iceberg-aws's "apache" client needs an AWS SDK newer than
Doris ships; accepting it would only defer
+ // the failure to a NoClassDefFoundError on the first S3 or Glue call.
The rebuild path (of()) must still
+ // accept it so an existing catalog comes back. MUTATION: dropping the
check -> accepted -> red.
+ Map<String, String> p = props("iceberg.catalog.type", "hadoop",
"warehouse", "s3://b/wh",
+ HttpClientProperties.CLIENT_TYPE, "Apache");
+ IllegalArgumentException e =
Assertions.assertThrows(IllegalArgumentException.class,
+ () ->
IcebergCatalogProperties.of(p).checkCreateTimeOnlyRules());
+
Assertions.assertTrue(e.getMessage().contains(HttpClientProperties.CLIENT_TYPE),
+ "the message must name the offending key: " + e.getMessage());
+ Assertions.assertDoesNotThrow(() -> IcebergCatalogProperties.of(p));
+ Assertions.assertDoesNotThrow(() -> IcebergCatalogProperties.of(props(
+ "iceberg.catalog.type", "hadoop", "warehouse", "s3://b/wh",
+ HttpClientProperties.CLIENT_TYPE,
HttpClientProperties.CLIENT_TYPE_URLCONNECTION))
+ .checkCreateTimeOnlyRules());
+ }
+
@Test
public void createTimeRulesRejectABadMetaCacheValue() {
// WHY: this restores the legacy fail-fast the SPI cutover dropped.
Without it a typo'd ttl is
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 74ec5b7589b..be4736618e3 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
@@ -214,7 +214,7 @@ public class IcebergConnectorCacheTest {
Table table = tableWithOneManifest();
ManifestFile manifest =
table.currentSnapshot().dataManifests(table.io()).get(0);
IcebergManifestCache manifestCache = connector.manifestCacheForTest();
- manifestCache.getManifestCacheValue(manifest, table);
+ manifestCache.getManifestCacheValue(manifest, table, table.specs());
Assertions.assertEquals(1, manifestCache.size(), "the manifest is
cached after a load");
// REFRESH TABLE must NOT drop the manifest cache (path-keyed
immutable content; legacy parity).
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTestConnectionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTestConnectionTest.java
index c1e665b905a..4db495c6414 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTestConnectionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTestConnectionTest.java
@@ -21,8 +21,10 @@ import org.apache.doris.connector.spi.ConnectorContext;
import org.apache.doris.connector.spi.ConnectorStorageContext;
import org.apache.doris.connector.spi.ConnectorTestResult;
+import org.apache.iceberg.aws.HttpClientProperties;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.s3.S3Client;
import java.lang.reflect.Method;
import java.util.HashMap;
@@ -125,7 +127,7 @@ public class IcebergConnectorTestConnectionTest {
.usePathStyle("false");
Map<String, String> ioProps =
IcebergConnector.buildStorageProbeProperties(
- IcebergCatalogProperties.TYPE_DLF, oss);
+ IcebergCatalogProperties.TYPE_DLF, oss, new HashMap<>());
Assertions.assertEquals("http://s3.oss-cn-hangzhou-internal.aliyuncs.com";,
ioProps.get("s3.endpoint"));
Assertions.assertEquals("bound-ak", ioProps.get("s3.access-key-id"));
@@ -135,6 +137,65 @@ public class IcebergConnectorTestConnectionTest {
Assertions.assertEquals("false", ioProps.get("s3.path-style-access"));
}
+ @Test
+ public void storageProbePropertiesBuildAnS3ClientOnTheShippedSdk() {
+ // WHY: the probe builds its own S3FileIO from this map, outside
buildBaseCatalogProperties. iceberg 1.12's
+ // default "apache" client is Apache5HttpClient, absent from the AWS
SDK Doris ships, so CREATE CATALOG failed
+ // with NoClassDefFoundError (an Error the probe's catch(Exception)
never turns into a test result).
+ // MUTATION: drop the default -> "apache" -> NoClassDefFoundError ->
red.
+ Map<String, String> ioProps =
IcebergConnector.buildStorageProbeProperties(
+ IcebergCatalogProperties.TYPE_REST,
+ new
FakeS3CompatibleStorageProperties("S3").endpoint("http://127.0.0.1:1";)
+ .region("us-east-1").accessKey("ak").secretKey("sk"),
+ new HashMap<>());
+
+ Assertions.assertEquals(HttpClientProperties.CLIENT_TYPE_URLCONNECTION,
+ ioProps.get(HttpClientProperties.CLIENT_TYPE));
+ Assertions.assertDoesNotThrow(
+ () -> new
HttpClientProperties(ioProps).applyHttpClientConfigurations(S3Client.builder()));
+ }
+
+ @Test
+ public void storageProbeTreatsABlankCatalogHttpClientTypeAsUnset() {
+ // WHY: the probe copies the catalog's http-client.* keys, so a blank
type must not slip through to iceberg-aws,
+ // which would turn it into "apache". MUTATION: copying "" without
normalizing -> NoClassDefFoundError -> red.
+ Map<String, String> catalogProperties = new HashMap<>();
+ catalogProperties.put(HttpClientProperties.CLIENT_TYPE, "");
+
+ Map<String, String> ioProps =
IcebergConnector.buildStorageProbeProperties(
+ IcebergCatalogProperties.TYPE_REST,
+ new
FakeS3CompatibleStorageProperties("S3").endpoint("http://127.0.0.1:1";)
+ .region("us-east-1").accessKey("ak").secretKey("sk"),
+ catalogProperties);
+
+ Assertions.assertEquals(HttpClientProperties.CLIENT_TYPE_URLCONNECTION,
+ ioProps.get(HttpClientProperties.CLIENT_TYPE));
+ Assertions.assertDoesNotThrow(
+ () -> new
HttpClientProperties(ioProps).applyHttpClientConfigurations(S3Client.builder()));
+ }
+
+ @Test
+ public void storageProbeUsesTheCatalogHttpClientSettings() {
+ // WHY: the probe must reach storage the way the catalog will (proxy,
timeouts), so the catalog's
+ // http-client.* keys win over the default. MUTATION: not copying them
-> the probe ignores the proxy -> red.
+ Map<String, String> catalogProperties = new HashMap<>();
+ catalogProperties.put(HttpClientProperties.CLIENT_TYPE,
HttpClientProperties.CLIENT_TYPE_URLCONNECTION);
+ catalogProperties.put("http-client.proxy-endpoint",
"http://proxy:3128";);
+ catalogProperties.put("s3.endpoint", "http://catalog-only-key";);
+
+ Map<String, String> ioProps =
IcebergConnector.buildStorageProbeProperties(
+ IcebergCatalogProperties.TYPE_REST,
+ new
FakeS3CompatibleStorageProperties("S3").endpoint("http://127.0.0.1:1";)
+ .region("us-east-1").accessKey("ak").secretKey("sk"),
+ catalogProperties);
+
+ Assertions.assertEquals("http://proxy:3128";,
ioProps.get("http-client.proxy-endpoint"));
+ Assertions.assertEquals(HttpClientProperties.CLIENT_TYPE_URLCONNECTION,
+ ioProps.get(HttpClientProperties.CLIENT_TYPE));
+ Assertions.assertEquals("http://127.0.0.1:1";,
ioProps.get("s3.endpoint"),
+ "only http-client.* keys are copied; the storage binder stays
the source of s3.*");
+ }
+
@Test
public void
dlfQualifiedOssWarehouseUsesSelectedStorageNormalizerForProbe() throws
Exception {
String warehouse =
"oss://bucket.oss-cn-hangzhou.aliyuncs.com/warehouse";
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
index c777f0ddfd3..d4e8edc6d42 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestCacheTest.java
@@ -18,9 +18,12 @@
package org.apache.doris.connector.iceberg;
import org.apache.doris.connector.cache.CatalogMetaCache;
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.ManifestContent;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.PartitionSpec;
@@ -34,6 +37,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -92,7 +96,7 @@ public class IcebergManifestCacheTest {
ManifestFile manifest = manifests.get(0);
IcebergManifestCache cache = new IcebergManifestCache();
- ManifestCacheValue first = cache.getManifestCacheValue(manifest,
table);
+ ManifestCacheValue first = cache.getManifestCacheValue(manifest,
table, table.specs());
// WHY: a DATA manifest yields the two appended data files. MUTATION:
returning the delete-file branch
// (empty) -> 0 -> red.
Assertions.assertEquals(2, first.getDataFiles().size());
@@ -101,7 +105,7 @@ public class IcebergManifestCacheTest {
// A second read of the SAME manifest path returns the SAME cached
value (no re-parse). MUTATION: not
// caching (re-loading) -> a different instance -> red.
- ManifestCacheValue second = cache.getManifestCacheValue(manifest,
table);
+ ManifestCacheValue second = cache.getManifestCacheValue(manifest,
table, table.specs());
Assertions.assertSame(first, second, "the manifest payload must be
served from the cache on a repeat");
Assertions.assertEquals(1, cache.size());
}
@@ -114,14 +118,36 @@ public class IcebergManifestCacheTest {
IcebergManifestCache cache = new IcebergManifestCache(owner,
Collections.singletonMap("meta.cache.iceberg.manifest.max-weight", "1MB"));
- ManifestCacheValue first = cache.getManifestCacheValue(manifest,
table);
- ManifestCacheValue second = cache.getManifestCacheValue(manifest,
table);
+ ManifestCacheValue first = cache.getManifestCacheValue(manifest,
table, table.specs());
+ ManifestCacheValue second = cache.getManifestCacheValue(manifest,
table, table.specs());
Assertions.assertSame(first, second);
Assertions.assertEquals(2, first.getDataFiles().size());
}
}
+ @Test
+ public void manifestKeyWeightCountsTheBoundPartitionType() {
+ // WHY: the key retains the bound partition type, and an empty value
(a delete manifest whose entries are all
+ // DELETED) holds no file that references it, so only the key estimate
can account for it. MUTATION:
+ // estimateManifestKey ignoring the partition type -> both entries
weigh the same -> red.
+ ManifestCacheValue empty =
ManifestCacheValue.forDeleteFiles(Collections.emptyList(), true);
+ MetaCacheSizeEstimate narrow =
IcebergCacheSizeEstimator.estimateManifestEntry(deleteManifestKey(1), empty);
+ MetaCacheSizeEstimate wide =
IcebergCacheSizeEstimator.estimateManifestEntry(deleteManifestKey(10), empty);
+ Assertions.assertTrue(narrow.isComplete() && wide.isComplete());
+ long extraFieldBytes = wide.getBytes() - narrow.getBytes();
+ Assertions.assertTrue(extraFieldBytes >= 9 *
JvmSizeUtils.instanceSize(Types.NestedField.class),
+ "narrow=" + narrow.getBytes() + ", wide=" + wide.getBytes());
+ }
+
+ private static IcebergManifestEntryKey deleteManifestKey(int
partitionFieldCount) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (int i = 0; i < partitionFieldCount; i++) {
+ fields.add(Types.NestedField.optional(1000 + i, "p" + i,
Types.IntegerType.get()));
+ }
+ return new IcebergManifestEntryKey("/m/deletes.avro",
ManifestContent.DELETES, Types.StructType.of(fields));
+ }
+
@Test
public void equalityDeleteFieldIdsLoadOncePerSnapshot() {
IcebergManifestCache cache = new IcebergManifestCache();
@@ -172,9 +198,9 @@ public class IcebergManifestCacheTest {
// maxSize 1: the first load fills it; a re-read still hits (same
key). The wholesale-flush valve is the
// legacy behavior — re-reads are harmless since the value is
immutable. Smoke that size stays bounded.
IcebergManifestCache cache = new IcebergManifestCache(1);
- cache.getManifestCacheValue(manifest, table);
+ cache.getManifestCacheValue(manifest, table, table.specs());
Assertions.assertEquals(1, cache.size());
- cache.getManifestCacheValue(manifest, table);
+ cache.getManifestCacheValue(manifest, table, table.specs());
Assertions.assertTrue(cache.size() <= 1, "a bounded cache must not
grow past its max size");
}
@@ -183,7 +209,7 @@ public class IcebergManifestCacheTest {
Table table = tableWithTwoDataFiles();
ManifestFile manifest =
table.currentSnapshot().dataManifests(table.io()).get(0);
IcebergManifestCache cache = new IcebergManifestCache();
- cache.getManifestCacheValue(manifest, table);
+ cache.getManifestCacheValue(manifest, table, table.specs());
AtomicInteger equalityLoads = new AtomicInteger();
cache.getOrLoadEqualityDeleteFieldIds(table.location(),
table.currentSnapshot().snapshotId(), () -> {
equalityLoads.incrementAndGet();
@@ -205,7 +231,8 @@ public class IcebergManifestCacheTest {
public void copiesDataFilesSoReaderReuseDoesNotAlias() {
Table table = tableWithTwoDataFiles();
ManifestFile manifest =
table.currentSnapshot().dataManifests(table.io()).get(0);
- List<DataFile> files = new
IcebergManifestCache().getManifestCacheValue(manifest, table).getDataFiles();
+ List<DataFile> files = new IcebergManifestCache()
+ .getManifestCacheValue(manifest, table,
table.specs()).getDataFiles();
// WHY: ManifestReader reuses one object across iterations, so the
cache must .copy() each file. Two
// distinct paths prove the entries were copied out (not the same
reused instance). MUTATION: dropping
// .copy() -> both entries alias the last file -> both paths equal ->
red.
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKeyTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKeyTest.java
index 96371cfa606..ecd4c08d213 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKeyTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergManifestEntryKeyTest.java
@@ -18,19 +18,25 @@
package org.apache.doris.connector.iceberg;
import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
/**
- * Unit tests for {@link IcebergManifestEntryKey} (T08). The key is path +
content; two manifests that share a
- * path (across tables) hit the same cache entry, which is what lets a REFRESH
TABLE keep the manifest cache.
+ * Unit tests for {@link IcebergManifestEntryKey} (T08). The key is path +
content + bound partition type; two
+ * manifests that share a path (across tables) hit the same cache entry, which
is what lets a REFRESH TABLE keep the
+ * manifest cache.
*/
public class IcebergManifestEntryKeyTest {
+ private static final Types.StructType INT_PARTITION = Types.StructType.of(
+ Types.NestedField.optional(1000, "p", Types.IntegerType.get()));
+ private static final Types.StructType LONG_PARTITION = Types.StructType.of(
+ Types.NestedField.optional(1000, "p", Types.LongType.get()));
@Test
public void samePathSameContentAreEqual() {
- IcebergManifestEntryKey a = new
IcebergManifestEntryKey("/m/shared.avro", ManifestContent.DATA);
- IcebergManifestEntryKey b = new
IcebergManifestEntryKey("/m/shared.avro", ManifestContent.DATA);
+ IcebergManifestEntryKey a = new
IcebergManifestEntryKey("/m/shared.avro", ManifestContent.DATA, INT_PARTITION);
+ IcebergManifestEntryKey b = new
IcebergManifestEntryKey("/m/shared.avro", ManifestContent.DATA, INT_PARTITION);
// WHY: the path-keyed cache must treat two references to the same
manifest path as one entry (cross-table
// sharing). MUTATION: keying on identity instead of (path, content)
-> not equal -> red.
Assertions.assertEquals(a, b);
@@ -39,8 +45,9 @@ public class IcebergManifestEntryKeyTest {
@Test
public void differentContentSamePathAreNotEqual() {
- IcebergManifestEntryKey data = new
IcebergManifestEntryKey("/m/x.avro", ManifestContent.DATA);
- IcebergManifestEntryKey deletes = new
IcebergManifestEntryKey("/m/x.avro", ManifestContent.DELETES);
+ IcebergManifestEntryKey data = new
IcebergManifestEntryKey("/m/x.avro", ManifestContent.DATA, INT_PARTITION);
+ IcebergManifestEntryKey deletes =
+ new IcebergManifestEntryKey("/m/x.avro",
ManifestContent.DELETES, INT_PARTITION);
// A data manifest and a delete manifest are distinct cache entries
even at the same path. MUTATION:
// ignoring content -> equal -> red.
Assertions.assertNotEquals(data, deletes);
@@ -49,7 +56,17 @@ public class IcebergManifestEntryKeyTest {
@Test
public void differentPathAreNotEqual() {
Assertions.assertNotEquals(
- new IcebergManifestEntryKey("/m/a.avro", ManifestContent.DATA),
- new IcebergManifestEntryKey("/m/b.avro",
ManifestContent.DATA));
+ new IcebergManifestEntryKey("/m/a.avro", ManifestContent.DATA,
INT_PARTITION),
+ new IcebergManifestEntryKey("/m/b.avro", ManifestContent.DATA,
INT_PARTITION));
+ }
+
+ @Test
+ public void differentBoundPartitionTypeAreNotEqual() {
+ // The reader converts partition values to the bound spec's types, so
a manifest decoded for a scan before an
+ // INT -> BIGINT promotion and one decoded for a current scan hold
different values. MUTATION: ignoring the
+ // partition type -> equal -> one scan reads the other's Integer/Long
values -> red.
+ Assertions.assertNotEquals(
+ new IcebergManifestEntryKey("/m/x.avro", ManifestContent.DATA,
INT_PARTITION),
+ new IcebergManifestEntryKey("/m/x.avro", ManifestContent.DATA,
LONG_PARTITION));
}
}
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 77d39888fdf..305fbf81347 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
@@ -2497,6 +2497,60 @@ public class IcebergScanPlanProviderTest {
assertHistoricalStreamingDispatch(table, historicalSnapshotId,
historicalSchemaId);
}
+ @Test
+ public void
cachedStreamingScansKeepPartitionTypesAcrossIntToBigintPromotion() throws
IOException {
+ // WHY: the manifest cache shares one decoded manifest across scans,
but a scan pinned to the schema before an
+ // INT -> BIGINT promotion of an identity partition binds its residual
to INT, while a current scan binds it to
+ // LONG. Values decoded for one binding must not reach the other: the
historical scan failed with
+ // "Wrong class, expected java.lang.Integer, but was java.lang.Long"
(and the reverse for the current scan).
+ // The position delete in p=1 runs the same partitions through the
delete index. MUTATION: reading cached
+ // manifests with table.specs() (or the manifest's own spec) for every
scan -> one of the two orders -> red.
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "p", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "value",
Types.StringType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("p").build();
+ Table table = createTable("promoted_partition", schema, spec,
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ String location = "s3://b/db/promoted_partition/";
+ table.newRowDelta()
+ .addRows(dataFile(table.spec(), location + "old.parquet",
1024, null, "p=1"))
+ .addDeletes(FileMetadata.deleteFileBuilder(table.spec())
+ .ofPositionDeletes()
+ .withPath(location + "old-deletes.parquet")
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(128L)
+ .withRecordCount(1L)
+ .withPartitionPath("p=1")
+ .build())
+ .commit();
+ long historicalSnapshotId = table.currentSnapshot().snapshotId();
+ int historicalSchemaId = table.currentSnapshot().schemaId();
+ table.updateSchema().updateColumn("p", Types.LongType.get()).commit();
+ table.newAppend().appendFile(dataFile(table.spec(), location +
"new.parquet", 1024, null, "p=2")).commit();
+
+ IcebergTableHandle historical = new IcebergTableHandle("db1",
"promoted_partition")
+ .withSnapshot(historicalSnapshotId, null, historicalSchemaId);
+ IcebergTableHandle current = new IcebergTableHandle("db1",
"promoted_partition")
+ .withSnapshot(table.currentSnapshot().snapshotId(), null,
table.schema().schemaId());
+ ConnectorExpression intFilter = eqInt("p", 1);
+ ConnectorExpression bigintFilter = new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("p", ConnectorType.of("BIGINT")),
+ new ConnectorLiteral(ConnectorType.of("BIGINT"), 1L));
+
+ for (boolean historicalFirst : new boolean[] {true, false}) {
+ IcebergScanPlanProvider provider =
+ manifestProvider(manifestCacheProps(), table, new
IcebergManifestCache());
+ for (int i = 0; i < 2; i++) {
+ boolean scanHistorical = (i == 0) == historicalFirst;
+ List<ConnectorScanRange> ranges =
drain(provider.streamSplits(batchSession(1, true),
+ scanHistorical ? historical : current,
Collections.emptyList(),
+ Optional.of(scanHistorical ? intFilter :
bigintFilter), -1L));
+ Assertions.assertEquals(Collections.singletonList(location +
"old.parquet"), sortedPaths(ranges),
+ (scanHistorical ? "historical" : "current") + " scan,
historical first=" + historicalFirst);
+ }
+ }
+ }
+
private static void assertHistoricalStreamingDispatch(
Table table, long historicalSnapshotId, int historicalSchemaId)
throws IOException {
IcebergTableHandle handle = new IcebergTableHandle("db1",
"historical_batch_estimate")
diff --git a/fe/pom.xml b/fe/pom.xml
index b2226f40cdf..a0f247bfe64 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -246,7 +246,9 @@ under the License.
<module>fe-grpc</module>
</modules>
<properties>
- <!-- iceberg 1.9.1 depends avro on 1.12 -->
+ <!-- iceberg 1.12.0 is built against avro 1.12.2, but stays on 1.12.1:
avro 1.12.2 forbids SpecificDatumReader
+ from instantiating classes outside
org.apache.avro.SERIALIZABLE_PACKAGES (AVRO-4189), which breaks Hudi's
+ commit metadata readers. Every avro API iceberg 1.12.0 references
exists in 1.12.1. -->
<avro.version>1.12.1</avro.version>
<parquet.version>1.18.0</parquet.version>
<spark.version>3.4.3</spark.version>
@@ -360,7 +362,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.11.0</iceberg.version>
+ <iceberg.version>1.12.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. -->
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
