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 7224a2d382b [test](iceberg) Expand capability, transaction and
snapshot coverage (#69551)
7224a2d382b is described below
commit 7224a2d382b4276d30ccd32324e0499ae9a23fd7
Author: Gabriel <[email protected]>
AuthorDate: Sun Oct 25 22:34:06 2026 +0800
[test](iceberg) Expand capability, transaction and snapshot coverage
(#69551)
### What problem does this PR solve?
Iceberg has broad regression coverage, but feature names alone miss
state transitions and weak oracles. For example, expiring fresh
snapshots with a cutoff in 2020 never exercises ref retention.
Add 45 connector unit tests, two external regression suites, and an
inventory mapping the published 84 operations plus read, version, model,
lineage, type, scanner and storage dimensions to existing tests and
remaining gaps. The tests cover snapshot/schema/ref isolation, atomic
writes and failure paths, nested type metadata, exact predicate
semantics, partition serialization, actual ref retention, and file
positions across deletes and history.
Nine individually disabled reproductions preserve correct assertions for
five confirmed issue families: DECIMAL-to-DOUBLE pruning, lineage-name
collisions during a V2-to-V3 upgrade, ignored per-operation isolation
properties, mixed MERGE affected-row counts, and expiration counters
built from stale metadata. These were run and observed failing before
disabling. Production behavior is unchanged.
### Release note
None
### Validation
- Full connector reactor: `mvn -f fe/pom.xml -pl :fe-connector-iceberg
-am test -DfailIfNoTests=false -Dmaven.build.cache.enabled=false` with
the matching Thrift 0.24.0 compiler: BUILD SUCCESS, including FE
Checkstyle.
- Iceberg connector: 1,734 discovered, 1,720 passed, 14 skipped. New
tests: 45 discovered, 36 passed, 9 intentionally disabled defect
reproductions.
- Both Groovy suites compile with Groovy 4.0.19. They use existing
REST/MinIO/Spark fixtures and dynamic comparisons, with no handwritten
golden results.
- SQL suites were not executed: the managed QA/Jenkins endpoints are
unreachable from the validation host. BE unit tests and the full
backend/deployment matrix were not run.
### Check List (For Author)
- Test
- [x] Regression test (added; compilation checked; SQL execution
pending)
- [x] Unit Test
- Behavior changed:
- [x] No.
- Does this need documentation?
- [x] No. Test coverage inventory is included.
---
.../iceberg/IcebergLineageDdlCoverageTest.java | 122 +++++++
.../iceberg/IcebergLineageDdlKnownBugsTest.java | 81 +++++
.../iceberg/IcebergPartitionUtilsCoverageTest.java | 159 ++++++++
.../IcebergPredicateConverterCoverageTest.java | 211 +++++++++++
.../iceberg/IcebergSnapshotSchemaCoverageTest.java | 157 ++++++++
.../iceberg/IcebergTypeMappingCoverageTest.java | 113 ++++++
.../iceberg/IcebergWriteKnownBugsTest.java | 179 +++++++++
.../IcebergWriteTransactionCoverageTest.java | 404 +++++++++++++++++++++
.../IcebergSnapshotMaintenanceCoverageTest.java | 193 ++++++++++
.../iceberg/ICEBERG_CAPABILITY_TEST_COVERAGE.md | 160 ++++++++
...est_iceberg_file_metadata_delete_history.groovy | 110 ++++++
...iceberg_ref_retention_expiration_control.groovy | 76 ++++
12 files changed, 1965 insertions(+)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlCoverageTest.java
new file mode 100644
index 00000000000..8bf23e7e190
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlCoverageTest.java
@@ -0,0 +1,122 @@
+// 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.doris.connector.iceberg.IcebergCatalogOps.CatalogBackedIcebergCatalogOps;
+import org.apache.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.iceberg.HasTableOperations;
+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.inmemory.InMemoryCatalog;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+public class IcebergLineageDdlCoverageTest {
+ private static final TableIdentifier IDENTIFIER = TableIdentifier.of("db",
"events");
+ private static final IcebergTableHandle HANDLE = new
IcebergTableHandle("db", "events");
+
+ @Test
+ public void addDoesNotExposeDuplicateRowIdColumns() {
+ assertAddPreservesUniqueMetadataNames("_row_id");
+ }
+
+ @Test
+ public void addDoesNotExposeDuplicateSequenceColumns() {
+ assertAddPreservesUniqueMetadataNames("_last_updated_sequence_number");
+ }
+
+ @Test
+ public void renameDoesNotExposeDuplicateRowIdColumns() {
+ assertRenamePreservesUniqueMetadataNames("_row_id");
+ }
+
+ @Test
+ public void renameDoesNotExposeDuplicateSequenceColumns() {
+
assertRenamePreservesUniqueMetadataNames("_last_updated_sequence_number");
+ }
+
+ private void assertAddPreservesUniqueMetadataNames(String reservedName) {
+ InMemoryCatalog catalog = catalog();
+ createV3(catalog);
+ IcebergConnectorMetadata metadata = metadata(catalog);
+ String previousMetadata = metadataLocation(catalog);
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadata.addColumn(null, HANDLE,
+ new ConnectorColumn(reservedName,
ConnectorType.of("BIGINT"), "", true, null, false), null));
+ Assertions.assertTrue(failure.getMessage().contains("reserved row
lineage column"));
+ Assertions.assertEquals(previousMetadata, metadataLocation(catalog));
+
Assertions.assertNull(catalog.loadTable(IDENTIFIER).schema().findField(reservedName));
+ assertUniqueColumnName(metadata, reservedName);
+ }
+
+ private void assertRenamePreservesUniqueMetadataNames(String reservedName)
{
+ InMemoryCatalog catalog = catalog();
+ createV3(catalog);
+ IcebergConnectorMetadata metadata = metadata(catalog);
+ String previousMetadata = metadataLocation(catalog);
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadata.renameColumn(null, HANDLE, "id", reservedName));
+ Assertions.assertTrue(failure.getMessage().contains("reserved row
lineage column"));
+ Assertions.assertEquals(previousMetadata, metadataLocation(catalog));
+
Assertions.assertNotNull(catalog.loadTable(IDENTIFIER).schema().findField("id"));
+
Assertions.assertNull(catalog.loadTable(IDENTIFIER).schema().findField(reservedName));
+ assertUniqueColumnName(metadata, reservedName);
+ }
+
+ private static void assertUniqueColumnName(IcebergConnectorMetadata
metadata, String reservedName) {
+ // A public schema must not contain both a physical user field and
hidden metadata under one name.
+ long matchingColumns = metadata.getTableSchema(null,
HANDLE).getColumns().stream()
+ .filter(column ->
reservedName.equalsIgnoreCase(column.getName())).count();
+ Assertions.assertEquals(1L, matchingColumns,
+ "The connector exposed duplicate physical and hidden columns
named " + reservedName);
+ }
+
+ private static String metadataLocation(InMemoryCatalog catalog) {
+ Table table = catalog.loadTable(IDENTIFIER);
+ return ((HasTableOperations)
table).operations().current().metadataFileLocation();
+ }
+
+ private static InMemoryCatalog catalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("repro", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db"));
+ return catalog;
+ }
+
+ private static void createV3(InMemoryCatalog catalog) {
+ catalog.createTable(IDENTIFIER, new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get())),
+ PartitionSpec.unpartitioned(),
Collections.singletonMap("format-version", "3"));
+ }
+
+ private static IcebergConnectorMetadata metadata(InMemoryCatalog catalog) {
+ return new IcebergConnectorMetadata(new
CatalogBackedIcebergCatalogOps(catalog),
+ IcebergCatalogProperties.of(Collections.singletonMap(
+ IcebergCatalogProperties.ICEBERG_CATALOG_TYPE,
IcebergCatalogProperties.TYPE_REST)),
+ new RecordingConnectorContext());
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlKnownBugsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlKnownBugsTest.java
new file mode 100644
index 00000000000..ebe9d419db6
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLineageDdlKnownBugsTest.java
@@ -0,0 +1,81 @@
+// 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.doris.connector.iceberg.IcebergCatalogOps.CatalogBackedIcebergCatalogOps;
+
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+public class IcebergLineageDdlKnownBugsTest {
+ private static final TableIdentifier IDENTIFIER = TableIdentifier.of("db",
"events");
+ private static final IcebergTableHandle HANDLE = new
IcebergTableHandle("db", "events");
+
+ @Test
+ @Disabled("Known issue: a V2-to-V3 upgrade permits a user column to
collide with row lineage.")
+ public void upgradeDoesNotExposeDuplicateRowIdColumns() {
+ assertUpgradePreservesUniqueMetadataNames("_row_id");
+ }
+
+ @Test
+ @Disabled("Known issue: a V2-to-V3 upgrade permits a user column to
collide with row lineage.")
+ public void upgradeDoesNotExposeDuplicateSequenceColumns() {
+
assertUpgradePreservesUniqueMetadataNames("_last_updated_sequence_number");
+ }
+
+ private void assertUpgradePreservesUniqueMetadataNames(String
reservedName) {
+ InMemoryCatalog catalog = catalog();
+ catalog.createTable(IDENTIFIER, new Schema(
+ Types.NestedField.optional(1, reservedName,
Types.LongType.get())),
+ PartitionSpec.unpartitioned(),
Collections.singletonMap("format-version", "2"));
+ IcebergConnectorMetadata metadata = metadata(catalog);
+ metadata.setTableProperties(null, HANDLE,
Collections.singletonMap("format-version", "3"));
+ assertUniqueColumnName(metadata, reservedName);
+ }
+
+ private static void assertUniqueColumnName(IcebergConnectorMetadata
metadata, String reservedName) {
+ // A public schema must not contain both a physical user field and
hidden metadata under one name.
+ long matchingColumns = metadata.getTableSchema(null,
HANDLE).getColumns().stream()
+ .filter(column ->
reservedName.equalsIgnoreCase(column.getName())).count();
+ Assertions.assertEquals(1L, matchingColumns,
+ "The connector exposed duplicate physical and hidden columns
named " + reservedName);
+ }
+
+ private static InMemoryCatalog catalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("repro", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db"));
+ return catalog;
+ }
+
+ private static IcebergConnectorMetadata metadata(InMemoryCatalog catalog) {
+ return new IcebergConnectorMetadata(new
CatalogBackedIcebergCatalogOps(catalog),
+ IcebergCatalogProperties.of(Collections.singletonMap(
+ IcebergCatalogProperties.ICEBERG_CATALOG_TYPE,
IcebergCatalogProperties.TYPE_REST)),
+ new RecordingConnectorContext());
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsCoverageTest.java
new file mode 100644
index 00000000000..b6a0ca9c729
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPartitionUtilsCoverageTest.java
@@ -0,0 +1,159 @@
+// 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.PartitionData;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.List;
+
+public class IcebergPartitionUtilsCoverageTest {
+ @Test
+ public void compoundPartitionWireFormatPreservesEscapingAndExactValues() {
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "text", Types.StringType.get()),
+ Types.NestedField.optional(2, "nullable",
Types.StringType.get()),
+ Types.NestedField.optional(3, "amount",
Types.DecimalType.of(38, 18)),
+ Types.NestedField.optional(4, "date", Types.DateType.get()),
+ Types.NestedField.optional(5, "timestamp",
Types.TimestampNanoType.withoutZone()),
+ Types.NestedField.optional(6, "binary",
Types.BinaryType.get()));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("text").identity("nullable")
+
.identity("amount").identity("date").identity("timestamp").identity("binary").build();
+ PartitionData original = new PartitionData(spec.partitionType());
+ List<Object> values = Arrays.asList("null/\"quoted\"\\line\n\u0000雪",
null,
+ new BigDecimal("-99999999999999999999.000000000000000001"), -1,
+ -1L, ByteBuffer.wrap(new byte[] {0, (byte) 0xff, (byte) 0x80,
34, 92}));
+ for (int i = 0; i < values.size(); i++) {
+ original.set(i, values.get(i));
+ }
+
+ for (ZoneId zone : Arrays.asList(ZoneOffset.UTC,
ZoneId.of("America/New_York"))) {
+ List<String> encoded =
IcebergPartitionUtils.parsePartitionValuesFromJson(
+ IcebergPartitionUtils.getPartitionDataJson(original, spec,
zone));
+ Assertions.assertEquals(values.size(), encoded.size());
+ Assertions.assertEquals(values.get(0), encoded.get(0));
+ Assertions.assertNull(encoded.get(1));
+ PartitionData decoded = new PartitionData(spec.partitionType());
+ for (int i = 0; i < encoded.size(); i++) {
+ decoded.set(i,
IcebergPartitionUtils.parsePartitionValueFromString(
+ encoded.get(i),
spec.partitionType().fields().get(i).type(), zone));
+ }
+ // PartitionData stores binary as byte[], whose container equality
compares array identities.
+ for (int i = 0; i < original.size(); i++) {
+ Assertions.assertEquals(original.get(i), decoded.get(i),
"partition field " + i);
+ }
+ Assertions.assertEquals(18, ((BigDecimal) decoded.get(2)).scale());
+ }
+ }
+
+ @Test
+ public void
calendarRangesAgreeWithSdkTransformsAcrossEpochAndLeapBoundaries() {
+ List<LocalDateTime> timestamps = Arrays.asList(
+ LocalDateTime.of(1600, 2, 29, 23, 59, 59, 999999000),
+ LocalDateTime.of(1900, 3, 1, 0, 0),
+ LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999999000),
+ LocalDateTime.of(1970, 1, 1, 0, 0),
+ LocalDateTime.of(2000, 2, 29, 23, 59, 59, 999999000),
+ LocalDateTime.of(2100, 3, 1, 0, 0));
+ for (Type type : Arrays.asList(Types.DateType.get(),
+ Types.TimestampType.withoutZone(),
Types.TimestampType.withZone())) {
+ for (String name : Arrays.asList("year", "month", "day", "hour")) {
+ if (type.typeId() == Type.TypeID.DATE && name.equals("hour")) {
+ continue;
+ }
+ Transform<Object, Integer> transform = calendarTransform(name);
+ for (LocalDateTime timestamp : timestamps) {
+ int ordinal =
transform.bind(type).apply(internalValue(type, timestamp));
+ IcebergPartitionUtils.RangeBuild range =
IcebergPartitionUtils.buildRange(
+ name + "=" + ordinal, Integer.toString(ordinal),
name, type, 100L, 200L);
+ LocalDateTime lower =
parseBound(range.getLowerBound().get(0), type);
+ LocalDateTime upper =
parseBound(range.getUpperBound().get(0), type);
+ LocalDateTime row = type.typeId() == Type.TypeID.DATE
+ ? timestamp.toLocalDate().atStartOfDay() :
timestamp;
+ Assertions.assertFalse(row.isBefore(lower), name + " " +
timestamp);
+ Assertions.assertTrue(row.isBefore(upper), name + " " +
timestamp);
+ // Both ends must describe exactly one SDK transform
ordinal, including negative ordinals.
+ Assertions.assertEquals(ordinal,
transform.bind(type).apply(internalValue(type, lower)));
+ LocalDateTime last = type.typeId() == Type.TypeID.DATE
+ ? upper.minusDays(1) : upper.minusNanos(1000);
+ Assertions.assertEquals(ordinal,
transform.bind(type).apply(internalValue(type, last)));
+ Assertions.assertEquals(ordinal + 1,
transform.bind(type).apply(internalValue(type, upper)));
+ }
+ }
+ }
+ }
+
+ @Test
+ public void binarySerializationDoesNotConsumeDirectOrReadOnlyBuffers() {
+ ByteBuffer direct = ByteBuffer.allocateDirect(6);
+ direct.put(new byte[] {99, 0, (byte) 0xfe, (byte) 0xff, 1, 99});
+ direct.position(1);
+ direct.limit(5);
+ for (ByteBuffer buffer : Arrays.asList(direct,
direct.asReadOnlyBuffer(), direct.slice())) {
+ int position = buffer.position();
+ int limit = buffer.limit();
+ for (Type type : Arrays.asList(Types.BinaryType.get(),
Types.FixedType.ofLength(4))) {
+ String encoded =
IcebergPartitionUtils.serializePartitionValue(type, buffer, ZoneOffset.UTC);
+ Assertions.assertEquals("0x00feff01", encoded);
+ Assertions.assertEquals(buffer,
IcebergPartitionUtils.parsePartitionValueFromString(
+ encoded, type, ZoneOffset.UTC));
+ Assertions.assertEquals(position, buffer.position());
+ Assertions.assertEquals(limit, buffer.limit());
+ }
+ }
+ }
+
+ private static Transform<Object, Integer> calendarTransform(String name) {
+ switch (name) {
+ case "year":
+ return Transforms.year();
+ case "month":
+ return Transforms.month();
+ case "day":
+ return Transforms.day();
+ case "hour":
+ return Transforms.hour();
+ default:
+ throw new IllegalArgumentException(name);
+ }
+ }
+
+ private static Object internalValue(Type type, LocalDateTime timestamp) {
+ return type.typeId() == Type.TypeID.DATE ? (Object) (int)
timestamp.toLocalDate().toEpochDay()
+ : timestamp.toEpochSecond(ZoneOffset.UTC) * 1_000_000L +
timestamp.getNano() / 1000;
+ }
+
+ private static LocalDateTime parseBound(String bound, Type type) {
+ return type.typeId() == Type.TypeID.DATE ?
LocalDate.parse(bound).atStartOfDay()
+ : LocalDateTime.parse(bound.replace("+00:00", "").replace(' ',
'T'));
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPredicateConverterCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPredicateConverterCoverageTest.java
new file mode 100644
index 00000000000..6412a9f2ade
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergPredicateConverterCoverageTest.java
@@ -0,0 +1,211 @@
+// 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.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
+import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
+import org.apache.doris.connector.spi.pushdown.ConnectorIn;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
+
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class IcebergPredicateConverterCoverageTest {
+ private static final List<ConnectorComparison.Operator> OPERATORS =
Arrays.asList(
+ ConnectorComparison.Operator.EQ, ConnectorComparison.Operator.LT,
ConnectorComparison.Operator.LE,
+ ConnectorComparison.Operator.GT, ConnectorComparison.Operator.GE);
+
+ @Test
+ public void
integralPredicatesPreserveOrderingAtNarrowingAndLongBoundaries() {
+ long[] literals = {Long.MIN_VALUE, (long) Integer.MIN_VALUE - 1,
Integer.MIN_VALUE, -1, 0, 1,
+ Integer.MAX_VALUE, (long) Integer.MAX_VALUE + 1,
9007199254740993L, Long.MAX_VALUE};
+ for (Type type : Arrays.asList(Types.IntegerType.get(),
Types.LongType.get())) {
+ List<? extends Number> rows = type.typeId() == Type.TypeID.INTEGER
+ ? Arrays.asList(Integer.MIN_VALUE, -1, 0, 1,
Integer.MAX_VALUE)
+ : Arrays.asList(Long.MIN_VALUE, (long) Integer.MIN_VALUE,
-1L, 0L, 1L,
+ (long) Integer.MAX_VALUE, 9007199254740993L,
Long.MAX_VALUE);
+ Schema schema = schema(type);
+ for (IcebergPredicateConverter.Mode mode :
IcebergPredicateConverter.Mode.values()) {
+ for (long literal : literals) {
+ for (ConnectorComparison.Operator operator : OPERATORS) {
+ Evaluator evaluator = evaluator(schema, mode, operator,
+ new
ConnectorLiteral(ConnectorType.of("BIGINT"), literal));
+ for (Number row : rows) {
+ Assertions.assertEquals(matches(operator,
Long.compare(row.longValue(), literal)),
+ evaluator.eval(new SingleValueRow(row)),
+ mode + " " + type + " " + row + " " +
operator + " " + literal);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Test
+ public void decimalPredicatesPreservePrecisionAndEquivalentLiteralScales()
{
+ Schema schema = schema(Types.DecimalType.of(38, 18));
+ List<BigDecimal> values = Arrays.asList(
+ new BigDecimal("-99999999999999999999.999999999999999999"),
+ new BigDecimal("-0.000000000000000001"),
+ new BigDecimal("0.000000000000000000"),
+ new BigDecimal("0.000000000000000001"),
+ new BigDecimal("99999999999999999999.999999999999999999"));
+ for (IcebergPredicateConverter.Mode mode :
IcebergPredicateConverter.Mode.values()) {
+ for (BigDecimal value : values) {
+ for (BigDecimal literal : Arrays.asList(value,
value.setScale(20))) {
+ for (ConnectorComparison.Operator operator : OPERATORS) {
+ Evaluator evaluator = evaluator(schema, mode, operator,
+ new
ConnectorLiteral(ConnectorType.of("DECIMAL256", 40, literal.scale()), literal));
+ for (BigDecimal row : values) {
+ Assertions.assertEquals(matches(operator,
row.compareTo(literal)),
+ evaluator.eval(new SingleValueRow(row)),
+ mode + " " + row + " " + operator + " " +
literal);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @Test
+ @Disabled("Known issue: stripping a DECIMAL-to-DOUBLE cast can prune
matching rows and files.")
+ public void decimalToDoubleComparisonMustNotPruneMatchingRows() {
+ Types.DecimalType type = Types.DecimalType.of(38, 0);
+ Schema schema = schema(type);
+ BigDecimal row = new BigDecimal("9007199254740993");
+ double literal = 9007199254740992d;
+ Assertions.assertTrue(row.doubleValue() == literal);
+ // The engine removes the column's DOUBLE cast; pruning must preserve
its many-to-one equality.
+ ConnectorComparison comparison = new
ConnectorComparison(ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("value", ConnectorType.of("DECIMALV3",
38, 0)),
+ new ConnectorLiteral(ConnectorType.of("DOUBLE"), literal));
+ List<Expression> converted = new IcebergPredicateConverter(schema,
ZoneOffset.UTC).convert(comparison);
+ Map<Integer, ByteBuffer> bounds = Collections.singletonMap(1,
Conversions.toByteBuffer(type, row));
+ Metrics metrics = new Metrics(1L, null, Collections.singletonMap(1,
1L),
+ Collections.singletonMap(1, 0L), null, bounds, bounds);
+ DataFile file = DataFiles.builder(PartitionSpec.unpartitioned())
+
.withPath("data/decimal-double.parquet").withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(1024L).withMetrics(metrics).build();
+ // Dropping the predicate is safe; every predicate that is pushed must
retain this matching file.
+ for (Expression predicate : converted) {
+ Assertions.assertAll(
+ () -> Assertions.assertTrue(new
Evaluator(schema.asStruct(), predicate)
+ .eval(new SingleValueRow(row)), "a matching row
must survive pruning"),
+ () -> Assertions.assertTrue(new
InclusiveMetricsEvaluator(schema, predicate).eval(file),
+ "a file containing a matching row must survive
pruning"));
+ }
+ }
+
+ @Test
+ public void lineageColumnsNeverBecomeFilePruningPredicates() {
+ for (String name : Arrays.asList("_row_id", "_ROW_ID",
"_last_updated_sequence_number",
+ "_LAST_UPDATED_SEQUENCE_NUMBER")) {
+ Schema schema = new Schema(Types.NestedField.optional(1, name,
Types.LongType.get()));
+ ConnectorColumnRef column = new ConnectorColumnRef(name,
ConnectorType.of("BIGINT"));
+ for (IcebergPredicateConverter.Mode mode :
IcebergPredicateConverter.Mode.values()) {
+ IcebergPredicateConverter converter = new
IcebergPredicateConverter(schema, ZoneOffset.UTC, mode);
+ for (ConnectorComparison.Operator operator :
ConnectorComparison.Operator.values()) {
+ Assertions.assertTrue(converter.convert(new
ConnectorComparison(
+ operator, column,
ConnectorLiteral.ofLong(1L))).isEmpty(), mode + " " + operator);
+ }
+ Assertions.assertTrue(converter.convert(new ConnectorIn(column,
+
Collections.singletonList(ConnectorLiteral.ofLong(1L)), false)).isEmpty());
+ Assertions.assertTrue(converter.convert(new
ConnectorComparison(
+ ConnectorComparison.Operator.EQ_FOR_NULL, column,
+
ConnectorLiteral.ofNull(ConnectorType.of("BIGINT")))).isEmpty());
+ }
+ }
+ }
+
+ private static Schema schema(Type type) {
+ return new Schema(Types.NestedField.optional(1, "value", type));
+ }
+
+ private static Evaluator evaluator(Schema schema,
IcebergPredicateConverter.Mode mode,
+ ConnectorComparison.Operator operator, ConnectorLiteral literal) {
+ ConnectorComparison comparison = new ConnectorComparison(operator,
+ new ConnectorColumnRef("value", ConnectorType.of("UNKNOWN")),
literal);
+ List<Expression> converted = new IcebergPredicateConverter(schema,
ZoneOffset.UTC, mode).convert(comparison);
+ Assertions.assertEquals(1, converted.size(), mode + " " + comparison);
+ return new Evaluator(schema.asStruct(), converted.get(0));
+ }
+
+ private static boolean matches(ConnectorComparison.Operator operator, int
comparison) {
+ switch (operator) {
+ case EQ:
+ return comparison == 0;
+ case LT:
+ return comparison < 0;
+ case LE:
+ return comparison <= 0;
+ case GT:
+ return comparison > 0;
+ case GE:
+ return comparison >= 0;
+ default:
+ throw new IllegalArgumentException(operator.toString());
+ }
+ }
+
+ private static final class SingleValueRow implements StructLike {
+ private final Object value;
+
+ private SingleValueRow(Object value) {
+ this.value = value;
+ }
+
+ @Override
+ public int size() {
+ return 1;
+ }
+
+ @Override
+ public <T> T get(int position, Class<T> javaClass) {
+ Assertions.assertEquals(0, position);
+ return javaClass.cast(value);
+ }
+
+ @Override
+ public <T> void set(int position, T value) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java
new file mode 100644
index 00000000000..bab9b4c20e9
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java
@@ -0,0 +1,157 @@
+// 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.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.ConnectorTableSchema;
+import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
+import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot;
+import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec;
+
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+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.inmemory.InMemoryCatalog;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class IcebergSnapshotSchemaCoverageTest {
+ private static final IcebergTableHandle HANDLE = new
IcebergTableHandle("db1", "history");
+
+ private static Table newTable() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ return catalog.createTable(TableIdentifier.of("db1", "history"), new
Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "original_name",
Types.StringType.get())));
+ }
+
+ private static long append(Table table, String name) {
+
table.newAppend().appendFile(DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("s3://bucket/history/" + name +
".parquet").withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(100).withRecordCount(1).build()).commit();
+ return table.currentSnapshot().snapshotId();
+ }
+
+ private static IcebergConnectorMetadata metadata(Table table) {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.table = table;
+ return new IcebergConnectorMetadata(ops,
IcebergCatalogProperties.of(Collections.emptyMap()),
+ new RecordingConnectorContext());
+ }
+
+ private static List<String> names(ConnectorTableSchema schema) {
+ return
schema.getColumns().stream().map(ConnectorColumn::getName).collect(Collectors.toList());
+ }
+
+ @Test
+ public void
branchUsesLatestSchemaWhileTagAtTheSameSnapshotUsesHistoricalSchema() {
+ Table table = newTable();
+ long snapshotId = append(table, "initial");
+ int originalSchemaId = table.schema().schemaId();
+ table.manageSnapshots().createTag("audit_tag", snapshotId)
+ .createBranch("audit_branch", snapshotId).commit();
+ table.updateSchema().renameColumn("original_name", "renamed").commit();
+ IcebergConnectorMetadata metadata = metadata(table);
+
+ ConnectorMvccSnapshot branch = metadata.resolveTimeTravel(null, HANDLE,
+ ConnectorTimeTravelSpec.branch("audit_branch")).get();
+ ConnectorMvccSnapshot tag = metadata.resolveTimeTravel(null, HANDLE,
+ ConnectorTimeTravelSpec.tag("audit_tag")).get();
+
+ // Iceberg branches follow the table schema, whereas tags retain the
referenced snapshot's schema.
+ Assertions.assertEquals(snapshotId, branch.getSnapshotId());
+ Assertions.assertEquals(snapshotId, tag.getSnapshotId());
+ Assertions.assertEquals(table.schema().schemaId(),
branch.getSchemaId());
+ Assertions.assertEquals(originalSchemaId, tag.getSchemaId());
+ Assertions.assertEquals(Arrays.asList("id", "renamed", "_file",
"_pos"),
+ names(metadata.getTableSchema(null, HANDLE, branch)));
+ Assertions.assertEquals(Arrays.asList("id", "original_name", "_file",
"_pos"),
+ names(metadata.getTableSchema(null, HANDLE, tag)));
+ }
+
+ @Test
+ public void
explicitSnapshotRetainsItsSchemaAfterRollbackChangesTheCurrentSnapshot() {
+ Table table = newTable();
+ long initial = append(table, "initial");
+ table.updateSchema().renameColumn("original_name", "renamed").commit();
+ long newer = append(table, "newer");
+ int newerSchema = table.schema().schemaId();
+ table.manageSnapshots().rollbackTo(initial).commit();
+ IcebergConnectorMetadata metadata = metadata(table);
+
+ ConnectorMvccSnapshot historical = metadata.resolveTimeTravel(null,
HANDLE,
+
ConnectorTimeTravelSpec.snapshotId(String.valueOf(newer))).get();
+
+ Assertions.assertEquals(initial, table.currentSnapshot().snapshotId());
+ Assertions.assertEquals(newer, historical.getSnapshotId());
+ Assertions.assertEquals(newerSchema, historical.getSchemaId());
+ Assertions.assertEquals(Arrays.asList("id", "renamed", "_file",
"_pos"),
+ names(metadata.getTableSchema(null, HANDLE, historical)));
+ }
+
+ @Test
+ public void
historicalSchemaKeepsLineageColumnsAfterFormatUpgradeWithoutAnotherSnapshot() {
+ Table table = newTable();
+ long initial = append(table, "initial");
+ int originalSchema = table.schema().schemaId();
+ table.updateSchema().renameColumn("original_name", "renamed").commit();
+ table.updateProperties().set("format-version", "3").commit();
+ IcebergConnectorMetadata metadata = metadata(table);
+ ConnectorMvccSnapshot historical = metadata.resolveTimeTravel(null,
HANDLE,
+
ConnectorTimeTravelSpec.snapshotId(String.valueOf(initial))).get();
+
+ Assertions.assertEquals(originalSchema, historical.getSchemaId());
+ List<ConnectorColumn> columns = metadata.getTableSchema(null, HANDLE,
historical).getColumns();
+ Assertions.assertEquals(Arrays.asList("id", "original_name", "_file",
"_pos",
+ "_row_id", "_last_updated_sequence_number"),
+
columns.stream().map(ConnectorColumn::getName).collect(Collectors.toList()));
+ Assertions.assertFalse(columns.get(4).isVisible());
+ Assertions.assertFalse(columns.get(5).isVisible());
+ Assertions.assertTrue(columns.get(4).isReservedPassthrough());
+ Assertions.assertTrue(columns.get(5).isReservedPassthrough());
+ }
+
+ @Test
+ public void
metadataTablesNeverAcquireDataFileOrLineageColumnsFromAV3Base() {
+ Table table = newTable();
+ table.updateProperties().set("format-version", "3").commit();
+ append(table, "initial");
+ IcebergConnectorMetadata metadata = metadata(table);
+
+ for (String systemTable : metadata.listSupportedSysTables(null,
HANDLE)) {
+ ConnectorTableHandle system = metadata.getSysTableHandle(null,
HANDLE, systemTable).get();
+ List<String> columnNames = names(metadata.getTableSchema(null,
system));
+ Assertions.assertFalse(columnNames.contains("_file"), systemTable);
+ Assertions.assertFalse(columnNames.contains("_pos"), systemTable);
+ Assertions.assertFalse(columnNames.contains("_row_id"),
systemTable);
+
Assertions.assertFalse(columnNames.contains("_last_updated_sequence_number"),
systemTable);
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTypeMappingCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTypeMappingCoverageTest.java
new file mode 100644
index 00000000000..6350b02952b
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTypeMappingCoverageTest.java
@@ -0,0 +1,113 @@
+// 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.doris.connector.spi.ConnectorColumn;
+import org.apache.doris.connector.spi.ConnectorType;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+public class IcebergTypeMappingCoverageTest {
+ @Test
+ public void nestedCollectionMetadataSurvivesReadAndCreateRoundTrip() {
+ Types.StructType value = Types.StructType.of(
+ Types.NestedField.required(71, "EventTime",
Types.TimestampType.withZone(), "event instant"),
+ Types.NestedField.optional(19, "Amounts",
Types.ListType.ofRequired(113,
+ Types.DecimalType.of(38, 18)), "exact amounts"),
+ Types.NestedField.required(104, "Labels",
Types.MapType.ofOptional(92, 27,
+ Types.StringType.get(), Types.UUIDType.get()),
"optional labels"));
+ Type nested = Types.ListType.ofRequired(31,
+ Types.MapType.ofRequired(46, 58, Types.StringType.get(),
value));
+ Schema original = new Schema(Types.NestedField.required(53, "Events",
nested, "nested events"));
+
+ for (boolean binaryFlag : new boolean[] {false, true}) {
+ for (boolean timestampFlag : new boolean[] {false, true}) {
+ ConnectorType mapped =
IcebergTypeMapping.fromIcebergType(nested, binaryFlag, timestampFlag);
+ assertNestedMetadata(nested, mapped);
+ ConnectorColumn column = new ConnectorColumn("Events", mapped,
"nested events", false, null, false);
+ Schema rebuilt =
IcebergSchemaBuilder.buildSchema(Collections.singletonList(column));
+ // CREATE allocates new IDs; all other metadata must survive
the recursive read/write conversion.
+ Assertions.assertEquals(original.asStruct(),
TypeUtil.reassignIds(rebuilt, original).asStruct());
+ assertNestedMetadata(rebuilt.findType("Events"),
IcebergTypeMapping.fromIcebergType(
+ rebuilt.findType("Events"), binaryFlag,
timestampFlag));
+ }
+ }
+ }
+
+ @Test
+ public void
unsupportedNestedLeavesPreserveSupportedSiblingsAndFieldMetadata() {
+ Types.StructType nested = Types.StructType.of(
+ Types.NestedField.required(7, "Values",
Types.ListType.ofOptional(8,
+ Types.StructType.of(
+ Types.NestedField.optional(9,
"UnsupportedTime", Types.TimeType.get()),
+ Types.NestedField.required(10,
"SupportedCount", Types.LongType.get(), "count")))),
+ Types.NestedField.optional(11, "ZonedNanos",
Types.TimestampNanoType.withZone()),
+ Types.NestedField.required(12, "Payload",
Types.VariantType.get()),
+ Types.NestedField.optional(13, "Bytes",
Types.FixedType.ofLength(16)));
+
+ ConnectorType mapped = IcebergTypeMapping.fromIcebergType(nested,
false, false);
+ assertNestedMetadata(nested, mapped);
+ ConnectorType element =
mapped.getChildren().get(0).getChildren().get(0);
+ Assertions.assertEquals("UNSUPPORTED",
element.getChildren().get(0).getTypeName());
+ Assertions.assertEquals("BIGINT",
element.getChildren().get(1).getTypeName());
+ Assertions.assertEquals("UNSUPPORTED",
mapped.getChildren().get(1).getTypeName());
+ Assertions.assertEquals("VARIANT_COMPUTE_V2",
mapped.getChildren().get(2).getTypeName());
+ Assertions.assertEquals(ConnectorType.of("VARBINARY", 16, 0),
mapped.getChildren().get(3));
+ }
+
+ @Test
+ public void decimalPrecisionAndScaleRoundTripAtEverySupportedBoundary() {
+ for (int precision : new int[] {1, 9, 18, 28, 38}) {
+ for (int scale : new int[] {0, precision / 2, precision}) {
+ Types.DecimalType source = Types.DecimalType.of(precision,
scale);
+ ConnectorType mapped =
IcebergTypeMapping.fromIcebergType(source, false, false);
+ Assertions.assertEquals(ConnectorType.of("DECIMALV3",
precision, scale), mapped);
+ Assertions.assertEquals(source,
IcebergTypeMapping.toIcebergPrimitive(mapped));
+ Assertions.assertTrue(IcebergTypeMapping.isReadAs(source,
mapped));
+ Assertions.assertFalse(IcebergTypeMapping.isReadAs(source,
+ ConnectorType.of("DECIMALV3", precision + 1, scale)));
+ }
+ }
+ }
+
+ private static void assertNestedMetadata(Type source, ConnectorType
mapped) {
+ if (!source.isNestedType()) {
+ return;
+ }
+ List<Types.NestedField> fields = source.asNestedType().fields();
+ Assertions.assertEquals(fields.size(), mapped.getChildren().size());
+ for (int i = 0; i < fields.size(); i++) {
+ Types.NestedField field = fields.get(i);
+ Assertions.assertEquals(field.fieldId(),
mapped.getChildFieldId(i), field.toString());
+ Assertions.assertEquals(field.isOptional(),
mapped.isChildNullable(i), field.toString());
+ if (source.isStructType()) {
+ Assertions.assertEquals(field.name(),
mapped.getFieldNames().get(i));
+ Assertions.assertEquals(field.doc(),
mapped.getChildComment(i));
+ }
+ assertNestedMetadata(field.type(), mapped.getChildren().get(i));
+ }
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
new file mode 100644
index 00000000000..f18a79a4e95
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteKnownBugsTest.java
@@ -0,0 +1,179 @@
+// 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.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.handle.WriteOperation;
+import org.apache.doris.thrift.TFileContent;
+import org.apache.doris.thrift.TIcebergCommitData;
+
+import org.apache.iceberg.DataFiles;
+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.inmemory.InMemoryCatalog;
+import org.apache.iceberg.types.Types;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+
+public class IcebergWriteKnownBugsTest {
+ private static final TableIdentifier IDENTIFIER = TableIdentifier.of("db",
"events");
+ private static final String SEED = "s3://bucket/events/seed.parquet";
+ private static final String ADDED = "s3://bucket/events/added.parquet";
+ private static final String DELETED = "s3://bucket/events/deleted.parquet";
+
+ @Test
+ @Disabled("Known issue: DELETE ignores the standard snapshot isolation
property.")
+ public void deleteHonorsStandardSnapshotIsolationProperty() throws
Exception {
+ assertStandardSnapshotIsolation(WriteOperation.DELETE,
"write.delete.isolation-level");
+ }
+
+ @Test
+ @Disabled("Known issue: UPDATE ignores the standard snapshot isolation
property.")
+ public void updateHonorsStandardSnapshotIsolationProperty() throws
Exception {
+ assertStandardSnapshotIsolation(WriteOperation.UPDATE,
"write.update.isolation-level");
+ }
+
+ @Test
+ @Disabled("Known issue: MERGE ignores the standard snapshot isolation
property.")
+ public void mergeHonorsStandardSnapshotIsolationProperty() throws
Exception {
+ assertStandardSnapshotIsolation(WriteOperation.MERGE,
"write.merge.isolation-level");
+ }
+
+ private void assertStandardSnapshotIsolation(WriteOperation operation,
String property) throws Exception {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog);
+ table.updateProperties().set(property, "snapshot").commit();
+
table.newAppend().appendFile(DataFiles.builder(table.spec()).withPath(SEED)
+
.withFileSizeInBytes(1024).withRecordCount(3).build()).commit();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(), "db", "events", new
IcebergWriteContext(
+ operation, false, Collections.emptyMap(), Optional.empty()));
+
catalog.loadTable(IDENTIFIER).newAppend().appendFile(DataFiles.builder(table.spec()).withPath(ADDED)
+
.withFileSizeInBytes(1024).withRecordCount(1).build()).commit();
+ transaction.addCommitData(serialize(fragment(DELETED, 1,
TFileContent.POSITION_DELETES)));
+ // Snapshot isolation permits a concurrent append while still
validating referenced data files.
+ Assertions.assertDoesNotThrow(transaction::commit);
+ }
+
+ @Test
+ @Disabled("Known issue: mixed INSERT and DELETE MERGE actions undercount
affected rows.")
+ public void mergeCountsBothInsertedAndDeletedRows() throws Exception {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog);
+
table.newAppend().appendFile(DataFiles.builder(table.spec()).withPath(SEED)
+
.withFileSizeInBytes(1024).withRecordCount(3).build()).commit();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(), "db", "events", new
IcebergWriteContext(
+ WriteOperation.MERGE, false, Collections.emptyMap(),
Optional.empty()));
+ transaction.addCommitData(serialize(fragment(ADDED, 1,
TFileContent.DATA)));
+ transaction.addCommitData(serialize(fragment(DELETED, 1,
TFileContent.POSITION_DELETES)));
+ transaction.commit();
+ // These actions target disjoint rows; counting both does not
double-count an UPDATE pair.
+ Assertions.assertEquals(2, transaction.getUpdateCnt());
+ }
+
+ private static InMemoryCatalog catalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("repro", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db"));
+ return catalog;
+ }
+
+ private static Table create(InMemoryCatalog catalog) {
+ return catalog.createTable(IDENTIFIER, new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
Collections.singletonMap("format-version", "2"));
+ }
+
+ private static IcebergConnectorTransaction transaction(Table table) {
+ RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
+ operations.table = table;
+ return new IcebergConnectorTransaction(9, operations, new
RecordingConnectorContext());
+ }
+
+ private static TIcebergCommitData fragment(String path, long count,
TFileContent content) {
+ TIcebergCommitData fragment = new TIcebergCommitData();
+ fragment.setFilePath(path);
+ fragment.setFileContent(content);
+ fragment.setFileSize(1024);
+ fragment.setRowCount(count);
+ fragment.setAffectedRows(count);
+ if (content == TFileContent.POSITION_DELETES) {
+ fragment.setReferencedDataFilePath(SEED);
+ }
+ return fragment;
+ }
+
+ private static byte[] serialize(TIcebergCommitData fragment) throws
Exception {
+ return new TSerializer(new
TBinaryProtocol.Factory()).serialize(fragment);
+ }
+
+ private static ConnectorSession session() {
+ return new ConnectorSession() {
+ @Override
+ public String getQueryId() {
+ return "repro";
+ }
+
+ @Override
+ public String getUser() {
+ return "test";
+ }
+
+ @Override
+ public String getTimeZone() {
+ return "UTC";
+ }
+
+ @Override
+ public String getLocale() {
+ return "en_US";
+ }
+
+ @Override
+ public long getCatalogId() {
+ return 1;
+ }
+
+ @Override
+ public String getCatalogName() {
+ return "repro";
+ }
+
+ @Override
+ public Map<String, String> getCatalogProperties() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public <T> T getProperty(String name, Class<T> type) {
+ return null;
+ }
+ };
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
new file mode 100644
index 00000000000..83412d30c0c
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java
@@ -0,0 +1,404 @@
+// 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.doris.connector.spi.ConnectorSession;
+import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.doris.connector.spi.handle.WriteOperation;
+import org.apache.doris.thrift.TFileContent;
+import org.apache.doris.thrift.TIcebergCommitData;
+
+import org.apache.iceberg.DataFile;
+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.Table;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Types;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+public class IcebergWriteTransactionCoverageTest {
+ private static final Schema SCHEMA = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "region", Types.StringType.get()));
+ private static final TableIdentifier IDENTIFIER = TableIdentifier.of("db",
"events");
+ private static final String SEED = "s3://bucket/events/seed.parquet";
+ private static final String ADDED = "s3://bucket/events/added.parquet";
+ private static final String DELETED = "s3://bucket/events/deleted.parquet";
+
+ @Test
+ public void appendV1PreservesHistoricalSnapshot() throws IOException {
+ assertAppendPreservesHistory(1);
+ }
+
+ @Test
+ public void appendV2PreservesHistoricalSnapshot() throws IOException {
+ assertAppendPreservesHistory(2);
+ }
+
+ @Test
+ public void appendV3PreservesHistoricalSnapshot() throws IOException {
+ assertAppendPreservesHistory(3);
+ }
+
+ private void assertAppendPreservesHistory(int formatVersion) throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, formatVersion, false);
+ table.newAppend().appendFile(file(table.spec(), SEED, 2,
null)).commit();
+ long previousSnapshot = table.currentSnapshot().snapshotId();
+
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.INSERT));
+ transaction.addCommitData(serialize(data(ADDED, 3)));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(SEED, ADDED), paths(current, null,
null));
+ Assertions.assertEquals(paths(SEED), paths(current, null,
previousSnapshot));
+ Assertions.assertEquals(3, transaction.getUpdateCnt());
+ Assertions.assertEquals("5",
current.currentSnapshot().summary().get("total-records"));
+ }
+
+ @Test
+ public void dynamicOverwriteReplacesEveryTouchedPartitionOnly() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, true);
+ String west = "s3://bucket/events/west.parquet";
+ String east = "s3://bucket/events/east.parquet";
+ String north = "s3://bucket/events/north.parquet";
+ table.newAppend().appendFile(file(table.spec(), west, 2, "west"))
+ .appendFile(file(table.spec(), east, 3, "east"))
+ .appendFile(file(table.spec(), north, 4, "north")).commit();
+ long previousSnapshot = table.currentSnapshot().snapshotId();
+
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session("dynamic"), "db", "events",
context(WriteOperation.OVERWRITE));
+ TIcebergCommitData westReplacement = data(ADDED, 1);
+ westReplacement.setPartitionValues(Collections.singletonList("west"));
+ TIcebergCommitData eastReplacement = data(SEED, 2);
+ eastReplacement.setPartitionValues(Collections.singletonList("east"));
+ transaction.addCommitData(serialize(westReplacement));
+ transaction.addCommitData(serialize(eastReplacement));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(ADDED, SEED, north), paths(current,
null, null));
+ Assertions.assertEquals(paths(west, east, north), paths(current, null,
previousSnapshot));
+ }
+
+ @Test
+ public void
staticPartitionOverwriteOnBranchPreservesMainAndOtherPartitions() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, true);
+ table.newAppend().appendFile(file(table.spec(), SEED, 2, "west"))
+ .appendFile(file(table.spec(), ADDED, 3, "east")).commit();
+ table.manageSnapshots().createBranch("audit",
table.currentSnapshot().snapshotId()).commit();
+ IcebergWriteSchemaContext schema = IcebergWriteSchemaContext.create(
+ table, table.name(), Optional.of("audit"), false, false);
+ IcebergWriteContext writeContext = new
IcebergWriteContext(WriteOperation.OVERWRITE, true,
+ Collections.singletonMap("region", "west"),
Optional.of("audit"), -1L, schema);
+
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events", writeContext);
+ TIcebergCommitData replacement = data(DELETED, 1);
+ replacement.setPartitionValues(Collections.singletonList("west"));
+ transaction.addCommitData(serialize(replacement));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(SEED, ADDED), paths(current, null,
null));
+ Assertions.assertEquals(paths(DELETED, ADDED), paths(current, "audit",
null));
+ }
+
+ @Test
+ public void branchAppendDoesNotPublishConcurrentMainFilesToBranch() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, false);
+ table.newAppend().appendFile(file(table.spec(), SEED, 1,
null)).commit();
+ table.manageSnapshots().createBranch("audit",
table.currentSnapshot().snapshotId()).commit();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events", new
IcebergWriteContext(
+ WriteOperation.INSERT, false, Collections.emptyMap(),
Optional.of("audit")));
+
catalog.loadTable(IDENTIFIER).newAppend().appendFile(file(table.spec(), ADDED,
1, null)).commit();
+ transaction.addCommitData(serialize(data(DELETED, 1)));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(SEED, ADDED), paths(current, null,
null));
+ Assertions.assertEquals(paths(SEED, DELETED), paths(current, "audit",
null));
+ }
+
+ @Test
+ public void updatePublishesDataAndDeletesInOneSnapshot() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, false);
+ table.newAppend().appendFile(file(table.spec(), SEED, 3,
null)).commit();
+ long previousSnapshot = table.currentSnapshot().snapshotId();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.UPDATE));
+ transaction.addCommitData(serialize(data(ADDED, 2)));
+ transaction.addCommitData(serialize(delete(DELETED, 2, SEED)));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(Long.valueOf(previousSnapshot),
current.currentSnapshot().parentId());
+ Assertions.assertEquals(paths(SEED, ADDED), paths(current, null,
null));
+ Assertions.assertEquals(paths(SEED), paths(current, null,
previousSnapshot));
+ Assertions.assertEquals("1",
current.currentSnapshot().summary().get("added-data-files"));
+ Assertions.assertEquals("1",
current.currentSnapshot().summary().get("added-delete-files"));
+ Assertions.assertEquals(2, transaction.getUpdateCnt());
+ }
+
+ @Test
+ public void mergeWithOnlyInsertActionsCommitsWithoutDeleteFiles() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ create(catalog, 2, false);
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.MERGE));
+ transaction.addCommitData(serialize(data(ADDED, 3)));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(ADDED), paths(current, null, null));
+ Assertions.assertEquals("0",
current.currentSnapshot().summary().get("total-delete-files"));
+ Assertions.assertEquals(3, transaction.getUpdateCnt());
+ }
+
+ @Test
+ public void mergeWithOnlyDeleteActionsPreservesOriginalDataFile() throws
IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, false);
+ table.newAppend().appendFile(file(table.spec(), SEED, 3,
null)).commit();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.MERGE));
+ transaction.addCommitData(serialize(delete(DELETED, 2, SEED)));
+ transaction.commit();
+
+ Table current = catalog.loadTable(IDENTIFIER);
+ Assertions.assertEquals(paths(SEED), paths(current, null, null));
+ Assertions.assertEquals("1",
current.currentSnapshot().summary().get("total-delete-files"));
+ Assertions.assertEquals("1",
current.currentSnapshot().summary().get("total-data-files"));
+ Assertions.assertEquals(2, transaction.getUpdateCnt());
+ }
+
+ @Test
+ public void updateV1IsRejectedBeforeAnySnapshotIsPublished() {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 1, false);
+ IcebergConnectorTransaction transaction = transaction(table);
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.UPDATE)));
+ Assertions.assertTrue(failure.getMessage().contains("format version 2
or higher"));
+ Assertions.assertNull(catalog.loadTable(IDENTIFIER).currentSnapshot());
+ }
+
+ @Test
+ public void malformedDeleteSpecLeavesPublishedSnapshotUntouched() {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, true);
+ table.newAppend().appendFile(file(table.spec(), SEED, 3,
"west")).commit();
+ long previousSnapshot = table.currentSnapshot().snapshotId();
+ IcebergConnectorTransaction transaction =
transaction(catalog.loadTable(IDENTIFIER));
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.DELETE));
+ TIcebergCommitData fragment = delete(DELETED, 1, SEED);
+ fragment.setPartitionSpecId(99);
+ fragment.setPartitionValues(Collections.singletonList("west"));
+ transaction.addCommitData(serialize(fragment));
+
+ DorisConnectorException failure =
Assertions.assertThrows(DorisConnectorException.class, transaction::commit);
+ Assertions.assertTrue(failure.getMessage().contains("Unknown partition
spec id 99"));
+ Assertions.assertEquals(previousSnapshot,
catalog.loadTable(IDENTIFIER).currentSnapshot().snapshotId());
+ }
+
+ @Test
+ public void beginCanRetryAfterCatalogFailureWithoutKeepingPartialState()
throws IOException {
+ InMemoryCatalog catalog = catalog();
+ Table table = create(catalog, 2, false);
+ RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
+ operations.table = table;
+ operations.throwOnLoadTable = true;
+ IcebergConnectorTransaction transaction = new
IcebergConnectorTransaction(
+ 7, operations, new RecordingConnectorContext());
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.INSERT)));
+ Assertions.assertNull(transaction.getTransaction());
+ Assertions.assertNull(transaction.getTable());
+ operations.throwOnLoadTable = false;
+ transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.INSERT));
+ transaction.addCommitData(serialize(data(ADDED, 1)));
+ transaction.commit();
+ Assertions.assertEquals(paths(ADDED),
paths(catalog.loadTable(IDENTIFIER), null, null));
+ }
+
+ @Test
+ public void closeBeforeBeginPreventsCatalogAccess() {
+ RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
+ IcebergConnectorTransaction transaction = new
IcebergConnectorTransaction(
+ 7, operations, new RecordingConnectorContext());
+ transaction.close();
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> transaction.beginWrite(session(null), "db", "events",
context(WriteOperation.INSERT)));
+ Assertions.assertTrue(operations.log.isEmpty());
+ }
+
+ private static InMemoryCatalog catalog() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("coverage", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db"));
+ return catalog;
+ }
+
+ private static Table create(InMemoryCatalog catalog, int version, boolean
partitioned) {
+ PartitionSpec spec = partitioned
+ ? PartitionSpec.builderFor(SCHEMA).identity("region").build()
: PartitionSpec.unpartitioned();
+ return catalog.createTable(IDENTIFIER, SCHEMA, spec,
+ Collections.singletonMap("format-version",
Integer.toString(version)));
+ }
+
+ private static DataFile file(PartitionSpec spec, String path, long rows,
String region) {
+ DataFiles.Builder builder =
DataFiles.builder(spec).withPath(path).withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(1024).withRecordCount(rows);
+ if (region != null) {
+ builder.withPartitionPath("region=" + region);
+ }
+ return builder.build();
+ }
+
+ private static Set<String> paths(String... paths) {
+ return new HashSet<>(Arrays.asList(paths));
+ }
+
+ private static Set<String> paths(Table table, String branch, Long
snapshot) throws IOException {
+ org.apache.iceberg.TableScan scan = table.newScan();
+ if (branch != null) {
+ scan = scan.useRef(branch);
+ }
+ if (snapshot != null) {
+ scan = scan.useSnapshot(snapshot);
+ }
+ Set<String> paths = new HashSet<>();
+ try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
+ for (FileScanTask task : tasks) {
+ paths.add(task.file().path().toString());
+ }
+ }
+ return paths;
+ }
+
+ private static IcebergConnectorTransaction transaction(Table table) {
+ RecordingIcebergCatalogOps operations = new
RecordingIcebergCatalogOps();
+ operations.table = table;
+ return new IcebergConnectorTransaction(7, operations, new
RecordingConnectorContext());
+ }
+
+ private static IcebergWriteContext context(WriteOperation operation) {
+ return new IcebergWriteContext(operation, operation ==
WriteOperation.OVERWRITE,
+ Collections.emptyMap(), Optional.empty());
+ }
+
+ private static TIcebergCommitData data(String path, long rows) {
+ TIcebergCommitData fragment = new TIcebergCommitData();
+ fragment.setFilePath(path);
+ fragment.setFileSize(1024);
+ fragment.setRowCount(rows);
+ fragment.setAffectedRows(rows);
+ fragment.setFileContent(TFileContent.DATA);
+ return fragment;
+ }
+
+ private static TIcebergCommitData delete(String path, long rows, String
referencedDataFile) {
+ TIcebergCommitData fragment = data(path, rows);
+ fragment.setFileContent(TFileContent.POSITION_DELETES);
+ fragment.setReferencedDataFilePath(referencedDataFile);
+ return fragment;
+ }
+
+ private static byte[] serialize(TIcebergCommitData fragment) {
+ try {
+ return new TSerializer(new
TBinaryProtocol.Factory()).serialize(fragment);
+ } catch (TException exception) {
+ throw new AssertionError(exception);
+ }
+ }
+
+ private static ConnectorSession session(String overwriteMode) {
+ Map<String, String> properties = new HashMap<>();
+ if (overwriteMode != null) {
+ properties.put("iceberg_partition_overwrite_mode", overwriteMode);
+ }
+ return new ConnectorSession() {
+ @Override
+ public String getQueryId() {
+ return "coverage";
+ }
+
+ @Override
+ public String getUser() {
+ return "test";
+ }
+
+ @Override
+ public String getTimeZone() {
+ return "UTC";
+ }
+
+ @Override
+ public String getLocale() {
+ return "en_US";
+ }
+
+ @Override
+ public long getCatalogId() {
+ return 1;
+ }
+
+ @Override
+ public String getCatalogName() {
+ return "coverage";
+ }
+
+ @Override
+ public Map<String, String> getCatalogProperties() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public <T> T getProperty(String name, Class<T> type) {
+ return type.cast(properties.get(name));
+ }
+ };
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java
new file mode 100644
index 00000000000..0c3323bd4d0
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java
@@ -0,0 +1,193 @@
+// 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.action;
+
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+public class IcebergSnapshotMaintenanceCoverageTest {
+
+ private static IcebergExpireSnapshotsAction expiration(String key, String
value) {
+ IcebergExpireSnapshotsAction action = new IcebergExpireSnapshotsAction(
+ ImmutableMap.of("older_than", "4102444800000", key, value),
Collections.emptyList(), null);
+ action.validate();
+ return action;
+ }
+
+ @Test
+ public void retainLastPreservesSnapshotsReferencedByTagsAndBranches() {
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ Table table = ActionTestTables.createTable(catalog, "retained_refs");
+ long tagged = ActionTestTables.appendSnapshot(catalog,
"retained_refs", "tagged.parquet", 1);
+ table.refresh();
+ table.manageSnapshots().createTag("audit_tag", tagged).commit();
+ long branched = ActionTestTables.appendSnapshot(catalog,
"retained_refs", "branched.parquet", 1);
+ table.refresh();
+ table.manageSnapshots().createBranch("audit_branch",
branched).commit();
+ long unreferenced = ActionTestTables.appendSnapshot(catalog,
"retained_refs", "expired.parquet", 1);
+ long current = ActionTestTables.appendSnapshot(catalog,
"retained_refs", "current.parquet", 1);
+ table.refresh();
+
+ expiration("retain_last", "1").execute(table,
ActionTestTables.session("UTC"));
+ table.refresh();
+
+ // Retention is evaluated for each live ref; a main-branch cutoff must
not erase pinned history.
+ Assertions.assertNotNull(table.snapshot(tagged));
+ Assertions.assertNotNull(table.snapshot(branched));
+ Assertions.assertNull(table.snapshot(unreferenced));
+ Assertions.assertEquals(current, table.currentSnapshot().snapshotId());
+ Assertions.assertEquals(tagged,
table.refs().get("audit_tag").snapshotId());
+ Assertions.assertEquals(branched,
table.refs().get("audit_branch").snapshotId());
+ }
+
+ @Test
+ public void
explicitExpirationOfTaggedSnapshotFailsWithoutChangingHistory() {
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ Table table = ActionTestTables.createTable(catalog, "protected_tag");
+ long tagged = ActionTestTables.appendSnapshot(catalog,
"protected_tag", "tagged.parquet", 1);
+ table.refresh();
+ table.manageSnapshots().createTag("audit_tag", tagged).commit();
+ long current = ActionTestTables.appendSnapshot(catalog,
"protected_tag", "current.parquet", 1);
+ table.refresh();
+
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> expiration("snapshot_ids", String.valueOf(tagged))
+ .execute(table, ActionTestTables.session("UTC")));
+ table.refresh();
+ Assertions.assertNotNull(table.snapshot(tagged));
+ Assertions.assertEquals(current, table.currentSnapshot().snapshotId());
+ Assertions.assertEquals(tagged,
table.refs().get("audit_tag").snapshotId());
+ }
+
+ @Test
+ public void fastForwardRejectsDivergedBranchesWithoutMovingEitherRef() {
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ Table table = ActionTestTables.createTable(catalog, "diverged_refs");
+ long base = ActionTestTables.appendSnapshot(catalog, "diverged_refs",
"base.parquet", 1);
+ table.refresh();
+ table.manageSnapshots().createBranch("audit_branch", base).commit();
+ long main = ActionTestTables.appendSnapshot(catalog, "diverged_refs",
"main.parquet", 1);
+ table.refresh();
+
table.newAppend().appendFile(ActionTestTables.dataFile("branch.parquet", 1))
+ .toBranch("audit_branch").commit();
+ long branch = table.snapshot("audit_branch").snapshotId();
+ IcebergFastForwardAction action = new IcebergFastForwardAction(
+ ImmutableMap.of("branch", "main", "to", "audit_branch"),
Collections.emptyList(), null);
+ action.validate();
+
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> action.execute(table, ActionTestTables.session("UTC")));
+ table.refresh();
+ Assertions.assertEquals(main, table.currentSnapshot().snapshotId());
+ Assertions.assertEquals(branch,
table.snapshot("audit_branch").snapshotId());
+ }
+
+ @Test
+ public void rollbackRejectsSnapshotOutsideTheCurrentAncestry() {
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ Table table = ActionTestTables.createTable(catalog,
"rollback_ancestry");
+ long base = ActionTestTables.appendSnapshot(catalog,
"rollback_ancestry", "base.parquet", 1);
+ table.refresh();
+ table.manageSnapshots().createBranch("audit_branch", base).commit();
+
table.newAppend().appendFile(ActionTestTables.dataFile("branch.parquet", 1))
+ .toBranch("audit_branch").commit();
+ long branch = table.snapshot("audit_branch").snapshotId();
+ long current = ActionTestTables.appendSnapshot(catalog,
"rollback_ancestry", "main.parquet", 1);
+ table.refresh();
+ IcebergRollbackToSnapshotAction action = new
IcebergRollbackToSnapshotAction(
+ ImmutableMap.of("snapshot_id", String.valueOf(branch)),
Collections.emptyList(), null);
+ action.validate();
+
+ Assertions.assertThrows(DorisConnectorException.class,
+ () -> action.execute(table, ActionTestTables.session("UTC")));
+ table.refresh();
+ Assertions.assertEquals(current, table.currentSnapshot().snapshotId());
+ Assertions.assertEquals(branch,
table.snapshot("audit_branch").snapshotId());
+ }
+
+ @Test
+ @Disabled("Known issue: stale expiration metadata misclassifies newly
committed position delete files.")
+ public void
expirationCountsPositionDeletesCommittedAfterTheTableWasLoaded() {
+ assertLateDeleteClassification(true, true);
+ }
+
+ @Test
+ @Disabled("Known issue: stale expiration metadata misclassifies newly
committed equality delete files.")
+ public void
expirationCountsEqualityDeletesCommittedAfterTheTableWasLoaded() {
+ assertLateDeleteClassification(false, true);
+ }
+
+ @Test
+ public void expirationCountsPositionDeletesFromAFreshTable() {
+ assertLateDeleteClassification(true, false);
+ }
+
+ @Test
+ public void expirationCountsEqualityDeletesFromAFreshTable() {
+ assertLateDeleteClassification(false, false);
+ }
+
+ private static void assertLateDeleteClassification(boolean positionDelete,
boolean staleTable) {
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ ActionTestTables.createTable(catalog, "late_delete");
+ ActionTestTables.appendSnapshot(catalog, "late_delete",
"base.parquet", 2);
+ Table loadedBeforeDelete =
catalog.loadTable(ActionTestTables.id("late_delete"));
+ FileMetadata.Builder builder =
FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned());
+ if (positionDelete) {
+
builder.ofPositionDeletes().withReferencedDataFile("s3://b/db1/base.parquet");
+ } else {
+ builder.ofEqualityDeletes(1);
+ }
+ DeleteFile delete = builder.withPath("s3://b/db1/late-delete.parquet")
+
.withFormat(FileFormat.PARQUET).withFileSizeInBytes(64).withRecordCount(1).build();
+ Table writer = catalog.loadTable(ActionTestTables.id("late_delete"));
+ writer.newRowDelta().addDeletes(delete).commit();
+ writer.newRowDelta().removeDeletes(delete).commit();
+ ActionTestTables.appendSnapshot(catalog, "late_delete",
"final.parquet", 1);
+ if (!staleTable) {
+ loadedBeforeDelete.refresh();
+ }
+
+ // SDK expiration refreshes metadata. Its counters must classify
delete files from that same generation.
+ List<String> row = expiration("retain_last", "1")
+ .execute(loadedBeforeDelete,
ActionTestTables.session("UTC")).getRows().get(0);
+
+ Assertions.assertEquals(1,
Iterables.size(catalog.loadTable(ActionTestTables.id("late_delete")).snapshots()),
+ "the append and delete-removal history must actually expire");
+ Assertions.assertEquals("0", row.get(0),
+ "data files remain reachable; stale=" + staleTable + ",
counters=" + row);
+ Assertions.assertEquals(positionDelete ? "1" : "0", row.get(1),
+ "position delete classification; stale=" + staleTable + ",
counters=" + row);
+ Assertions.assertEquals(positionDelete ? "0" : "1", row.get(2),
+ "equality delete classification; stale=" + staleTable + ",
counters=" + row);
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/ICEBERG_CAPABILITY_TEST_COVERAGE.md
b/regression-test/suites/external_table_p0/iceberg/ICEBERG_CAPABILITY_TEST_COVERAGE.md
new file mode 100644
index 00000000000..15c952f8a8d
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/ICEBERG_CAPABILITY_TEST_COVERAGE.md
@@ -0,0 +1,160 @@
+# Iceberg capability and test coverage inventory
+
+This inventory maps the [published capability
matrix](https://doris.apache.org/docs/dev/lakehouse/best-practices/doris-iceberg-capability-matrix/)
+and [catalog
guide](https://doris.apache.org/docs/dev/lakehouse/catalogs/iceberg-catalog/)
+to tests in this repository. Test presence does not imply that every backend,
format, version,
+scanner, and deployment combination has been executed.
+
+## Scope
+
+- Iceberg format V1/V2/V3, the bundled Iceberg SDK version, and Doris release
versions are separate dimensions.
+- Iceberg has no Doris Duplicate/Unique/Aggregate target model. Those are
INSERT SELECT source
+ models; see `write/test_iceberg_write_source_models.groovy`. Iceberg target
dimensions include
+ format version, partition/sort specs, delete mode, data format, and branch.
+- A recording fake checks routing and errors; it does not validate a live
catalog or filesystem.
+- `Y`/`N` below transcribe the published matrix in REST/HMS/Hadoop/Glue/DLF/S3
Tables/JDBC order.
+ They are documentation input, not new executable promises. Current
implementation takes precedence.
+- The published matrix lags code for table-property mutation and
`remove_orphan_files`, which
+ have implementation and tests despite all-N rows. The catalog guide's
backend summary differs
+ from the detailed matrix too. Documentation reconciliation remains necessary.
+
+## Published operations
+
+`Partial` means representative tests exist, with unvalidated combinations
remaining.
+`Negative gap` means the published operation is unsupported and exhaustive
rejection coverage
+is absent. Evidence on those rows is related routing/contract coverage, not
successful execution.
+
+| Category | Operation | Published backend support | Assessment | Unit
evidence | SQL evidence |
+| --- | --- | --- | --- | --- | --- |
+| Namespace / Database | `LIST_NAMESPACES` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `LOAD_NAMESPACE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `NAMESPACE_EXISTS` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `CREATE_NAMESPACE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `ALTER_NAMESPACE_PROPERTIES` | N/N/N/N/N/N/N |
Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `DROP_NAMESPACE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `DROP_NAMESPACE_CASCADE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Namespace / Database | `MULTILEVEL_NAMESPACE` | Y/N/N/N/N/N/N | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_drop_db.groovy](test_iceberg_drop_db.groovy) |
+| Table lifecycle | `LIST_TABLES` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `LOAD_TABLE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `TABLE_EXISTS` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `CREATE_TABLE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `CREATE_TABLE_WITH_LOCATION` | N/N/N/N/N/N/N | Negative
gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `CTAS` | Y/Y/Y/Y/N/Y/Y | Partial |
[IcebergCreateTableValidationTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergCreateTableValidationTest.java)
|
[write/test_iceberg_write_ctas_format_boundary.groovy](write/test_iceberg_write_ctas_format_boundary.groovy)
|
+| Table lifecycle | `ATOMIC_CTAS` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `REGISTER_TABLE` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `REPLACE_TABLE` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `DROP_TABLE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `PURGE_TABLE` | Y/Y/Y/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `RENAME_TABLE` | Y/Y/N/Y/N/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Table lifecycle | `SET_TABLE_LOCATION` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[write/test_iceberg_create_table.groovy](write/test_iceberg_create_table.groovy)
|
+| Schema and layout evolution | `ADD_COLUMN` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `DROP_COLUMN` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `RENAME_COLUMN` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `ALTER_COLUMN_TYPE` | Y/Y/Y/Y/Y/Y/Y | Partial
|
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `ALTER_COLUMN_NULLABILITY` | Y/Y/Y/Y/Y/Y/Y |
Partial |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `SET_TABLE_PROPERTIES` | N/N/N/N/N/N/N |
Documentation differs |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
| [test_iceberg_alter_properties.groovy](test_iceberg_alter_properties.groovy)
|
+| Schema and layout evolution | `UNSET_TABLE_PROPERTIES` | N/N/N/N/N/N/N |
Negative gap |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `EVOLVE_PARTITION_SPEC` | Y/Y/Y/Y/Y/Y/Y |
Partial |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Schema and layout evolution | `EVOLVE_SORT_ORDER` | N/N/N/N/N/N/N | Negative
gap |
[CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| View | `LIST_VIEWS` | Y/Y/N/N/N/N/Y | Partial |
[IcebergConnectorMetadataTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `LOAD_VIEW` | Y/Y/N/N/N/N/Y | Partial |
[IcebergConnectorMetadataTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `READ_VIEW` | Y/Y/N/N/N/N/Y | Partial |
[IcebergConnectorMetadataTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `CREATE_VIEW` | N/N/N/N/N/N/N | Negative gap |
[IcebergJdbcViewProbeTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `REPLACE_VIEW` | N/N/N/N/N/N/N | Negative gap |
[IcebergJdbcViewProbeTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `ALTER_VIEW` | N/N/N/N/N/N/N | Negative gap |
[IcebergJdbcViewProbeTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `RENAME_VIEW` | N/N/N/N/N/N/N | Negative gap |
[IcebergJdbcViewProbeTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| View | `DROP_VIEW` | Y/Y/N/N/N/N/Y | Partial |
[IcebergJdbcViewProbeTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java)
| [test_iceberg_view_query_p0.groovy](test_iceberg_view_query_p0.groovy) |
+| Data write | `APPEND` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_insert.groovy](write/test_iceberg_write_insert.groovy)
|
+| Data write | `OVERWRITE_TABLE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_partition_overwrite_mode.groovy](write/test_iceberg_partition_overwrite_mode.groovy)
|
+| Data write | `OVERWRITE_PARTITION` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_static_partition_overwrite.groovy](write/test_iceberg_static_partition_overwrite.groovy)
|
+| Data write | `DYNAMIC_PARTITION_OVERWRITE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_overwrite_evolution.groovy](write/test_iceberg_write_overwrite_evolution.groovy)
|
+| Data write | `UPDATE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[dml/test_iceberg_update_delete_advanced.groovy](dml/test_iceberg_update_delete_advanced.groovy)
|
+| Data write | `DELETE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergMetadataDeleteTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergMetadataDeleteTest.java)
|
[dml/test_iceberg_update_delete_basic.groovy](dml/test_iceberg_update_delete_basic.groovy)
|
+| Data write | `MERGE_INTO` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_merge_semantics.groovy](write/test_iceberg_write_merge_semantics.groovy)
|
+| Data write | `UPSERT` | N/N/N/N/N/N/N | Negative gap |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_dml_modes_evolution.groovy](write/test_iceberg_write_dml_modes_evolution.groovy)
|
+| Data write | `PARTIAL_UPDATE` | N/N/N/N/N/N/N | Negative gap |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_dml_modes_evolution.groovy](write/test_iceberg_write_dml_modes_evolution.groovy)
|
+| Data write | `WRITE_TO_BRANCH` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_evolution_refs.groovy](write/test_iceberg_write_evolution_refs.groovy)
|
+| Data write | `SCHEMA_EVOLUTION_ON_WRITE` | N/N/N/N/N/N/N | Negative gap |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_dml_modes_evolution.groovy](write/test_iceberg_write_dml_modes_evolution.groovy)
|
+| Transactions and concurrency | `ATOMIC_SINGLE_TABLE_COMMIT` | Y/Y/Y/Y/Y/Y/Y
| Partial |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `MULTI_TABLE_COMMIT` | N/N/N/N/N/N/N |
Negative gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `STAGED_CREATE` | N/N/N/N/N/N/N | Negative
gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `OPTIMISTIC_CONCURRENCY` | Y/Y/Y/Y/Y/Y/Y |
Partial |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `COMMIT_CONFLICT_DETECTION` | Y/Y/Y/Y/Y/Y/Y |
Partial |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `COMMIT_RETRY` | N/N/N/N/N/N/N | Negative gap
|
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `COMMIT_IDEMPOTENCY` | N/N/N/N/N/N/N |
Negative gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `COMMIT_STATUS_RECOVERY` | N/N/N/N/N/N/N |
Negative gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `LOCK_REQUIRED` | N/N/N/N/N/N/N | Negative
gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `LOCK_CONFIGURED` | N/N/N/N/N/N/N | Negative
gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `SERIALIZABLE_ISOLATION` | Y/Y/Y/Y/Y/Y/Y |
Partial |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Transactions and concurrency | `FAILED_WRITE_CLEANUP` | N/N/N/N/N/N/N |
Negative gap |
[IcebergConnectorTransactionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTransactionTest.java)
|
[write/test_iceberg_write_concurrent_merge_invariants.groovy](write/test_iceberg_write_concurrent_merge_invariants.groovy)
|
+| Branch / Tag | `BRANCH_CREATE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Branch / Tag | `BRANCH_DROP` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Branch / Tag | `READ_BY_BRANCH` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotSchemaCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java)
| [iceberg_query_tag_branch.groovy](iceberg_query_tag_branch.groovy) |
+| Branch / Tag | `BRANCH_WRITE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergWriteTransactionCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriteTransactionCoverageTest.java)
|
[write/test_iceberg_write_branch_dml_boundary.groovy](write/test_iceberg_write_branch_dml_boundary.groovy)
|
+| Branch / Tag | `BRANCH_FAST_FORWARD` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotMaintenanceCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java)
|
[branch_tag/iceberg_branch_cross_operations.groovy](branch_tag/iceberg_branch_cross_operations.groovy)
|
+| Branch / Tag | `TAG_CREATE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Branch / Tag | `TAG_DROP` | Y/Y/Y/Y/Y/Y/Y | Partial |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Branch / Tag | `READ_BY_TAG` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotSchemaCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java)
| [iceberg_query_tag_branch.groovy](iceberg_query_tag_branch.groovy) |
+| Branch / Tag | `TAG_WRITE` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Branch / Tag | `TAG_FAST_FORWARD` | N/N/N/N/N/N/N | Negative gap |
[CatalogBackedIcebergCatalogOpsDdlTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsDdlTest.java)
|
[test_iceberg_schema_ref_actions_matrix.groovy](test_iceberg_schema_ref_actions_matrix.groovy)
|
+| Snapshot | `READ_BY_VERSION` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotSchemaCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSnapshotSchemaCoverageTest.java)
|
[test_iceberg_schema_time_travel_matrix.groovy](test_iceberg_schema_time_travel_matrix.groovy)
|
+| Snapshot | `READ_BY_TIME` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergTimeUtilsTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTimeUtilsTest.java)
| [test_iceberg_time_travel.groovy](test_iceberg_time_travel.groovy) |
+| Snapshot | `ROLLBACK` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotMaintenanceCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java)
|
[action/test_iceberg_execute_actions.groovy](action/test_iceberg_execute_actions.groovy)
|
+| Snapshot | `EXPIRE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergSnapshotMaintenanceCoverageTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergSnapshotMaintenanceCoverageTest.java)
|
[test_iceberg_ref_retention_expiration_control.groovy](test_iceberg_ref_retention_expiration_control.groovy)
|
+| Metadata Table | `READ_METADATA_TABLE` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergConnectorMetadataSysTableTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataSysTableTest.java)
|
[test_iceberg_system_table_projection.groovy](test_iceberg_system_table_projection.groovy)
|
+| Maintenance | `REMOVE_ORPHAN_FILES` | N/N/N/N/N/N/N | Documentation differs
|
[IcebergRemoveOrphanFilesActionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRemoveOrphanFilesActionTest.java)
|
[action/test_iceberg_execute_actions.groovy](action/test_iceberg_execute_actions.groovy)
|
+| Maintenance | `REWRITE_DATA_FILES / COMPACT` | Y/Y/Y/Y/Y/Y/Y | Partial |
[RewriteDataFilePlannerTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/rewrite/RewriteDataFilePlannerTest.java)
|
[action/test_iceberg_v3_row_lineage_rewrite_data_files.groovy](action/test_iceberg_v3_row_lineage_rewrite_data_files.groovy)
|
+| Maintenance | `REWRITE_MANIFESTS` | Y/Y/Y/Y/Y/Y/Y | Partial |
[IcebergRewriteManifestsActionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergRewriteManifestsActionTest.java)
|
[action/test_iceberg_rewrite_manifests.groovy](action/test_iceberg_rewrite_manifests.groovy)
|
+| Maintenance | `REWRITE_DELETE_FILES` | N/N/N/N/N/N/N | Negative gap |
[IcebergExpireSnapshotsActionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java)
|
[action/test_iceberg_execute_actions.groovy](action/test_iceberg_execute_actions.groovy)
|
+| Maintenance | `EXPIRE_CHANGELOGS` | N/N/N/N/N/N/N | Negative gap |
[IcebergExpireSnapshotsActionTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java)
|
[action/test_iceberg_execute_actions.groovy](action/test_iceberg_execute_actions.groovy)
|
+| Data ingestion | `ADD_FILES` | N/N/N/N/N/N/N | Negative gap |
[IcebergConnectorTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTest.java)
|
[test_external_catalog_iceberg_common.groovy](test_external_catalog_iceberg_common.groovy)
|
+| Data ingestion | `MIGRATE_TABLE` | N/N/N/N/N/N/N | Negative gap |
[IcebergConnectorTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTest.java)
|
[test_external_catalog_iceberg_common.groovy](test_external_catalog_iceberg_common.groovy)
|
+| Data ingestion | `COMPUTE_TABLE_STATS` | N/N/N/N/N/N/N | Negative gap |
[IcebergConnectorTest.java](../../../../fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorTest.java)
|
[test_external_catalog_iceberg_common.groovy](test_external_catalog_iceberg_common.groovy)
|
+
+## Read, model, version, lineage, and data-type dimensions
+
+| Dimension | Existing evidence | Added coverage / remaining limit |
+| --- | --- | --- |
+| V1/V2/V3 and delete representations |
`write/test_iceberg_write_v1_void_partition.groovy`,
`test_iceberg_equality_delete.groovy`, `test_iceberg_deletion_vector.groovy` |
New transaction append preserves historical files in all three versions.
Position/Equality/DV ordering and concurrent rewrite/failure combinations
remain partial. |
+| Source models and COW/MOR | `write/test_iceberg_write_source_models.groovy`,
`write/test_iceberg_write_dml_modes_evolution.groovy` | Final rows alone cannot
validate affected-row protocol or per-operation isolation properties. |
+| Parquet, ORC, mixed files | `test_iceberg_mixed_file_formats.groovy`,
`test_iceberg_partition_evolution_format_scanner.groovy` | New
metadata/delete/history suite crosses V2/V3 with Parquet/ORC; unsupported AVRO
native data reads are not a positive-test promise. |
+| Branch versus tag/snapshot schema |
`test_iceberg_schema_dual_relation_matrix.groovy`,
`test_iceberg_schema_ref_actions_matrix.groovy` | New unit puts branch and tag
at the same old snapshot after schema-only evolution; tests explicit snapshot
after rollback. |
+| V3 row lineage | `test_iceberg_v3_row_lineage_uniqueness_stability.groovy`,
`test_iceberg_v3_row_lineage_schema_evolution.groovy`,
`action/test_iceberg_v3_row_lineage_rewrite_data_files.groovy` | New V2-to-V3
schema-only transition and ADD/RENAME reserved-name atomicity checks. Overflow
and mixed inherited/physical lineage remain partial. |
+| File metadata and TopN | `test_iceberg_file_metadata_columns.groovy` | New
`test_iceberg_file_metadata_delete_history.groovy` checks physical position
holes after deletes, partition predicates, historical refs and
eager/TopN-setting parity. |
+| Boolean/integer/long/float/double | `IcebergTypeMappingReadTest`,
`test_iceberg_float_predicate_pushdown.groovy` | New evaluator oracles test
integer narrowing and long limits in SCAN/CONFLICT/REWRITE modes. A separate
disabled reproduction preserves the DECIMAL-to-DOUBLE cast pruning invariant. |
+| Decimal precision/scale | `write/test_iceberg_write_decimal_bounds.groovy`,
`IcebergPredicateConverterTest` | New precision 1/9/18/28/38 roundtrips and
exact 38-digit comparison oracle. Manifest bound encoding needs separate
reader/writer tests. |
+| Dates, micro/nanoseconds, timestamp zones |
`test_iceberg_timestamp_tz.groovy`, `test_iceberg_timestamp_ns.groovy`,
`write/test_iceberg_write_timestamp_partitions.groovy` | New
negative-epoch/leap-century partition range oracle uses SDK transforms; DST
overlap/gap filtering still needs managed execution. |
+| Binary/fixed/UUID | `test_iceberg_uuid_binary_compatibility.groovy`,
`test_iceberg_varbinary.groovy`,
`write/test_iceberg_write_binary_partitions.groovy` | New compound partition
JSON covers binary values; fixed/binary serialization preserves
direct/read-only buffer positions. |
+| List/map/struct and nested evolution |
`test_iceberg_nested_schema_evolution_spark_doris_interop.groovy`,
`write/test_iceberg_write_complex_evolution.groovy` | New deep roundtrip checks
field IDs, requiredness, comments and supported siblings beside unsupported
leaves. |
+| Initial/write defaults and Variant | `test_iceberg_initial_defaults.groovy`,
`test_iceberg_variant_defaults.groovy`, `test_iceberg_variant_read.groovy` |
Existing detailed reader tests do not establish all catalog, encryption and
scanner combinations. |
+| Identity/bucket/truncate/calendar/void partitions |
`test_iceberg_transform_partitions.groovy`,
`write/test_iceberg_write_v1_void_partition.groovy`,
`test_iceberg_partition_evolution_query_write.groovy` | New compound wire and
SDK calendar oracles; transform/type/scanner cross-product remains partial. |
+| Metadata tables/auth/projections |
`test_iceberg_system_table_projection.groovy`,
`test_iceberg_sys_table_auth.groovy` | New unit enumerates every exposed
metadata table on a real V3 base and checks absence of data-only hidden
columns. |
+| Cache/REFRESH/statistics/count pushdown |
`test_iceberg_table_meta_cache.groovy`, `test_iceberg_statistics.groovy`,
`test_iceberg_optimize_count.groovy` | Concurrent metadata generation, TTL
changes and multi-FE failover require stress execution. |
+| Native/JNI/streaming/Arrow/REST planning |
`test_iceberg_rest_server_planning_guard.groovy`,
`test_iceberg_arrow_flight_split_source.groovy`, scanner unit tests | REST
server scan planning is explicitly rejected. Default settings do not prove
alternate scanner/planning paths ran. |
+| Catalogs/filesystems/credentials | `IcebergAuthenticatedFileIOTest`,
`IcebergProcedureOpsKerberosIoTest`, external P2 catalog/storage suites |
Recording fakes do not validate live HMS/Glue/DLF/S3 Tables/JDBC/Hadoop,
credential vending, HDFS or all object stores. |
+| Cleanup/retry/conflicts/resources |
`write/test_iceberg_failed_write_atomicity_retry.groovy`,
`IcebergCatalogResourceTrackerTest`, `IcebergConnectorWorkerPoolPinTest` | New
begin-failure retry and close-before-begin tests. Unknown commit status, FE
restart/replay, cancellation, spill/OOM and outages remain partial. |
+
+## Oracle gaps addressed
+
+Existing branch/tag retention suites expire newly created snapshots with a
cutoff in 2020.
+Those snapshots are not expiration candidates, so preserved refs alone cannot
prove protection.
+`test_iceberg_ref_retention_expiration_control.groovy` adds a future cutoff,
`retain_last=1`,
+and an unreferenced snapshot that must disappear while protected
main/branch/tag snapshots stay readable.
+The equivalent unit uses real in-memory Iceberg metadata.
+
+New transaction tests verify published file sets and historical snapshot
contents, rather than
+only API calls. They cover multi-partition dynamic overwrite, static overwrite
on a branch,
+concurrent main/branch append, and atomic publication of update data plus
delete files.
+
+Known-bug reproductions retain correct assertions and are individually
disabled with a reason.
+A skipped test is not passing coverage. Enable them when validating a fix by
adding
+`-Djunit.jupiter.conditions.deactivate=org.junit.*DisabledCondition` to the
Maven test command.
+
+## Execution boundaries
+
+Added FE tests use actual Iceberg metadata operations and recording connector
contexts.
+The Groovy suites use existing REST/MinIO/Spark fixtures with metamorphic and
cross-engine
+comparisons, so they do not require handwritten golden results. SQL product
execution needs
+the managed test environment. Offline unit success and Groovy compilation do
not imply SQL
+execution, BE unit success, a release compatibility matrix, or full backend
coverage.
+Actual commands and results belong in the PR validation section.
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_delete_history.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_delete_history.groovy
new file mode 100644
index 00000000000..4380c05ff83
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_file_metadata_delete_history.groovy
@@ -0,0 +1,110 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_file_metadata_delete_history",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String host = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalog = "iceberg_file_metadata_delete_history"
+ String database = "iceberg_file_metadata_delete_history_db"
+ def lazyThreshold = sql("SELECT
@@topn_lazy_materialization_threshold")[0][0]
+
+ sql "DROP CATALOG IF EXISTS ${catalog}"
+ sql """CREATE CATALOG ${catalog} PROPERTIES (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest',
+ 'uri' = 'http://${host}:${restPort}',
+ 's3.endpoint' = 'http://${host}:${minioPort}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.region' = 'us-east-1')"""
+ sql "SWITCH ${catalog}"
+ sql "CREATE DATABASE IF NOT EXISTS ${database}"
+ sql "USE ${database}"
+
+ try {
+ for (int version : [2, 3]) {
+ for (String format : ["parquet", "orc"]) {
+ String table = "metadata_v${version}_${format}"
+ spark_iceberg "DROP TABLE IF EXISTS demo.${database}.${table}"
+ spark_iceberg """CREATE TABLE demo.${database}.${table} (
+ id INT, batch INT, payload STRING) USING iceberg
PARTITIONED BY (batch)
+ TBLPROPERTIES ('format-version' = '${version}',
+ 'write.format.default' = '${format}',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.distribution-mode' = 'none')"""
+ // A single writer keeps multiple rows in each file, so
deleting two rows really
+ // exercises holes in physical positions rather than removing
one-row files.
+ spark_iceberg """INSERT INTO demo.${database}.${table}
+ SELECT /*+ COALESCE(1) */ id, batch, payload FROM VALUES
+ (1, 1, 'a'), (2, 1, 'b'), (3, 1, 'c'), (4, 1, 'd') AS
v(id, batch, payload)"""
+ spark_iceberg """INSERT INTO demo.${database}.${table}
+ SELECT /*+ COALESCE(1) */ id, batch, payload FROM VALUES
+ (5, 2, 'e'), (6, 2, 'f'), (7, 2, 'g'), (8, 2, 'h') AS
v(id, batch, payload)"""
+ sql "REFRESH DATABASE ${database}"
+ sql "SET topn_lazy_materialization_threshold = -1"
+ def before = sql("""SELECT id, batch, payload, `_file`, `_pos`
+ FROM ${table} ORDER BY id""")
+ def snapshot = sql("""SELECT snapshot_id FROM ${table}\$refs
WHERE name = 'main'""")[0][0]
+ sql "ALTER TABLE ${table} CREATE TAG before_delete AS OF
VERSION ${snapshot}"
+
+ // Deletes leave holes in physical positions. A filtered-row
ordinal would silently
+ // change the identity of a surviving row, even when its
visible values look correct.
+ sql "DELETE FROM ${table} WHERE id IN (2, 4)"
+ def survivors = before.findAll {
!(it[0].toString().toInteger() in [2, 4]) }
+ assertEquals(survivors, sql("""SELECT id, batch, payload,
`_file`, `_pos`
+ FROM ${table} ORDER BY id"""))
+ spark_iceberg "REFRESH TABLE demo.${database}.${table}"
+ assertSparkDorisResultEquals(
+ spark_iceberg("SELECT id, batch, payload FROM
demo.${database}.${table} ORDER BY id"),
+ sql("SELECT id, batch, payload FROM ${table} ORDER BY
id"))
+
+ // Reusing the current scan's delete state for a historical
relation loses rows.
+ for (String relation : ["${table} FOR VERSION AS OF
${snapshot}",
+ "${table}@tag(before_delete)"]) {
+ assertEquals(before, sql("""SELECT id, batch, payload,
`_file`, `_pos`
+ FROM ${relation} ORDER BY id"""))
+ }
+
+ // Phase two must reconstruct metadata from physical file
coordinates after deletes,
+ // including when the predicate column is absent from the
final projection.
+ for (String projection : ["`_file`", "`_pos`", "`_file`,
`_pos`",
+ "payload, `_file`, `_pos`"]) {
+ String query = """SELECT ${projection} FROM ${table}
+ WHERE id > 1 ORDER BY id LIMIT 3"""
+ sql "SET topn_lazy_materialization_threshold = -1"
+ def eager = sql(query)
+ sql "SET topn_lazy_materialization_threshold = 10"
+ assertEquals(eager, sql(query))
+ }
+
+ // An identity-partition predicate must preserve both file
coordinates and the
+ // delete mask; it must not synthesize positions relative to
the selected partition.
+ sql "SET topn_lazy_materialization_threshold = -1"
+ assertEquals(survivors.findAll { it[1].toString() == "1" },
+ sql("""SELECT id, batch, payload, `_file`, `_pos`
+ FROM ${table} WHERE batch = 1 ORDER BY id"""))
+ }
+ }
+ } finally {
+ sql "SET topn_lazy_materialization_threshold = ${lazyThreshold}"
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_ref_retention_expiration_control.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_ref_retention_expiration_control.groovy
new file mode 100644
index 00000000000..16fe286142b
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_ref_retention_expiration_control.groovy
@@ -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.
+
+suite("test_iceberg_ref_retention_expiration_control",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ logger.info("Iceberg test is disabled")
+ return
+ }
+ String host = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ sql "DROP CATALOG IF EXISTS iceberg_ref_retention_control"
+ sql """CREATE CATALOG iceberg_ref_retention_control PROPERTIES (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest',
+ 'uri' = 'http://${host}:${restPort}',
+ 's3.endpoint' = 'http://${host}:${minioPort}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.region' = 'us-east-1')"""
+ sql "SWITCH iceberg_ref_retention_control"
+ sql "CREATE DATABASE IF NOT EXISTS iceberg_ref_retention_control_db"
+ sql "USE iceberg_ref_retention_control_db"
+ sql "DROP TABLE IF EXISTS retention_control"
+ sql "CREATE TABLE retention_control (id INT, payload STRING) PROPERTIES
('format-version' = '2')"
+
+ def currentSnapshot = {
+ sql("SELECT snapshot_id FROM retention_control\$refs WHERE name =
'main'")[0][0].toString()
+ }
+ sql "INSERT INTO retention_control VALUES (1, 'tagged')"
+ String tagSnapshot = currentSnapshot()
+ sql "ALTER TABLE retention_control CREATE TAG protected_tag AS OF VERSION
${tagSnapshot}"
+ def tagRows = sql("SELECT * FROM retention_control@tag(protected_tag)
ORDER BY id")
+ sql "INSERT INTO retention_control VALUES (2, 'branched')"
+ String branchSnapshot = currentSnapshot()
+ sql """ALTER TABLE retention_control CREATE BRANCH protected_branch
+ AS OF VERSION ${branchSnapshot} WITH SNAPSHOT RETENTION 1 SNAPSHOTS"""
+ def branchRows = sql("SELECT * FROM
retention_control@branch(protected_branch) ORDER BY id")
+ sql "INSERT INTO retention_control VALUES (3, 'unreferenced')"
+ String unreferencedSnapshot = currentSnapshot()
+ sql "INSERT INTO retention_control VALUES (4, 'current')"
+ String mainSnapshot = currentSnapshot()
+ def mainRows = sql("SELECT * FROM retention_control ORDER BY id")
+ def refsBefore = sql("SELECT name, type, snapshot_id FROM
retention_control\$refs ORDER BY name")
+
+ // A cutoff older than the freshly written snapshots cannot test retention
protection. The
+ // unreferenced snapshot is a positive control proving that this call
actually expires history.
+ sql """ALTER TABLE retention_control EXECUTE expire_snapshots(
+ 'older_than' = '2100-01-01T00:00:00', 'retain_last' = '1')"""
+ def retained = sql("SELECT snapshot_id FROM retention_control\$snapshots
ORDER BY snapshot_id")
+ .collect { it[0].toString() }
+ assertEquals([tagSnapshot, branchSnapshot, mainSnapshot].sort(),
retained.sort())
+ assertFalse(retained.contains(unreferencedSnapshot))
+ assertEquals(refsBefore,
+ sql("SELECT name, type, snapshot_id FROM retention_control\$refs
ORDER BY name"))
+ assertEquals(tagRows, sql("SELECT * FROM
retention_control@tag(protected_tag) ORDER BY id"))
+ assertEquals(branchRows, sql("SELECT * FROM
retention_control@branch(protected_branch) ORDER BY id"))
+ assertEquals(mainRows, sql("SELECT * FROM retention_control ORDER BY id"))
+ spark_iceberg "REFRESH TABLE
demo.iceberg_ref_retention_control_db.retention_control"
+ assertSparkDorisResultEquals(
+ spark_iceberg("SELECT * FROM
demo.iceberg_ref_retention_control_db.retention_control ORDER BY id"),
+ mainRows)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
