This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new c150be2456 [#14438] fix(iceberg-rest): validate commits and fix
pagination and purge (#14530)
c150be2456 is described below
commit c150be2456be42a61660ed046912784c9f64867d
Author: roryqi <[email protected]>
AuthorDate: Mon Nov 2 19:32:20 2026 +0800
[#14438] fix(iceberg-rest): validate commits and fix pagination and purge
(#14530)
### What changes were proposed in this pull request?
Integrate the fixes from #14439, #14455, #14458, and #14466, preserving
the original authors' contributions.
- Validate namespace names, registration UUIDs, snapshots, schema
evolution, partition field IDs, referenced specs, and default transforms
before committing. Preserve staged-create requirements and validate the
metadata built on each retry. Accept partition specs without a
client-supplied `spec-id`.
- Propagate unexpected import errors without automatically deleting
tables or namespaces. Keep event-copy failures from failing completed
requests.
- Return all list results when `pageToken` is absent, even with
`pageSize`; an empty token starts pagination. Continue rejecting invalid
page sizes.
- Serialize direct and staged table creation with namespace drop on one
server. For an explicit purge, use the table's FileIO to clean up
reachable files when a manifest list is missing, preserving content
files when `gc.enabled=false`.
### Why are the changes needed?
Invalid commits can leave unreadable metadata or silently remove refs.
List requests can be truncated without opting into pagination,
concurrent creates and namespace drops can leave inconsistent ownership,
and a missing manifest list can stop purge cleanup. Automatic deletion
after an import failure can race with other clients.
Fixes #14438
Fixes #14454
Fixes #14457
Fixes #14465
### Does this PR introduce _any_ user-facing change?
Yes. Invalid metadata changes return 400, or 409 for commit conflicts.
Lists without a page token return all results. Import failures do not
trigger deletion, and explicit purge continues cleaning up reachable
files while respecting GC settings.
### How was this patch tested?
- Formatting: `:iceberg:iceberg-common:spotlessApply` and
`:iceberg:iceberg-rest-server:spotlessApply`.
- `./gradlew :iceberg:iceberg-common:test
:iceberg:iceberg-rest-server:test -PskipITs` — 88 common tests and 531
REST server tests passed, with no failures or skips.
- Regression coverage includes HTTP pagination for
namespaces/tables/views, invalid page sizes without a token,
staged-create validation and locking, schema/ref validation bypasses,
partition ID collisions within one commit, non-default unknown
transforms, and purge with a missing manifest list with GC enabled and
disabled.
- `git diff --check`.
---------
Co-authored-by: Mark Hoerth <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../iceberg/common/ops/IcebergCatalogWrapper.java | 27 +-
.../iceberg/common/ops/IcebergPurgeUtil.java | 124 +++++
.../common/ops/TestIcebergCatalogWrapper.java | 54 ++
.../iceberg/service/CatalogWrapperForREST.java | 60 +-
.../iceberg/service/IcebergCommitValidation.java | 446 +++++++++++++++
.../iceberg/service/IcebergObjectMapper.java | 56 ++
.../iceberg/service/IcebergRESTUtils.java | 23 +-
.../dispatcher/IcebergNamespaceHookDispatcher.java | 34 +-
.../dispatcher/IcebergTableHookDispatcher.java | 25 +-
.../service/rest/IcebergPaginationHelper.java | 30 +-
.../iceberg/service/TestCatalogWrapperForREST.java | 61 ++
.../service/TestIcebergCommitValidation.java | 613 +++++++++++++++++++++
.../iceberg/service/TestIcebergRESTUtils.java | 17 +
.../TestIcebergNamespaceHookDispatcher.java | 45 ++
.../dispatcher/TestIcebergTableHookDispatcher.java | 100 ++++
.../rest/TestIcebergNamespaceOperations.java | 17 +-
.../service/rest/TestIcebergPaginationHelper.java | 81 ++-
.../service/rest/TestIcebergTableOperations.java | 15 +-
.../service/rest/TestIcebergViewOperations.java | 15 +-
19 files changed, 1808 insertions(+), 35 deletions(-)
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
index cc69b8eb4d..c36cc09d3e 100644
---
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java
@@ -41,11 +41,13 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.iceberg.BaseMetadataTable;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.Catalog;
@@ -280,7 +282,30 @@ public class IcebergCatalogWrapper implements
AutoCloseable {
public void purgeTable(TableIdentifier tableIdentifier) {
getMetadataCache().invalidate(tableIdentifier);
- CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ Table table = getCatalog().loadTable(tableIdentifier);
+ if (!(table instanceof HasTableOperations)) {
+ CatalogHandlers.purgeTable(getCatalog(), tableIdentifier);
+ return;
+ }
+ // Iceberg drops the catalog entry first and then deletes the files,
giving up on the first
+ // file it cannot read (for example a missing manifest list). The request
then fails although
+ // the table is gone, and no request is left that could delete the files.
Read the metadata
+ // first, drop the entry, then delete every file that can still be reached.
+ TableOperations ops = ((HasTableOperations) table).operations();
+ TableMetadata metadata = ops.current();
+ FileIO io = ops.io();
+ if (!getCatalog().dropTable(tableIdentifier, false)) {
+ throw new NoSuchTableException("Table does not exist: %s",
tableIdentifier);
+ }
+ try {
+ CatalogUtil.dropTableData(io, metadata);
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Purge of {} could not read all table files ({}); deleting the files
that can be read",
+ tableIdentifier,
+ e.getMessage());
+ IcebergPurgeUtil.deleteReachableFiles(io, metadata);
+ }
}
/**
diff --git
a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergPurgeUtil.java
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergPurgeUtil.java
new file mode 100644
index 0000000000..1b30307376
--- /dev/null
+++
b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergPurgeUtil.java
@@ -0,0 +1,124 @@
+/*
+ * 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.gravitino.iceberg.common.ops;
+
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.ManifestContent;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestReader;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StatisticsFile;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Deletes a dropped table's files when some of them can no longer be read. */
+final class IcebergPurgeUtil {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergPurgeUtil.class);
+
+ private IcebergPurgeUtil() {}
+
+ /**
+ * Deletes every data, delete, manifest, manifest list, statistics and
metadata file reachable
+ * from the metadata, skipping the parts that cannot be read. A missing
manifest list only loses
+ * the files listed in it. Content files are preserved when {@code
gc.enabled} is false.
+ *
+ * @param io the table's FileIO
+ * @param metadata the table metadata read before the table was dropped
+ */
+ static void deleteReachableFiles(FileIO io, TableMetadata metadata) {
+ boolean gcEnabled =
+ PropertyUtil.propertyAsBoolean(
+ metadata.properties(), TableProperties.GC_ENABLED,
TableProperties.GC_ENABLED_DEFAULT);
+ Set<String> manifestLists = new HashSet<>();
+ Set<String> manifests = new HashSet<>();
+ Set<String> contentFiles = new HashSet<>();
+ for (Snapshot snapshot : metadata.snapshots()) {
+ if (snapshot.manifestListLocation() != null) {
+ manifestLists.add(snapshot.manifestListLocation());
+ }
+ try {
+ for (ManifestFile manifest : snapshot.allManifests(io)) {
+ if (manifests.add(manifest.path()) && gcEnabled) {
+ collectContentFiles(io, metadata, manifest, contentFiles);
+ }
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Cannot read the manifest list of snapshot {}; its files are not
deleted: {}",
+ snapshot.snapshotId(),
+ e.getMessage());
+ }
+ }
+ Set<String> metadataFiles = new HashSet<>();
+ metadata.previousFiles().forEach(entry -> metadataFiles.add(entry.file()));
+ if (metadata.metadataFileLocation() != null) {
+ metadataFiles.add(metadata.metadataFileLocation());
+ }
+ for (StatisticsFile stats : metadata.statisticsFiles()) {
+ metadataFiles.add(stats.path());
+ }
+ metadata.partitionStatisticsFiles().forEach(stats ->
metadataFiles.add(stats.path()));
+
+ deleteAll(io, contentFiles);
+ deleteAll(io, manifests);
+ deleteAll(io, manifestLists);
+ deleteAll(io, metadataFiles);
+ }
+
+ private static void collectContentFiles(
+ FileIO io, TableMetadata metadata, ManifestFile manifest, Set<String>
paths) {
+ try {
+ if (manifest.content() == ManifestContent.DATA) {
+ try (ManifestReader<DataFile> reader =
+ ManifestFiles.read(manifest, io, metadata.specsById())) {
+ reader.forEach(file -> paths.add(file.location()));
+ }
+ } else {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, io,
metadata.specsById())) {
+ reader.forEach(file -> paths.add(file.location()));
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Cannot read manifest {}; its files are not deleted: {}",
+ manifest.path(),
+ e.getMessage());
+ }
+ }
+
+ private static void deleteAll(FileIO io, Set<String> paths) {
+ for (String path : paths) {
+ try {
+ io.deleteFile(path);
+ } catch (RuntimeException e) {
+ LOG.warn("Cannot delete {}: {}", path, e.getMessage());
+ }
+ }
+ }
+}
diff --git
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
index c21e951fa3..e0dfe9dd98 100644
---
a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
+++
b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/ops/TestIcebergCatalogWrapper.java
@@ -21,9 +21,11 @@ package org.apache.gravitino.iceberg.common.ops;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.lang.reflect.Method;
+import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -37,9 +39,12 @@ import
org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
import org.apache.gravitino.iceberg.common.cache.TableMetadataCache;
import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil;
import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadataParser;
import org.apache.iceberg.catalog.Catalog;
@@ -59,6 +64,8 @@ import org.apache.iceberg.view.ViewOperations;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
public class TestIcebergCatalogWrapper {
@@ -306,6 +313,53 @@ public class TestIcebergCatalogWrapper {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testPurgeDeletesWhatCanBeReadWhenAManifestListIsMissing(
+ boolean gcEnabled, @TempDir Path warehouseDir) throws Exception {
+ try (IcebergCatalogWrapper wrapper =
+ new IcebergCatalogWrapper(new
IcebergConfig(sqliteConfig(warehouseDir)))) {
+ Namespace namespace = Namespace.of("db");
+ TableIdentifier ident = TableIdentifier.of(namespace, "purged");
+
wrapper.createNamespace(CreateNamespaceRequest.builder().withNamespace(namespace).build());
+ Table table =
+ wrapper
+ .getCatalog()
+ .createTable(
+ ident,
+ new Schema(Types.NestedField.required(1, "id",
Types.LongType.get())),
+ PartitionSpec.unpartitioned(),
+ ImmutableMap.of("gc.enabled", Boolean.toString(gcEnabled)));
+ Path first = Files.createFile(warehouseDir.resolve("first.parquet"));
+ Path second = Files.createFile(warehouseDir.resolve("second.parquet"));
+ for (Path data : new Path[] {first, second}) {
+ table
+ .newFastAppend()
+ .appendFile(
+ DataFiles.builder(table.spec())
+ .withPath(data.toUri().toString())
+ .withFileSizeInBytes(1)
+ .withRecordCount(1)
+ .build())
+ .commit();
+ }
+ table.refresh();
+ String metadataFile =
+ ((HasTableOperations)
table).operations().current().metadataFileLocation();
+ // The first snapshot's manifest list is gone.
+ Files.delete(
+
Paths.get(URI.create(table.snapshots().iterator().next().manifestListLocation())));
+
+ wrapper.purgeTable(ident);
+
+ Assertions.assertFalse(wrapper.getCatalog().tableExists(ident));
+ // Files reachable through the remaining manifest list, and the
metadata, are deleted.
+ Assertions.assertEquals(!gcEnabled, Files.exists(first));
+ Assertions.assertEquals(!gcEnabled, Files.exists(second));
+
Assertions.assertFalse(Files.exists(Paths.get(URI.create(metadataFile))));
+ }
+ }
+
private static TableMetadataCache
invokeGetMetadataCache(IcebergCatalogWrapper wrapper)
throws Exception {
Method method =
IcebergCatalogWrapper.class.getDeclaredMethod("getMetadataCache");
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index 64d5f090c0..c6d44c86d8 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -59,6 +59,7 @@ import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.exceptions.NoSuchPlanTaskException;
import org.apache.iceberg.exceptions.ServiceUnavailableException;
import org.apache.iceberg.io.CloseableIterable;
@@ -68,6 +69,7 @@ import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
@@ -139,9 +141,41 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return loadTableResponse;
}
+ /**
+ * Registers a table. An overwrite with a metadata file that belongs to a
different table (another
+ * table UUID) is refused before anything is written: Iceberg would replace
the table and then
+ * fail the request when it reads the replaced table back.
+ *
+ * @param namespace the namespace of the table
+ * @param request the register request
+ * @return the registered table
+ */
+ @Override
+ public LoadTableResponse registerTable(Namespace namespace,
RegisterTableRequest request) {
+ checkOverwriteKeepsTableUuid(namespace, request);
+ return super.registerTable(namespace, request);
+ }
+
+ /**
+ * Commits table updates, with {@link IcebergCommitValidation} checking the
updates against the
+ * metadata they are applied to.
+ *
+ * @param tableIdentifier the table to update
+ * @param updateTableRequest the updates and requirements
+ * @return the updated table
+ */
+ @Override
+ public LoadTableResponse updateTable(
+ TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
+ return super.updateTable(
+ tableIdentifier,
+ IcebergCommitValidation.withValidation(
+ updateTableRequest, () ->
getCatalog().loadTable(tableIdentifier).io()));
+ }
+
public LoadTableResponse registerTable(
Namespace namespace, RegisterTableRequest request, boolean
requestCredential) {
- LoadTableResponse loadTableResponse = super.registerTable(namespace,
request);
+ LoadTableResponse loadTableResponse = registerTable(namespace, request);
if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
// Vend WRITE credentials: the registering user becomes the table owner
// (IcebergNamespaceHookDispatcher.setTableOwner runs after this call
@@ -719,4 +753,28 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
properties.put(newProperty, deprecatedValue);
}
}
+
+ private void checkOverwriteKeepsTableUuid(Namespace namespace,
RegisterTableRequest request) {
+ if (request == null || !request.overwrite() || request.metadataLocation()
== null) {
+ return;
+ }
+ TableIdentifier identifier = TableIdentifier.of(namespace, request.name());
+ if (!tableExists(identifier)) {
+ return;
+ }
+ String currentUuid = super.loadTable(identifier).tableMetadata().uuid();
+ String newUuid;
+ try {
+ newUuid =
loadTableMetadataFromLocation(request.metadataLocation()).uuid();
+ } catch (RuntimeException e) {
+ // A metadata file that cannot be read is reported by the register
itself.
+ return;
+ }
+ if (currentUuid != null && newUuid != null &&
!currentUuid.equals(newUuid)) {
+ throw new BadRequestException(
+ "Cannot overwrite table %s with metadata of a different table:
current table UUID %s,"
+ + " metadata location %s has table UUID %s",
+ identifier, currentUuid, request.metadataLocation(), newUuid);
+ }
+ }
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
new file mode 100644
index 0000000000..d59271d6ab
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCommitValidation.java
@@ -0,0 +1,446 @@
+/*
+ * 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.gravitino.iceberg.service;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import org.apache.iceberg.DataOperations;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RetryableValidationException;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotRefType;
+import org.apache.iceberg.SortField;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Checks a table commit against the metadata it is applied to, before Iceberg
applies it.
+ *
+ * <p>Iceberg applies some metadata updates without checking them, and the
result is a table that
+ * clients can no longer write or load. The checks run as an extra {@link
UpdateRequirement}, so
+ * Iceberg evaluates them against the exact metadata the commit is built on,
again on every retry. A
+ * check that depends on the table's current refs therefore cannot be bypassed
by a concurrent
+ * commit.
+ */
+public final class IcebergCommitValidation implements UpdateRequirement {
+
+ private static final Set<String> SNAPSHOT_OPERATIONS =
+ ImmutableSet.of(
+ DataOperations.APPEND,
+ DataOperations.REPLACE,
+ DataOperations.OVERWRITE,
+ DataOperations.DELETE);
+
+ /** Transforms the table spec defines; anything else is an unknown
transform. */
+ private static final Pattern KNOWN_TRANSFORM =
+
Pattern.compile("identity|year|month|day|hour|void|bucket\\[\\d+]|truncate\\[\\d+]");
+
+ private static final int PARTITION_DATA_ID_START = 1000;
+
+ private final List<MetadataUpdate> updates;
+ private final Supplier<FileIO> fileIO;
+
+ private IcebergCommitValidation(List<MetadataUpdate> updates,
Supplier<FileIO> fileIO) {
+ this.updates = updates;
+ this.fileIO = fileIO;
+ }
+
+ /**
+ * Returns the request with the commit checks added to its requirements. The
request itself is not
+ * changed.
+ *
+ * @param request the table commit request
+ * @return a request with the same updates and the checks as an additional
requirement
+ */
+ @Nullable
+ public static UpdateTableRequest withValidation(@Nullable UpdateTableRequest
request) {
+ return withValidation(request, () -> null);
+ }
+
+ /**
+ * Returns the request with commit checks that use the table's FileIO to
read manifest lists.
+ *
+ * @param request the table commit request
+ * @param fileIO supplies the table's FileIO only when a check reads
manifest lists
+ * @return a request with validation, or the original request when it has no
updates
+ */
+ @Nullable
+ public static UpdateTableRequest withValidation(
+ @Nullable UpdateTableRequest request, Supplier<FileIO> fileIO) {
+ if (request == null || request.updates() == null ||
request.updates().isEmpty()) {
+ return request;
+ }
+ ImmutableList.Builder<UpdateRequirement> requirements =
ImmutableList.builder();
+ boolean create = false;
+ for (UpdateRequirement requirement : request.requirements()) {
+ if (requirement instanceof UpdateRequirement.AssertTableDoesNotExist) {
+ // CatalogHandlers accepts only AssertTableDoesNotExist requirements
for a staged create.
+ requirements.add(new CreateValidation(request.updates(), fileIO));
+ create = true;
+ } else {
+ requirements.add(requirement);
+ }
+ }
+ if (!create) {
+ requirements.add(new IcebergCommitValidation(request.updates(), fileIO));
+ }
+ return new UpdateTableRequest(requirements.build(), request.updates());
+ }
+
+ @Override
+ public void validate(@Nullable TableMetadata base) {
+ TableMetadata updated = validate(base, updates);
+ checkRemovedSpecsAreUnused(updated);
+ if (base != null) {
+ checkSchemas(base, updated);
+ }
+ checkPartitionSpecs(base, updated);
+ }
+
+ @VisibleForTesting
+ static TableMetadata validate(@Nullable TableMetadata base,
List<MetadataUpdate> updates) {
+ Set<Long> removedSnapshots = new HashSet<>();
+ Map<String, Long> remainingRefs = new HashMap<>();
+ if (base != null) {
+ base.refs().forEach((name, ref) -> remainingRefs.put(name,
ref.snapshotId()));
+ }
+ for (MetadataUpdate update : updates) {
+ if (update instanceof MetadataUpdate.RemoveSnapshots) {
+ removedSnapshots.addAll(((MetadataUpdate.RemoveSnapshots)
update).snapshotIds());
+ } else if (update instanceof MetadataUpdate.RemoveSnapshotRef) {
+ remainingRefs.remove(((MetadataUpdate.RemoveSnapshotRef)
update).name());
+ } else if (update instanceof MetadataUpdate.SetSnapshotRef) {
+ MetadataUpdate.SetSnapshotRef setRef = (MetadataUpdate.SetSnapshotRef)
update;
+ checkMainIsBranch(setRef);
+ remainingRefs.put(setRef.name(), setRef.snapshotId());
+ }
+ }
+
+ // Iceberg drops every ref that points at a removed snapshot. Removing a
snapshot that a branch
+ // or tag still references is only valid when the same commit also moves
or removes that ref.
+ for (Map.Entry<String, Long> ref : remainingRefs.entrySet()) {
+ if (removedSnapshots.contains(ref.getValue())) {
+ throw new CommitFailedException(
+ "Cannot remove snapshot %s: it is referenced by ref %s",
ref.getValue(), ref.getKey());
+ }
+ }
+
+ // Apply only in memory. Iceberg assigns schema ids while applying
AddSchema, so checking the
+ // resulting metadata avoids accepting an arbitrary id just because the
request adds a schema.
+ TableMetadata.Builder builder;
+ if (base == null) {
+ builder =
+ updates.stream()
+ .filter(MetadataUpdate.UpgradeFormatVersion.class::isInstance)
+ .map(MetadataUpdate.UpgradeFormatVersion.class::cast)
+ .map(MetadataUpdate.UpgradeFormatVersion::formatVersion)
+ .findFirst()
+ .map(TableMetadata::buildFromEmpty)
+ .orElseGet(TableMetadata::buildFromEmpty);
+ } else {
+ builder = TableMetadata.buildFrom(base);
+ }
+ try {
+ updates.forEach(update -> update.applyTo(builder));
+ } catch (RetryableValidationException e) {
+ // Preserve CatalogHandlers' conflict response for stale sequence
numbers and row ids.
+ throw new CommitFailedException(e, "Validation failed, please retry:
%s", e.getMessage());
+ }
+ TableMetadata updated = builder.build();
+ for (MetadataUpdate update : updates) {
+ if (update instanceof MetadataUpdate.AddSnapshot) {
+ checkSnapshot(updated, ((MetadataUpdate.AddSnapshot)
update).snapshot());
+ }
+ }
+ return updated;
+ }
+
+ /**
+ * A partition spec can only be removed when no manifest of any snapshot was
written with it:
+ * readers look up every manifest's spec in the table metadata.
+ */
+ private void checkRemovedSpecsAreUnused(TableMetadata base) {
+ Set<Integer> removed = new HashSet<>();
+ for (MetadataUpdate update : updates) {
+ if (update instanceof MetadataUpdate.RemovePartitionSpecs) {
+ removed.addAll(((MetadataUpdate.RemovePartitionSpecs)
update).specIds());
+ }
+ }
+ if (removed.isEmpty() || base.snapshots().isEmpty()) {
+ return;
+ }
+ FileIO io = fileIO.get();
+ if (io == null) {
+ throw new ValidationException(
+ "Cannot remove partition specs %s: their use by existing manifests
cannot be checked",
+ removed);
+ }
+ for (Snapshot snapshot : base.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(io)) {
+ if (removed.contains(manifest.partitionSpecId())) {
+ throw new ValidationException(
+ "Cannot remove partition spec %s: snapshot %s has manifests
written with it",
+ manifest.partitionSpecId(), snapshot.snapshotId());
+ }
+ }
+ }
+ }
+
+ /**
+ * An added schema must keep the data already written readable: a field id
that exists in the
+ * current schema keeps its kind and may only change type by an allowed
promotion, a field id not
+ * in the current schema must be new (above last-column-id) or identical to
the field that had the
+ * id before, and a new required field needs a default. A replace (which
removes main) starts from
+ * no data, so only the id rule applies to it.
+ */
+ private void checkSchemas(TableMetadata base, TableMetadata updated) {
+ boolean replacesTable =
+ updates.stream()
+ .anyMatch(
+ u ->
+ u instanceof MetadataUpdate.RemoveSnapshotRef
+ && SnapshotRef.MAIN_BRANCH.equals(
+ ((MetadataUpdate.RemoveSnapshotRef) u).name()))
+ && (updated.currentSnapshot() == null
+ || base.snapshot(updated.currentSnapshot().snapshotId()) ==
null);
+ boolean hasData = !base.snapshots().isEmpty();
+ Map<Integer, Types.NestedField> current =
TypeUtil.indexById(base.schema().asStruct());
+ for (MetadataUpdate update : updates) {
+ if (!(update instanceof MetadataUpdate.AddSchema)) {
+ continue;
+ }
+ Schema schema = ((MetadataUpdate.AddSchema) update).schema();
+ for (Types.NestedField field :
TypeUtil.indexById(schema.asStruct()).values()) {
+ Types.NestedField existing = current.get(field.fieldId());
+ if (existing == null) {
+ checkNewField(base, field, hasData && !replacesTable);
+ } else if (hasData && !replacesTable) {
+ checkTypeChange(existing, field);
+ }
+ }
+ }
+ }
+
+ private static void checkNewField(
+ TableMetadata base, Types.NestedField field, boolean checkRequired) {
+ if (field.fieldId() <= base.lastColumnId()) {
+ Types.NestedField previous = previousField(base, field.fieldId());
+ boolean sameField =
+ previous != null
+ && previous.name().equals(field.name())
+ && previous.type().equals(field.type());
+ if (!sameField) {
+ throw new ValidationException(
+ "Invalid schema: field %s reuses id %s, which the table assigned
before (last column"
+ + " id %s); new fields need new ids",
+ field.name(), field.fieldId(), base.lastColumnId());
+ }
+ }
+ if (checkRequired && field.isRequired() && field.initialDefault() == null)
{
+ throw new ValidationException(
+ "Invalid schema: required field %s has no initial default, so
existing rows cannot be"
+ + " read",
+ field.name());
+ }
+ }
+
+ private static Types.NestedField previousField(TableMetadata base, int
fieldId) {
+ for (Schema schema : base.schemas()) {
+ Types.NestedField field = schema.findField(fieldId);
+ if (field != null) {
+ return field;
+ }
+ }
+ return null;
+ }
+
+ private static void checkTypeChange(Types.NestedField existing,
Types.NestedField field) {
+ Type from = existing.type();
+ Type to = field.type();
+ if (from.equals(to)) {
+ return;
+ }
+ if (from.isPrimitiveType() && to.isPrimitiveType()) {
+ if (!TypeUtil.isPromotionAllowed(from, to.asPrimitiveType())) {
+ throw new ValidationException(
+ "Invalid schema: field %s cannot change type from %s to %s",
field.name(), from, to);
+ }
+ } else if (from.typeId() != to.typeId()) {
+ throw new ValidationException(
+ "Invalid schema: field %s cannot change from %s to %s",
field.name(), from, to);
+ }
+ }
+
+ /**
+ * Partition field ids identify partition data across specs (v2+): a field
id another spec uses
+ * for a different field, or a new field id at or below last-partition-id,
would mix up partition
+ * values. A spec or sort order with a transform the table spec does not
define cannot be the
+ * default: writers are not allowed to commit data with it.
+ */
+ private void checkPartitionSpecs(@Nullable TableMetadata base, TableMetadata
updated) {
+ boolean setsDefaultSpec = false;
+ boolean setsDefaultOrder = false;
+ for (MetadataUpdate update : updates) {
+ if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) {
+ setsDefaultSpec = true;
+ } else if (update instanceof MetadataUpdate.SetDefaultSortOrder) {
+ setsDefaultOrder = true;
+ }
+ }
+
+ // Check the actual defaults, not every spec/order added by the request.
An unknown transform
+ // is allowed in a retained, non-default spec/order.
+ if (setsDefaultSpec || base == null) {
+ checkKnownTransforms(updated.spec().fields(), "partition spec " +
updated.defaultSpecId());
+ }
+ if (setsDefaultOrder || base == null) {
+ checkKnownSortTransforms(updated.sortOrder(), "sort order " +
updated.defaultSortOrderId());
+ }
+
+ if (updated.formatVersion() < 2) {
+ return;
+ }
+ Map<Integer, PartitionField> existingFields = new HashMap<>();
+ if (base != null) {
+ for (PartitionSpec existing : base.specs()) {
+ for (PartitionField field : existing.fields()) {
+ existingFields.putIfAbsent(field.fieldId(), field);
+ }
+ }
+ }
+ int lastAssigned =
+ base == null
+ ? PARTITION_DATA_ID_START - 1
+ : Math.max(base.lastAssignedPartitionId(), PARTITION_DATA_ID_START
- 1);
+ for (PartitionSpec spec : updated.specs()) {
+ if (base == null || !base.specsById().containsKey(spec.specId())) {
+ lastAssigned = checkPartitionFieldIds(existingFields, lastAssigned,
spec);
+ }
+ }
+ }
+
+ private static int checkPartitionFieldIds(
+ Map<Integer, PartitionField> existingFields, int lastAssigned,
PartitionSpec spec) {
+ int highestId = lastAssigned;
+ for (PartitionField field : spec.fields()) {
+ PartitionField existing = existingFields.get(field.fieldId());
+ if (existing != null) {
+ if (existing.sourceId() != field.sourceId()
+ ||
!existing.transform().toString().equals(field.transform().toString())) {
+ throw new ValidationException(
+ "Invalid partition spec: field id %s is already used for %s, not
%s",
+ field.fieldId(), existing, field);
+ }
+ } else if (field.fieldId() <= lastAssigned) {
+ throw new ValidationException(
+ "Invalid partition spec: new field %s needs an id above
last-partition-id %s",
+ field, lastAssigned);
+ }
+ existingFields.putIfAbsent(field.fieldId(), field);
+ highestId = Math.max(highestId, field.fieldId());
+ }
+ return highestId;
+ }
+
+ private static void checkKnownTransforms(List<PartitionField> fields, String
what) {
+ for (PartitionField field : fields) {
+ checkKnownTransform(field.transform().toString(), field.name(), what);
+ }
+ }
+
+ private static void checkKnownSortTransforms(SortOrder order, String what) {
+ for (SortField field : order.fields()) {
+ checkKnownTransform(field.transform().toString(), "source " +
field.sourceId(), what);
+ }
+ }
+
+ private static void checkKnownTransform(String transform, String field,
String what) {
+ if (!KNOWN_TRANSFORM.matcher(transform).matches()) {
+ throw new ValidationException(
+ "Cannot make %s the default: field %s uses transform %s, which the
table spec does not"
+ + " define, and writers cannot commit data with it",
+ what, field, transform);
+ }
+ }
+
+ private static void checkMainIsBranch(MetadataUpdate.SetSnapshotRef setRef) {
+ if (SnapshotRef.MAIN_BRANCH.equals(setRef.name())
+ && SnapshotRefType.TAG.name().equalsIgnoreCase(setRef.type())) {
+ throw new ValidationException(
+ "Cannot set %s to a tag: it must be a branch",
SnapshotRef.MAIN_BRANCH);
+ }
+ }
+
+ private static void checkSnapshot(TableMetadata base, Snapshot snapshot) {
+ // The summary and its operation are required from format version 2 on.
+ String operation = snapshot.operation();
+ if (operation == null) {
+ if (base.formatVersion() >= 2) {
+ throw new ValidationException(
+ "Invalid snapshot %s: a summary with an operation is required",
snapshot.snapshotId());
+ }
+ } else if (!SNAPSHOT_OPERATIONS.contains(operation)) {
+ throw new ValidationException(
+ "Invalid snapshot %s: unknown operation '%s', must be one of %s",
+ snapshot.snapshotId(), operation, SNAPSHOT_OPERATIONS);
+ }
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId != null && !base.schemasById().containsKey(schemaId)) {
+ throw new ValidationException(
+ "Invalid snapshot %s: the table has no schema with id %s",
+ snapshot.snapshotId(), schemaId);
+ }
+ }
+
+ private static final class CreateValidation extends
UpdateRequirement.AssertTableDoesNotExist {
+ private final IcebergCommitValidation validation;
+
+ private CreateValidation(List<MetadataUpdate> updates, Supplier<FileIO>
fileIO) {
+ this.validation = new IcebergCommitValidation(updates, fileIO);
+ }
+
+ @Override
+ public void validate(@Nullable TableMetadata base) {
+ super.validate(base);
+ validation.validate(base);
+ }
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergObjectMapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergObjectMapper.java
index 76fb9a6610..1fe00b2eff 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergObjectMapper.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergObjectMapper.java
@@ -20,12 +20,25 @@ package org.apache.gravitino.iceberg.service;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.IOException;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.UnboundPartitionSpec;
import org.apache.iceberg.rest.RESTSerializers;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequestParser;
public class IcebergObjectMapper {
+ private static final String SPEC_ID = "spec-id";
+
private static final ObjectMapper objectMapper = createIcebergObjectMapper();
private IcebergObjectMapper() {}
@@ -36,9 +49,52 @@ public class IcebergObjectMapper {
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
mapper.setPropertyNamingStrategy(new
PropertyNamingStrategies.KebabCaseStrategy());
RESTSerializers.registerAll(mapper);
+ // Registered after Iceberg's serializers, so these take precedence for
the same types.
+ SimpleModule module = new SimpleModule("gravitino-iceberg-rest");
+ module.addDeserializer(UnboundPartitionSpec.class, new
PartitionSpecDeserializer());
+ module.addDeserializer(UpdateTableRequest.class, new
UpdateTableRequestDeserializer());
+ mapper.registerModule(module);
return mapper;
}
+ /**
+ * The REST spec makes a partition spec's {@code spec-id} read-only, so
clients may leave it out,
+ * but Iceberg's parser requires it. The server assigns the id of a new spec
itself, so a missing
+ * id is read as 0.
+ */
+ private static void defaultSpecId(JsonNode spec) {
+ if (spec instanceof ObjectNode && !spec.has(SPEC_ID)) {
+ ((ObjectNode) spec).put(SPEC_ID, 0);
+ }
+ }
+
+ private static class PartitionSpecDeserializer extends
JsonDeserializer<UnboundPartitionSpec> {
+ @Override
+ public UnboundPartitionSpec deserialize(JsonParser parser,
DeserializationContext context)
+ throws IOException {
+ JsonNode node = parser.getCodec().readTree(parser);
+ defaultSpecId(node);
+ return PartitionSpecParser.fromJson(node);
+ }
+ }
+
+ private static class UpdateTableRequestDeserializer extends
JsonDeserializer<UpdateTableRequest> {
+ @Override
+ public UpdateTableRequest deserialize(JsonParser parser,
DeserializationContext context)
+ throws IOException {
+ JsonNode node = parser.getCodec().readTree(parser);
+ JsonNode updates = node == null ? null : node.get("updates");
+ if (updates != null && updates.isArray()) {
+ for (JsonNode update : updates) {
+ if ("add-spec".equals(update.path("action").asText())) {
+ defaultSpecId(update.get("spec"));
+ }
+ }
+ }
+ return UpdateTableRequestParser.fromJson(node);
+ }
+ }
+
public static ObjectMapper getInstance() {
return objectMapper;
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
index 1d9e85bb97..d38cea75bf 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
@@ -552,12 +552,33 @@ public class IcebergRESTUtils {
return catalogName;
}
+ /**
+ * Copies an Iceberg REST request or response through its JSON form, so
event listeners get an
+ * object detached from the one the server works on.
+ *
+ * <p>Events are built after the operation has been applied, so a copy that
fails must not turn a
+ * completed operation into an error. Some valid requests cannot be
serialized again by Iceberg,
+ * for example a partition field sent without its optional {@code field-id};
for those the
+ * original object, which Iceberg does not modify, is returned instead.
+ *
+ * @param message the object to copy
+ * @param className the class of the copy
+ * @param <T> the type of the copy
+ * @return the copy, or the original object if it cannot be copied
+ */
public static <T> T cloneIcebergRESTObject(Object message, Class<T>
className) {
ObjectMapper icebergObjectMapper = IcebergObjectMapper.getInstance();
try {
byte[] values = icebergObjectMapper.writeValueAsBytes(message);
return icebergObjectMapper.readValue(values, className);
- } catch (IOException e) {
+ } catch (IOException | RuntimeException e) {
+ if (className.isInstance(message)) {
+ LOG.warn(
+ "Cannot copy {} for an event, the event carries the original
object: {}",
+ className.getSimpleName(),
+ e.getMessage());
+ return className.cast(message);
+ }
throw new RuntimeException(e);
}
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceHookDispatcher.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceHookDispatcher.java
index 16481a3dd5..1759c2e1fb 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceHookDispatcher.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceHookDispatcher.java
@@ -18,11 +18,13 @@
*/
package org.apache.gravitino.iceberg.service.dispatcher;
+import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityFieldLimits;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
@@ -37,6 +39,7 @@ import org.apache.gravitino.lock.TreeLockUtils;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.iceberg.rest.requests.RegisterViewRequest;
@@ -69,6 +72,7 @@ public class IcebergNamespaceHookDispatcher implements
IcebergNamespaceOperation
IcebergRequestContext context, CreateNamespaceRequest createRequest) {
String catalogName = context.catalogName();
Namespace leaf = createRequest.namespace();
+ checkGravitinoSchemaName(leaf);
List<Namespace> newlyOwned = new ArrayList<>();
// Lock the top-level branch root rather than the whole catalog: any race
on shared ancestor
// ownership has to share leaf.level(0), so serializing per top-level
branch is sufficient
@@ -86,7 +90,8 @@ public class IcebergNamespaceHookDispatcher implements
IcebergNamespaceOperation
// SchemaMetaService.insertSchema splits the leaf's logical name and
auto-creates a
// Gravitino entity row for each ancestor, so a single leaf import covers
the branch.
// Failures propagate intentionally: swallowing would leave a namespace in
Iceberg
- // that Gravitino doesn't know about.
+ // that Gravitino doesn't know about. Do not drop the namespace on import
failure: another
+ // client may already have created objects in it.
importSchema(catalogName, leaf);
// getMissingAncestors() only returns the not-yet-existing ancestors;
append the leaf so that
@@ -101,6 +106,33 @@ public class IcebergNamespaceHookDispatcher implements
IcebergNamespaceOperation
return createNamespaceResponse;
}
+ /**
+ * Rejects a namespace Gravitino cannot hold as a schema before anything is
created: a level
+ * containing '.', which is reserved as the qualified-name separator (and
which the JDBC backend
+ * would store as a different, nested namespace), or a full name longer than
the schema name
+ * column.
+ */
+ @VisibleForTesting
+ static void checkGravitinoSchemaName(Namespace namespace) {
+ if (GravitinoEnv.getInstance().internalSchemaDispatcher() == null) {
+ return;
+ }
+ for (String level : namespace.levels()) {
+ if (level.contains(".")) {
+ throw new BadRequestException(
+ "Invalid namespace %s: the level '%s' contains '.', which is
reserved as the"
+ + " qualified-name separator",
+ namespace, level);
+ }
+ }
+ String name = String.join(HierarchicalSchemaUtil.schemaSeparator(),
namespace.levels());
+ if (name.codePointCount(0, name.length()) >
EntityFieldLimits.MAX_NAME_LENGTH) {
+ throw new BadRequestException(
+ "Invalid namespace %s: the full name must not exceed %s characters",
+ namespace, EntityFieldLimits.MAX_NAME_LENGTH);
+ }
+ }
+
/**
* Returns all ancestor namespaces of {@code namespace} that do not
currently exist in the
* catalog. Uses {@link HierarchicalSchemaUtil#getAncestorNames} (the same
utility as the
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java
index d804c14c14..c6578af0c4 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableHookDispatcher.java
@@ -30,6 +30,8 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.lock.LockType;
+import org.apache.gravitino.lock.TreeLockUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.TableEntity;
import org.apache.gravitino.utils.HierarchicalSchemaUtil;
@@ -69,7 +71,13 @@ public class IcebergTableHookDispatcher implements
IcebergTableOperationDispatch
@Override
public LoadTableResponse createTable(
IcebergRequestContext context, Namespace namespace, CreateTableRequest
createTableRequest) {
- LoadTableResponse response = dispatcher.createTable(context, namespace,
createTableRequest);
+ // Hold the namespace's tree lock in read mode, as dropNamespace holds it
in write mode, so a
+ // create and a drop of the same namespace on this server do not both
succeed.
+ LoadTableResponse response =
+ TreeLockUtils.doWithTreeLock(
+ NameIdentifier.of(metalake, context.catalogName(),
namespace.level(0)),
+ LockType.READ,
+ () -> dispatcher.createTable(context, namespace,
createTableRequest));
// Skip import and ownership for staged creates (credential vending).
// The table will be imported when the staged table is committed via
updateTable.
if (!createTableRequest.stageCreate()) {
@@ -84,14 +92,20 @@ public class IcebergTableHookDispatcher implements
IcebergTableOperationDispatch
IcebergRequestContext context,
TableIdentifier tableIdentifier,
UpdateTableRequest updateTableRequest) {
- LoadTableResponse response =
- dispatcher.updateTable(context, tableIdentifier, updateTableRequest);
// Import the table and set ownership only when committing a staged table
create.
// A staged create commit is identified by the AssertTableDoesNotExist
requirement,
// which is set by UpdateRequirements.forCreateTable().
boolean isStagedCreateCommit =
updateTableRequest.requirements().stream()
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance);
+ LoadTableResponse response =
+ isStagedCreateCommit
+ ? TreeLockUtils.doWithTreeLock(
+ NameIdentifier.of(
+ metalake, context.catalogName(),
tableIdentifier.namespace().level(0)),
+ LockType.READ,
+ () -> dispatcher.updateTable(context, tableIdentifier,
updateTableRequest))
+ : dispatcher.updateTable(context, tableIdentifier,
updateTableRequest);
if (isStagedCreateCommit) {
importTableAndSetOwner(context, tableIdentifier.namespace(),
tableIdentifier.name());
}
@@ -224,9 +238,8 @@ public class IcebergTableHookDispatcher implements
IcebergTableOperationDispatch
private void importTableAndSetOwner(
IcebergRequestContext context, Namespace namespace, String tableName) {
- // Import is intentionally NOT wrapped in try-catch: if it fails the table
exists in Iceberg
- // but not in Gravitino, and silently swallowing that would mislead
callers into thinking the
- // entity is registered. Surface the failure so the caller can react.
+ // Validate request metadata before the backend operation. An unexpected
import failure must
+ // propagate without deleting a table that another client may already have
updated or recreated.
importTableEntity(context.catalogName(), namespace, tableName);
IcebergOwnershipUtils.setTableOwner(
metalake,
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergPaginationHelper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergPaginationHelper.java
index 625a342562..69a4c5c49c 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergPaginationHelper.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergPaginationHelper.java
@@ -49,6 +49,8 @@ import org.apache.iceberg.rest.responses.ListTablesResponse;
* implementations and is generally acceptable.
* <li>When {@code pageToken} is provided without {@code pageSize}, the
response returns all items
* after the cursor with no {@code nextPageToken}.
+ * <li>Without a {@code pageToken} parameter all items are returned, even
when {@code pageSize} is
+ * set, as the REST spec requires; an empty {@code pageToken} requests
the first page.
* </ul>
*/
class IcebergPaginationHelper {
@@ -59,8 +61,8 @@ class IcebergPaginationHelper {
* Paginate a {@link ListNamespacesResponse}.
*
* @param response the full (unpaginated) response
- * @param pageToken cursor-based page token (name of last item on previous
page), or empty for the
- * first page
+ * @param pageToken cursor-based page token (name of last item on previous
page), an empty string
+ * for the first page, or absent to return all results
* @param pageSize maximum items per page, or empty for no limit
* @return a new response containing only the requested page
*/
@@ -77,8 +79,8 @@ class IcebergPaginationHelper {
* Paginate a {@link ListTablesResponse}. Works for both table and view list
responses.
*
* @param response the full (unpaginated) response
- * @param pageToken cursor-based page token (name of last item on previous
page), or empty for the
- * first page
+ * @param pageToken cursor-based page token (name of last item on previous
page), an empty string
+ * for the first page, or absent to return all results
* @param pageSize maximum items per page, or empty for no limit
* @return a new response containing only the requested page
*/
@@ -98,7 +100,8 @@ class IcebergPaginationHelper {
* cursor token. The next-page token is the key of the last item on the
returned page.
*
* @param items the complete list of items to paginate
- * @param pageToken cursor string (key of last item on previous page), or
empty for the first page
+ * @param pageToken cursor string (key of last item on previous page), an
empty string for the
+ * first page, or absent when the request has no pageToken parameter
* @param pageSize maximum items per page, or empty for no limit
* @param keyExtractor function to extract a comparable cursor key from each
item
* @return a {@link PaginatedPage} containing the requested page of items
and optional next token
@@ -108,13 +111,22 @@ class IcebergPaginationHelper {
Optional<String> pageToken,
Optional<Integer> pageSize,
Function<T, String> keyExtractor) {
- String token = pageToken.orElse("");
- if (!pageSize.isPresent() && token.isEmpty()) {
+ pageSize.ifPresent(
+ size -> Preconditions.checkArgument(size > 0, "pageSize must be
positive, got: %s", size));
+
+ // The REST spec: a server that supports pagination returns all results,
with no next-page
+ // token, when the request has no pageToken parameter. A client starts
paging by sending an
+ // empty pageToken, so pageSize alone does not start paging.
+ if (!pageToken.isPresent()) {
return new PaginatedPage<>(items, Optional.empty());
}
+ String token = pageToken.get();
- pageSize.ifPresent(
- size -> Preconditions.checkArgument(size > 0, "pageSize must be
positive, got: %s", size));
+ // An empty pageToken with no pageSize asks for every item from the start,
which is the
+ // unsorted list as given; skip the sort.
+ if (token.isEmpty() && !pageSize.isPresent()) {
+ return new PaginatedPage<>(items, Optional.empty());
+ }
List<T> sorted =
items.stream().sorted(Comparator.comparing(keyExtractor)).collect(Collectors.toList());
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
index 797ac23eb1..632bea9983 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
@@ -32,6 +32,7 @@ import com.sun.net.httpserver.HttpServer;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -47,6 +48,7 @@ import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.service.cache.LocalScanPlanCache;
import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.MetadataUpdate;
@@ -62,6 +64,7 @@ import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.exceptions.ForbiddenException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.NotAuthorizedException;
@@ -87,6 +90,7 @@ import
org.apache.iceberg.rest.responses.PlanTableScanResponseParser;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
public class TestCatalogWrapperForREST {
@@ -389,6 +393,63 @@ public class TestCatalogWrapperForREST {
}
}
+ @Test
+ void testRegisterOverwriteWithAnotherTablesMetadataIsRefused(@TempDir Path
warehouse) {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.<String, String>builder()
+ .put(IcebergConstants.CATALOG_BACKEND, "jdbc")
+ .put(IcebergConstants.URI, "jdbc:sqlite:" +
warehouse.resolve("catalog.db"))
+ .put(IcebergConstants.WAREHOUSE, warehouse.toUri().toString())
+ .put(IcebergConstants.GRAVITINO_JDBC_DRIVER, "org.sqlite.JDBC")
+ .put(IcebergConstants.ICEBERG_JDBC_USER, "test")
+ .put(IcebergConstants.ICEBERG_JDBC_PASSWORD, "test")
+ .put(IcebergConstants.ICEBERG_JDBC_INITIALIZE, "true")
+ .build());
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("register-test",
config);
+ Namespace namespace = Namespace.of("db");
+ Catalog catalog = wrapper.getCatalog();
+ ((SupportsNamespaces) catalog).createNamespace(namespace);
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ TableIdentifier target = TableIdentifier.of(namespace, "target");
+ TableIdentifier other = TableIdentifier.of(namespace, "other");
+ catalog.createTable(target, schema);
+ catalog.createTable(other, schema);
+ String targetMetadata =
+ ((BaseTable)
catalog.loadTable(target)).operations().current().metadataFileLocation();
+ String otherMetadata =
+ ((BaseTable)
catalog.loadTable(other)).operations().current().metadataFileLocation();
+
+ RegisterTableRequest otherTable =
+ ImmutableRegisterTableRequest.builder()
+ .name("target")
+ .metadataLocation(otherMetadata)
+ .overwrite(true)
+ .build();
+ BadRequestException e =
+ Assertions.assertThrows(
+ BadRequestException.class, () -> wrapper.registerTable(namespace,
otherTable));
+ Assertions.assertTrue(e.getMessage().contains("different table"),
e.getMessage());
+ // The REST executor uses the overload carrying the credential-vending
flag.
+ for (boolean requestCredential : new boolean[] {false, true}) {
+ Assertions.assertThrows(
+ BadRequestException.class,
+ () -> wrapper.registerTable(namespace, otherTable,
requestCredential));
+ }
+ Assertions.assertEquals(
+ targetMetadata,
+ ((BaseTable)
catalog.loadTable(target)).operations().current().metadataFileLocation());
+
+ // An overwrite with metadata of the same table is still accepted.
+ RegisterTableRequest sameTable =
+ ImmutableRegisterTableRequest.builder()
+ .name("target")
+ .metadataLocation(targetMetadata)
+ .overwrite(true)
+ .build();
+ Assertions.assertDoesNotThrow(() -> wrapper.registerTable(namespace,
sameTable));
+ }
+
@Test
void testLocalTableCredentials() {
IcebergConfig config =
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
new file mode 100644
index 0000000000..238941d98f
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCommitValidation.java
@@ -0,0 +1,613 @@
+/*
+ * 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.gravitino.iceberg.service;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.util.List;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.MetadataUpdateParser;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SnapshotParser;
+import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.SnapshotRefType;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.rest.CatalogHandlers;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergCommitValidation {
+
+ private static final Schema SCHEMA =
+ new Schema(Types.NestedField.required(1, "id", Types.LongType.get()));
+ private static final TableIdentifier IDENT =
TableIdentifier.of(Namespace.of("db"), "t");
+
+ private InMemoryCatalog catalog;
+ private long firstSnapshot;
+ private long secondSnapshot;
+
+ @BeforeEach
+ public void setUp() {
+ catalog = new InMemoryCatalog();
+ catalog.initialize("memory", ImmutableMap.of());
+ catalog.createNamespace(Namespace.of("db"));
+ Table table =
+ catalog.createTable(
+ IDENT, SCHEMA, PartitionSpec.unpartitioned(),
ImmutableMap.of("format-version", "2"));
+ append(table, "/data/f1.parquet");
+ firstSnapshot = table.currentSnapshot().snapshotId();
+ append(table, "/data/f2.parquet");
+ secondSnapshot = table.currentSnapshot().snapshotId();
+ }
+
+ @Test
+ public void testMainCannotBecomeATag() {
+ commitFails(
+ ValidationException.class,
+ new MetadataUpdate.SetSnapshotRef(
+ SnapshotRef.MAIN_BRANCH, secondSnapshot, SnapshotRefType.TAG,
null, null, null));
+
Assertions.assertTrue(catalog.loadTable(IDENT).refs().get(SnapshotRef.MAIN_BRANCH).isBranch());
+
+ // Other tags, and main as a branch, are unaffected.
+ commit(
+ new MetadataUpdate.SetSnapshotRef(
+ "v1", firstSnapshot, SnapshotRefType.TAG, null, null, null));
+ Assertions.assertTrue(catalog.loadTable(IDENT).refs().get("v1").isTag());
+ }
+
+ @Test
+ public void testAddSnapshotIsChecked() {
+ // An operation outside the spec's values.
+ commitFails(
+ ValidationException.class, addSnapshot(", \"summary\": {\"operation\":
\"bogus\"}"));
+ // No summary on a v2 table.
+ commitFails(ValidationException.class, addSnapshot(""));
+ // A schema id the table does not have.
+ commitFails(
+ ValidationException.class,
+ addSnapshot(", \"summary\": {\"operation\": \"append\"},
\"schema-id\": 99"));
+
+ Assertions.assertEquals(
+ secondSnapshot,
catalog.loadTable(IDENT).currentSnapshot().snapshotId());
+ Assertions.assertNull(catalog.loadTable(IDENT).snapshot(999L));
+
+ // A valid snapshot is still accepted.
+ commit(addSnapshot(", \"summary\": {\"operation\": \"append\"},
\"schema-id\": 0"));
+ Assertions.assertNotNull(catalog.loadTable(IDENT).snapshot(999L));
+ }
+
+ @Test
+ public void testRemovingReferencedSnapshotIsRefused() {
+ // The current snapshot of main.
+ commitFails(CommitFailedException.class, new
MetadataUpdate.RemoveSnapshots(secondSnapshot));
+ // A snapshot a tag references.
+ commit(
+ new MetadataUpdate.SetSnapshotRef(
+ "keep", firstSnapshot, SnapshotRefType.TAG, null, null, null));
+ commitFails(CommitFailedException.class, new
MetadataUpdate.RemoveSnapshots(firstSnapshot));
+
+ Table table = catalog.loadTable(IDENT);
+ Assertions.assertEquals(secondSnapshot,
table.currentSnapshot().snapshotId());
+ Assertions.assertTrue(table.refs().containsKey("keep"));
+ Assertions.assertNotNull(table.snapshot(firstSnapshot));
+
+ // Removing the tag in the same commit makes the removal valid.
+ commit(
+ new MetadataUpdate.RemoveSnapshotRef("keep"),
+ new MetadataUpdate.RemoveSnapshots(ImmutableSet.of(firstSnapshot)));
+ Assertions.assertNull(catalog.loadTable(IDENT).snapshot(firstSnapshot));
+ }
+
+ @Test
+ public void testAddingSchemaDoesNotBypassSnapshotSchemaValidation() {
+ Schema schema =
+ new Schema(
+ 99,
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()));
+ commitFails(
+ ValidationException.class,
+ new MetadataUpdate.AddSchema(schema),
+ addSnapshot(", \"summary\": {\"operation\": \"append\"},
\"schema-id\": 99"));
+ Assertions.assertEquals(1, catalog.loadTable(IDENT).schemas().size());
+ Assertions.assertNull(catalog.loadTable(IDENT).snapshot(999L));
+
+ // Iceberg assigns id 1 to the new schema, regardless of the id supplied
in add-schema.
+ commit(
+ new MetadataUpdate.AddSchema(schema),
+ addSnapshot(", \"summary\": {\"operation\": \"append\"},
\"schema-id\": 1"));
+ Assertions.assertNotNull(catalog.loadTable(IDENT).schemas().get(1));
+ Assertions.assertEquals(1,
catalog.loadTable(IDENT).snapshot(999L).schemaId());
+ }
+
+ @Test
+ public void testStaleSnapshotRemainsACommitConflict() {
+ MetadataUpdate snapshot =
+ new MetadataUpdate.AddSnapshot(
+ SnapshotParser.fromJson(
+ "{\"snapshot-id\": 999, \"parent-snapshot-id\": "
+ + secondSnapshot
+ + ", \"sequence-number\": 1, \"timestamp-ms\": "
+ + System.currentTimeMillis()
+ + ", \"manifest-list\": \"/data/snap-999.avro\","
+ + " \"summary\": {\"operation\": \"append\"},
\"schema-id\": 0}"));
+ commitFails(CommitFailedException.class, snapshot);
+ Assertions.assertEquals(
+ secondSnapshot,
catalog.loadTable(IDENT).currentSnapshot().snapshotId());
+ Assertions.assertNull(catalog.loadTable(IDENT).snapshot(999L));
+ }
+
+ @Test
+ public void testChangingRefMustActuallyMoveItOffRemovedSnapshot() {
+ commit(
+ new MetadataUpdate.SetSnapshotRef(
+ "keep", firstSnapshot, SnapshotRefType.TAG, null, null, null));
+ commitFails(
+ CommitFailedException.class,
+ new MetadataUpdate.SetSnapshotRef(
+ "keep", firstSnapshot, SnapshotRefType.TAG, null, null, null),
+ new MetadataUpdate.RemoveSnapshots(firstSnapshot));
+ Assertions.assertNotNull(catalog.loadTable(IDENT).snapshot(firstSnapshot));
+ Assertions.assertEquals(
+ firstSnapshot,
catalog.loadTable(IDENT).refs().get("keep").snapshotId());
+
+ commit(
+ new MetadataUpdate.SetSnapshotRef(
+ "keep", secondSnapshot, SnapshotRefType.TAG, null, null, null),
+ new MetadataUpdate.RemoveSnapshots(firstSnapshot));
+ Assertions.assertNull(catalog.loadTable(IDENT).snapshot(firstSnapshot));
+ Assertions.assertEquals(
+ secondSnapshot,
catalog.loadTable(IDENT).refs().get("keep").snapshotId());
+ }
+
+ @Test
+ public void testStagedCreateCommitIsAccepted() {
+ TableIdentifier staged = TableIdentifier.of("db", "staged");
+ UpdateTableRequest request = stagedCreateRequest(staged);
+ Assertions.assertFalse(catalog.tableExists(staged));
+
+ CatalogHandlers.updateTable(catalog, staged,
IcebergCommitValidation.withValidation(request));
+ Assertions.assertTrue(catalog.tableExists(staged));
+
Assertions.assertTrue(SCHEMA.sameSchema(catalog.loadTable(staged).schema()));
+
+ // Keep the original existence requirement: retrying the create cannot
replace the table.
+ Assertions.assertThrows(
+ CommitFailedException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog, staged,
IcebergCommitValidation.withValidation(request)));
+ }
+
+ @Test
+ public void testStagedCreateIsValidatedBeforeCommit() {
+ TableIdentifier staged = TableIdentifier.of("db", "invalid_staged");
+ UpdateTableRequest create = stagedCreateRequest(staged);
+ List<MetadataUpdate> updates =
+ ImmutableList.<MetadataUpdate>builder()
+ .addAll(create.updates())
+ .add(
+ new MetadataUpdate.SetSnapshotRef(
+ SnapshotRef.MAIN_BRANCH, 999L, SnapshotRefType.TAG, null,
null, null))
+ .build();
+ UpdateTableRequest invalid = new UpdateTableRequest(create.requirements(),
updates);
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog, staged,
IcebergCommitValidation.withValidation(invalid)));
+ Assertions.assertFalse(catalog.tableExists(staged));
+ }
+
+ @Test
+ public void testStagedCreateRejectsSnapshotWithoutSummary() {
+ TableIdentifier staged = TableIdentifier.of("db", "invalid_snapshot");
+ UpdateTableRequest create = stagedCreateRequest(staged);
+ MetadataUpdate snapshot =
+ new MetadataUpdate.AddSnapshot(
+ SnapshotParser.fromJson(
+ "{\"snapshot-id\": 999, \"sequence-number\": 1,
\"timestamp-ms\": "
+ + System.currentTimeMillis()
+ + ", \"manifest-list\": \"/data/snap-999.avro\",
\"schema-id\": 0}"));
+ List<MetadataUpdate> updates =
+
ImmutableList.<MetadataUpdate>builder().addAll(create.updates()).add(snapshot).build();
+ UpdateTableRequest invalid = new UpdateTableRequest(create.requirements(),
updates);
+
+ ValidationException thrown =
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog, staged,
IcebergCommitValidation.withValidation(invalid)));
+ Assertions.assertTrue(thrown.getMessage().contains("summary"));
+ Assertions.assertFalse(catalog.tableExists(staged));
+ }
+
+ @Test
+ public void testRefCreatedAfterTheClientReadIsProtected() {
+ // An expire job read the table before a user tagged the snapshot it is
about to remove: the
+ // commit is checked against the metadata it is applied to, so the tag is
not lost.
+ UpdateTableRequest expire =
+ new UpdateTableRequest(
+ ImmutableList.of(
+ new
UpdateRequirement.AssertTableUUID(catalog.loadTable(IDENT).uuid().toString())),
+ ImmutableList.of(new
MetadataUpdate.RemoveSnapshots(firstSnapshot)));
+ catalog.loadTable(IDENT).manageSnapshots().createTag("audit",
firstSnapshot).commit();
+
+ Assertions.assertThrows(
+ CommitFailedException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog, IDENT,
IcebergCommitValidation.withValidation(expire)));
+
Assertions.assertTrue(catalog.loadTable(IDENT).refs().containsKey("audit"));
+ }
+
+ @Test
+ public void testWithValidationKeepsTheRequest() {
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ ImmutableList.of(new UpdateRequirement.AssertTableUUID("uuid")),
+ ImmutableList.of(new
MetadataUpdate.RemoveSnapshots(firstSnapshot)));
+ UpdateTableRequest validated =
IcebergCommitValidation.withValidation(request);
+
+ Assertions.assertEquals(1, request.requirements().size());
+ Assertions.assertEquals(2, validated.requirements().size());
+ Assertions.assertEquals(request.updates(), validated.updates());
+ Assertions.assertInstanceOf(IcebergCommitValidation.class,
validated.requirements().get(1));
+ }
+
+ @Test
+ public void testSchemaChangesMustKeepDataReadable() {
+ // id 1 is "id long" and the table has data.
+ commitFails(
+ ValidationException.class,
+
addSchema("{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"int\"}"));
+ commitFails(
+ ValidationException.class,
+
addSchema("{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"string\"}"));
+ commitFails(
+ ValidationException.class,
+ addSchema(
+
"{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":{\"type\":\"struct\","
+ +
"\"fields\":[{\"id\":9,\"name\":\"z\",\"required\":false,\"type\":\"long\"}]}}"));
+ // A new required field without a default.
+ commitFails(
+ ValidationException.class,
+ addSchema(
+ "{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"long\"},"
+ +
"{\"id\":2,\"name\":\"r\",\"required\":true,\"type\":\"string\"}"));
+ Assertions.assertEquals(1, catalog.loadTable(IDENT).schemas().size());
+
+ // Allowed: an optional new field with a new id.
+ commit(
+ addSchema(
+ "{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"long\"},"
+ +
"{\"id\":2,\"name\":\"s\",\"required\":false,\"type\":\"string\"}"));
+ Assertions.assertEquals(2, catalog.loadTable(IDENT).schemas().size());
+ }
+
+ @Test
+ public void testReusingADroppedFieldIdIsRefused() {
+ // Schema 1 adds b (id 2); schema 2 drops it; a later schema may not give
id 2 to another field.
+ commit(
+ addSchema(
+ "{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"long\"},"
+ +
"{\"id\":2,\"name\":\"b\",\"required\":false,\"type\":\"int\"}"),
+ new MetadataUpdate.SetCurrentSchema(-1));
+ // Going back to the original schema drops b again.
+ commit(new MetadataUpdate.SetCurrentSchema(0));
+ commitFails(
+ ValidationException.class,
+ addSchema(
+ "{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"long\"},"
+ +
"{\"id\":2,\"name\":\"c\",\"required\":false,\"type\":\"string\"}"));
+ }
+
+ @Test
+ public void testReplaceMayRedefineColumns() {
+ // A replace removes main, so no existing data is read with the new schema.
+ commit(
+ new MetadataUpdate.RemoveSnapshotRef(SnapshotRef.MAIN_BRANCH),
+
addSchema("{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"string\"}"),
+ new MetadataUpdate.SetCurrentSchema(-1));
+ Assertions.assertEquals(
+ Types.StringType.get(),
catalog.loadTable(IDENT).schema().findType("id"));
+ }
+
+ @Test
+ public void testPartitionFieldIdsAreChecked() {
+ Table partitioned = createPartitionedTable("p");
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "p");
+ // Spec 0 has 1000: s identity(2). The same id for another source field is
refused.
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ request(
+ addSpec(
+
"{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":1000}"))));
+ // A new field with an id below last-partition-id is refused.
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ request(
+ addSpec(
+
"{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":999}"))));
+ // Keeping 1000 for the same field and adding 1001 is valid.
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ request(
+ addSpec(
+
"{\"name\":\"s\",\"transform\":\"identity\",\"source-id\":2,\"field-id\":1000},"
+ +
"{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":1001}")));
+ Assertions.assertEquals(2, catalog.loadTable(ident).specs().size());
+ Assertions.assertNotNull(partitioned);
+ }
+
+ @Test
+ public void testRemovingASpecWithLiveManifestsIsRefused() {
+ Table partitioned = createPartitionedTable("rm");
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "rm");
+ append(partitioned, "/data/rm1.parquet", "s=a");
+ // Make a new default spec, so spec 0 is no longer the default but is
still used by data.
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ request(
+ addSpec(
+
"{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":1001}"),
+ new MetadataUpdate.SetDefaultPartitionSpec(-1)));
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ IcebergCommitValidation.withValidation(
+ new UpdateTableRequest(
+ ImmutableList.of(),
+ ImmutableList.of(
+ new
MetadataUpdate.RemovePartitionSpecs(ImmutableSet.of(0)))),
+ () -> catalog.loadTable(ident).io())));
+ Assertions.assertNotNull(catalog.loadTable(ident).specs().get(0));
+
+ // An unused spec can be removed.
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ request(
+ addSpec(
+
"{\"name\":\"s_b\",\"transform\":\"bucket[4]\",\"source-id\":2,\"field-id\":1002}")));
+ CatalogHandlers.updateTable(
+ catalog,
+ ident,
+ IcebergCommitValidation.withValidation(
+ new UpdateTableRequest(
+ ImmutableList.of(),
+ ImmutableList.of(new
MetadataUpdate.RemovePartitionSpecs(ImmutableSet.of(2)))),
+ () -> catalog.loadTable(ident).io()));
+ Assertions.assertNull(catalog.loadTable(ident).specs().get(2));
+ }
+
+ @Test
+ public void testUnknownTransformCannotBecomeTheDefault() {
+ MetadataUpdate zorder =
+ MetadataUpdateParser.fromJson(
+
"{\"action\":\"add-sort-order\",\"sort-order\":{\"order-id\":1,\"fields\":["
+ +
"{\"transform\":\"zorder\",\"source-id\":1,\"direction\":\"asc\","
+ + "\"null-order\":\"nulls-first\"}]}}");
+ commitFails(ValidationException.class, zorder, new
MetadataUpdate.SetDefaultSortOrder(-1));
+ Assertions.assertTrue(catalog.loadTable(IDENT).sortOrder().isUnsorted());
+ // Adding it without making it the default is allowed by the table spec.
+ commit(zorder);
+ }
+
+ @Test
+ public void testPartitionSpecWithoutSpecIdIsAccepted() throws Exception {
+ // The REST spec makes spec-id read-only; the server assigns the id.
+ CreateTableRequest create =
+ IcebergObjectMapper.getInstance()
+ .readValue(
+ "{\"name\":\"t\",\"schema\":"
+ + SchemaParser.toJson(SCHEMA)
+ + ",\"partition-spec\":{\"fields\":[{\"name\":\"id_b\","
+ + "\"transform\":\"bucket[4]\",\"source-id\":1}]}}",
+ CreateTableRequest.class);
+ Assertions.assertEquals(1, create.spec().fields().size());
+
+ UpdateTableRequest update =
+ IcebergObjectMapper.getInstance()
+ .readValue(
+
"{\"requirements\":[],\"updates\":[{\"action\":\"add-spec\",\"spec\":"
+ +
"{\"fields\":[{\"name\":\"id_b\",\"transform\":\"bucket[4]\","
+ + "\"source-id\":1,\"field-id\":1000}]}}]}",
+ UpdateTableRequest.class);
+ CatalogHandlers.updateTable(catalog, IDENT,
IcebergCommitValidation.withValidation(update));
+ // The new spec gets the next free id, not the placeholder.
+ Assertions.assertEquals(2, catalog.loadTable(IDENT).specs().size());
+ Assertions.assertTrue(catalog.loadTable(IDENT).specs().containsKey(1));
+ }
+
+ @Test
+ void testPartitionIdsCannotCollideWithinOneCommit() {
+ commitFails(
+ ValidationException.class,
+
addSpec("{\"name\":\"id\",\"transform\":\"identity\",\"source-id\":1,\"field-id\":1000}"),
+ addSpec(
+
"{\"name\":\"id_bucket\",\"transform\":\"bucket[4]\",\"source-id\":1,\"field-id\":1000}"));
+ Assertions.assertEquals(1, catalog.loadTable(IDENT).specs().size());
+ }
+
+ @Test
+ void testUnusedUnknownSortOrderDoesNotBlockSettingKnownDefault() {
+ MetadataUpdate unknown =
+ MetadataUpdateParser.fromJson(
+
"{\"action\":\"add-sort-order\",\"sort-order\":{\"order-id\":1,\"fields\":["
+ +
"{\"transform\":\"zorder\",\"source-id\":1,\"direction\":\"asc\","
+ + "\"null-order\":\"nulls-first\"}]}}");
+ commit(unknown, new MetadataUpdate.SetDefaultSortOrder(0));
+ Assertions.assertTrue(catalog.loadTable(IDENT).sortOrder().isUnsorted());
+ Assertions.assertEquals(2, catalog.loadTable(IDENT).sortOrders().size());
+ }
+
+ @Test
+ void testStagedCreateCannotUseUnknownDefaultTransform() {
+ TableIdentifier staged = TableIdentifier.of("db", "unknown_transform");
+ UpdateTableRequest create = stagedCreateRequest(staged);
+ List<MetadataUpdate> updates =
+ ImmutableList.<MetadataUpdate>builder()
+ .addAll(create.updates())
+ .add(
+ addSpec(
+
"{\"name\":\"unknown\",\"transform\":\"custom\",\"source-id\":1,\"field-id\":1000}"))
+ .add(new MetadataUpdate.SetDefaultPartitionSpec(-1))
+ .build();
+ UpdateTableRequest request = new UpdateTableRequest(create.requirements(),
updates);
+
+ Assertions.assertThrows(
+ ValidationException.class,
+ () ->
+ CatalogHandlers.updateTable(
+ catalog, staged,
IcebergCommitValidation.withValidation(request)));
+ Assertions.assertFalse(catalog.tableExists(staged));
+ }
+
+ @Test
+ void testRemovingAndRestoringMainCannotBypassSchemaChecks() {
+ commitFails(
+ ValidationException.class,
+ new MetadataUpdate.RemoveSnapshotRef(SnapshotRef.MAIN_BRANCH),
+
addSchema("{\"id\":1,\"name\":\"id\",\"required\":true,\"type\":\"string\"}"),
+ new MetadataUpdate.SetCurrentSchema(-1),
+ new MetadataUpdate.SetSnapshotRef(
+ SnapshotRef.MAIN_BRANCH, secondSnapshot, SnapshotRefType.BRANCH,
null, null, null));
+ Assertions.assertEquals(Types.LongType.get(),
catalog.loadTable(IDENT).schema().findType("id"));
+ Assertions.assertEquals(
+ secondSnapshot,
catalog.loadTable(IDENT).currentSnapshot().snapshotId());
+ }
+
+ private Table createPartitionedTable(String name) {
+ Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "s", Types.StringType.get()));
+ return catalog.createTable(
+ TableIdentifier.of(Namespace.of("db"), name),
+ schema,
+ PartitionSpec.builderFor(schema).identity("s").build(),
+ ImmutableMap.of("format-version", "2"));
+ }
+
+ private static MetadataUpdate addSchema(String fields) {
+ return new MetadataUpdate.AddSchema(
+
SchemaParser.fromJson("{\"type\":\"struct\",\"schema-id\":7,\"fields\":[" +
fields + "]}"));
+ }
+
+ private static MetadataUpdate addSpec(String fields) {
+ return MetadataUpdateParser.fromJson(
+ "{\"action\":\"add-spec\",\"spec\":{\"spec-id\":0,\"fields\":[" +
fields + "]}}");
+ }
+
+ private static void append(Table table, String path, String partition) {
+ table
+ .newFastAppend()
+ .appendFile(
+ DataFiles.builder(table.spec())
+ .withPath(path)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .withPartitionPath(partition)
+ .build())
+ .commit();
+ }
+
+ private MetadataUpdate addSnapshot(String extraFields) {
+ String json =
+ "{\"snapshot-id\": 999, \"parent-snapshot-id\": "
+ + secondSnapshot
+ + ", \"sequence-number\": 3, \"timestamp-ms\": "
+ + System.currentTimeMillis()
+ + ", \"manifest-list\": \"/data/snap-999.avro\""
+ + extraFields
+ + "}";
+ return new MetadataUpdate.AddSnapshot(SnapshotParser.fromJson(json));
+ }
+
+ private UpdateTableRequest stagedCreateRequest(TableIdentifier identifier) {
+ TableMetadata metadata =
+ CatalogHandlers.stageTableCreate(
+ catalog,
+ identifier.namespace(),
+ CreateTableRequest.builder()
+ .withName(identifier.name())
+ .withSchema(SCHEMA)
+ .setProperties(ImmutableMap.of("format-version", "2"))
+ .stageCreate()
+ .build())
+ .tableMetadata();
+ return new UpdateTableRequest(
+ ImmutableList.of(new UpdateRequirement.AssertTableDoesNotExist()),
metadata.changes());
+ }
+
+ private void commit(MetadataUpdate... updates) {
+ CatalogHandlers.updateTable(catalog, IDENT, request(updates));
+ }
+
+ private void commitFails(Class<? extends RuntimeException> expected,
MetadataUpdate... updates) {
+ Assertions.assertThrows(
+ expected, () -> CatalogHandlers.updateTable(catalog, IDENT,
request(updates)));
+ }
+
+ private static UpdateTableRequest request(MetadataUpdate... updates) {
+ List<MetadataUpdate> list = ImmutableList.copyOf(updates);
+ return IcebergCommitValidation.withValidation(new
UpdateTableRequest(ImmutableList.of(), list));
+ }
+
+ private static void append(Table table, String path) {
+ table
+ .newFastAppend()
+ .appendFile(
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(path)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .build())
+ .commit();
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
index ee56ba31b9..ac57441f55 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
@@ -50,6 +50,7 @@ import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.rest.RESTRequest;
import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.ErrorResponse;
import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
@@ -473,6 +474,22 @@ public class TestIcebergRESTUtils {
}
}
+ @Test
+ void testCloneFallsBackToOriginalWhenIcebergCannotSerialize() throws
Exception {
+ // A partition field without its optional field-id parses, but Iceberg
cannot write it again.
+ UpdateTableRequest request =
+ IcebergObjectMapper.getInstance()
+ .readValue(
+ "{\"requirements\": [], \"updates\": [{\"action\":
\"add-spec\", \"spec\":"
+ + " {\"spec-id\": 1, \"fields\": [{\"name\":
\"id_bucket\", \"transform\": \"bucket[4]\","
+ + " \"source-id\": 1}]}}]}",
+ UpdateTableRequest.class);
+ UpdateTableRequest copy =
+ Assertions.assertDoesNotThrow(
+ () -> IcebergRESTUtils.cloneIcebergRESTObject(request,
UpdateTableRequest.class));
+ Assertions.assertEquals(request.updates().size(), copy.updates().size());
+ }
+
@Test
void testValidateRequestRejectsMissingOrInvalidBody() {
BadRequestException missing =
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceHookDispatcher.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceHookDispatcher.java
index 02bd730e1e..a78ecbe90e 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceHookDispatcher.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergNamespaceHookDispatcher.java
@@ -51,6 +51,7 @@ import
org.apache.gravitino.listener.api.event.IcebergRequestContext;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.lock.TreeLock;
import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.exceptions.BadRequestException;
import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
import org.apache.iceberg.rest.requests.ImmutableRegisterViewRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
@@ -226,6 +227,50 @@ public class TestIcebergNamespaceHookDispatcher {
verify(mockOwnerDispatcher, never()).setOwners(any(), any(), any(), any());
}
+ @Test
+ public void testCreateNamespaceWithDottedLevelIsRejectedBeforeCreating() {
+ CreateNamespaceRequest request = mock(CreateNamespaceRequest.class);
+ when(request.namespace()).thenReturn(Namespace.of("gi_n2l.dot"));
+
+ Assertions.assertThrows(
+ BadRequestException.class, () ->
hookDispatcher.createNamespace(mockContext, request));
+ verify(mockDispatcher, never()).createNamespace(any(), any());
+ }
+
+ @Test
+ public void testCreateNamespaceWithTooLongNameIsRejectedBeforeCreating() {
+ // Ten levels of 20 characters: each level is short, the full name is 209
characters.
+ String[] levels = new String[10];
+ for (int i = 0; i < levels.length; i++) {
+ levels[i] = String.format("level_%014d", i);
+ }
+ CreateNamespaceRequest request = mock(CreateNamespaceRequest.class);
+ when(request.namespace()).thenReturn(Namespace.of(levels));
+
+ Assertions.assertThrows(
+ BadRequestException.class, () ->
hookDispatcher.createNamespace(mockContext, request));
+ verify(mockDispatcher, never()).createNamespace(any(), any());
+ }
+
+ @Test
+ public void testCreateNamespaceIsNotDroppedWhenImportFails() {
+ Namespace namespace = Namespace.of("test_schema");
+ CreateNamespaceRequest request = mock(CreateNamespaceRequest.class);
+ when(request.namespace()).thenReturn(namespace);
+ when(mockDispatcher.createNamespace(mockContext, request))
+ .thenReturn(mock(CreateNamespaceResponse.class));
+ RuntimeException importFailure =
+ new RuntimeException("Failed to import schema entity to the store");
+
when(mockInternalSchemaDispatcher.loadSchema(any())).thenThrow(importFailure);
+
+ RuntimeException thrown =
+ Assertions.assertThrows(
+ RuntimeException.class, () ->
hookDispatcher.createNamespace(mockContext, request));
+ Assertions.assertSame(importFailure, thrown);
+ verify(mockDispatcher, never()).dropNamespace(any(), any());
+ verify(mockInternalOwnerDispatcher, never()).setOwners(any(), any(),
any(), any());
+ }
+
@Test
public void testRegisterTableThrowsWhenSetOwnerFails() {
Namespace namespace = Namespace.of("test_schema");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableHookDispatcher.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableHookDispatcher.java
index 27cf388ffd..7cdd799a02 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableHookDispatcher.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergTableHookDispatcher.java
@@ -19,8 +19,10 @@
package org.apache.gravitino.iceberg.service.dispatcher;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -46,6 +48,9 @@ import
org.apache.gravitino.iceberg.common.utils.IcebergIdentifierUtils;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.lock.LockType;
+import org.apache.gravitino.lock.TreeLock;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.TableEntity;
import org.apache.iceberg.Schema;
@@ -63,6 +68,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
public class TestIcebergTableHookDispatcher {
@@ -73,6 +79,9 @@ public class TestIcebergTableHookDispatcher {
new Schema(NestedField.required(1, "test_field", StringType.get()));
private IcebergTableHookDispatcher hookDispatcher;
+ private LockManager mockLockManager;
+ private TreeLock mockTreeLock;
+ private LockManager previousLockManager;
private IcebergTableOperationDispatcher mockDispatcher;
private IcebergNamespaceOperationDispatcher mockNamespaceDispatcher;
private EntityStore mockEntityStore;
@@ -126,6 +135,13 @@ public class TestIcebergTableHookDispatcher {
FieldUtils.writeField(
GravitinoEnv.getInstance(), "internalOwnerDispatcher",
mockInternalOwnerDispatcher, true);
+ previousLockManager =
+ (LockManager) FieldUtils.readField(GravitinoEnv.getInstance(),
"lockManager", true);
+ mockLockManager = mock(LockManager.class);
+ mockTreeLock = mock(TreeLock.class);
+ when(mockLockManager.createTreeLock(any())).thenReturn(mockTreeLock);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
mockLockManager, true);
+
// Create mock IcebergRESTServerContext
IcebergConfigProvider mockConfigProvider =
mock(IcebergConfigProvider.class);
when(mockConfigProvider.getMetalakeName()).thenReturn(TEST_METALAKE);
@@ -162,6 +178,8 @@ public class TestIcebergTableHookDispatcher {
previousInternalOwnerDispatcher,
true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
previousLockManager, true);
+
// Reset IcebergRESTServerContext singleton
Class<?> holderClass =
Arrays.stream(IcebergRESTServerContext.class.getDeclaredClasses())
@@ -171,6 +189,47 @@ public class TestIcebergTableHookDispatcher {
FieldUtils.writeStaticField(holderClass, "INSTANCE", null, true);
}
+ @Test
+ public void testCreateTableHoldsTheNamespaceBranchLockForReading() {
+ // dropNamespace holds the same lock for writing, so a create and a drop
of one namespace are
+ // serialized on this server.
+ Namespace namespace = Namespace.of("A", "B");
+ CreateTableRequest request =
+
CreateTableRequest.builder().withName("t").withSchema(TABLE_SCHEMA).build();
+ when(mockDispatcher.createTable(mockContext, namespace, request))
+ .thenReturn(mock(LoadTableResponse.class));
+
+ hookDispatcher.createTable(mockContext, namespace, request);
+
+ ArgumentCaptor<NameIdentifier> lockId =
ArgumentCaptor.forClass(NameIdentifier.class);
+ verify(mockLockManager).createTreeLock(lockId.capture());
+ Assertions.assertArrayEquals(
+ new String[] {TEST_METALAKE, TEST_CATALOG},
lockId.getValue().namespace().levels());
+ Assertions.assertEquals("A", lockId.getValue().name());
+ InOrder order = inOrder(mockTreeLock, mockDispatcher);
+ order.verify(mockTreeLock).lock(LockType.READ);
+ order.verify(mockDispatcher).createTable(mockContext, namespace, request);
+ order.verify(mockTreeLock).unlock();
+ }
+
+ @Test
+ void testStagedCreateCommitHoldsNamespaceBranchLock() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("A", "B"),
"staged");
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ List.of(new UpdateRequirement.AssertTableDoesNotExist()),
Collections.emptyList());
+ when(mockDispatcher.updateTable(mockContext, table, request))
+ .thenReturn(mock(LoadTableResponse.class));
+
+ hookDispatcher.updateTable(mockContext, table, request);
+
+ verify(mockLockManager).createTreeLock(NameIdentifier.of(TEST_METALAKE,
TEST_CATALOG, "A"));
+ InOrder order = inOrder(mockTreeLock, mockDispatcher);
+ order.verify(mockTreeLock).lock(LockType.READ);
+ order.verify(mockDispatcher).updateTable(mockContext, table, request);
+ order.verify(mockTreeLock).unlock();
+ }
+
@Test
public void testCreateTableCallsImportAndOwnership() {
Namespace namespace = Namespace.of("test_schema");
@@ -200,6 +259,47 @@ public class TestIcebergTableHookDispatcher {
Assertions.assertEquals(TEST_USER, userCaptor.getValue());
}
+ @Test
+ public void testCreateTableIsNotDroppedWhenImportFails() {
+ Namespace namespace = Namespace.of("test_schema");
+ CreateTableRequest request =
+
CreateTableRequest.builder().withName("test_table").withSchema(TABLE_SCHEMA).build();
+ when(mockDispatcher.createTable(mockContext, namespace, request))
+ .thenReturn(mock(LoadTableResponse.class));
+ RuntimeException importFailure = new RuntimeException("Failed to import
table entity");
+
when(mockInternalTableDispatcher.loadTable(any())).thenThrow(importFailure);
+
+ RuntimeException thrown =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> hookDispatcher.createTable(mockContext, namespace, request));
+ Assertions.assertSame(importFailure, thrown);
+
+ verify(mockDispatcher, never()).dropTable(any(), any(), anyBoolean());
+ verify(mockInternalOwnerDispatcher, never()).setOwner(any(), any(), any(),
any());
+ }
+
+ @Test
+ public void testStagedCreateCommitIsNotDroppedWhenImportFails() {
+ TableIdentifier tableId = TableIdentifier.of("test_schema",
"staged_table");
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ Collections.singletonList(new
UpdateRequirement.AssertTableDoesNotExist()),
+ Collections.emptyList());
+ when(mockDispatcher.updateTable(mockContext, tableId, request))
+ .thenReturn(mock(LoadTableResponse.class));
+ RuntimeException importFailure = new RuntimeException("Failed to import
table entity");
+
when(mockInternalTableDispatcher.loadTable(any())).thenThrow(importFailure);
+
+ RuntimeException thrown =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> hookDispatcher.updateTable(mockContext, tableId, request));
+ Assertions.assertSame(importFailure, thrown);
+ verify(mockDispatcher, never()).dropTable(any(), any(), anyBoolean());
+ verify(mockInternalOwnerDispatcher, never()).setOwner(any(), any(), any(),
any());
+ }
+
@Test
public void testDropTableDeletesEntity() throws IOException {
TableIdentifier tableId = TableIdentifier.of("test_schema", "test_table");
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java
index fb491d260a..9db34be734 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java
@@ -323,10 +323,23 @@ public class TestIcebergNamespaceOperations extends
IcebergNamespaceTestBase {
doCreateNamespace(Namespace.of("page_ns2"));
doCreateNamespace(Namespace.of("page_ns3"));
- // First page: pageSize=2, no pageToken
- Response firstPageResponse =
+ // pageSize alone must not truncate the list.
+ try (Response response =
getNamespaceClientBuilder(
Optional.empty(), Optional.empty(),
Optional.of(ImmutableMap.of("pageSize", "2")))
+ .get()) {
+ Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+ ListNamespacesResponse all =
response.readEntity(ListNamespacesResponse.class);
+ Assertions.assertEquals(3, all.namespaces().size());
+ Assertions.assertNull(all.nextPageToken());
+ }
+
+ // First page: an empty pageToken starts paging
+ Response firstPageResponse =
+ getNamespaceClientBuilder(
+ Optional.empty(),
+ Optional.empty(),
+ Optional.of(ImmutableMap.of("pageToken", "", "pageSize", "2")))
.get();
Assertions.assertEquals(Status.OK.getStatusCode(),
firstPageResponse.getStatus());
ListNamespacesResponse firstPage =
firstPageResponse.readEntity(ListNamespacesResponse.class);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergPaginationHelper.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergPaginationHelper.java
index 63804c9752..5558d98d4a 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergPaginationHelper.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergPaginationHelper.java
@@ -50,7 +50,7 @@ class TestIcebergPaginationHelper {
.add(Namespace.of("ns3"))
.build();
ListNamespacesResponse result =
- IcebergPaginationHelper.paginateNamespaces(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateNamespaces(response, Optional.of(""),
Optional.of(2));
Assertions.assertEquals(2, result.namespaces().size());
Assertions.assertNotNull(result.nextPageToken());
}
@@ -65,7 +65,7 @@ class TestIcebergPaginationHelper {
.build();
// First page
ListNamespacesResponse firstPage =
- IcebergPaginationHelper.paginateNamespaces(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateNamespaces(response, Optional.of(""),
Optional.of(2));
// Second page using cursor from first page
ListNamespacesResponse secondPage =
IcebergPaginationHelper.paginateNamespaces(
@@ -84,7 +84,7 @@ class TestIcebergPaginationHelper {
.add(Namespace.of("ns2"))
.build();
ListNamespacesResponse result =
- IcebergPaginationHelper.paginateNamespaces(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateNamespaces(response, Optional.of(""),
Optional.of(2));
List<String> names =
result.namespaces().stream().map(Namespace::toString).collect(Collectors.toList());
Assertions.assertEquals(Arrays.asList("ns1", "ns2"), names);
@@ -125,7 +125,7 @@ class TestIcebergPaginationHelper {
Assertions.assertThrows(
IllegalArgumentException.class,
() ->
- IcebergPaginationHelper.paginateNamespaces(response,
Optional.empty(), Optional.of(0)));
+ IcebergPaginationHelper.paginateNamespaces(response,
Optional.of(""), Optional.of(0)));
}
@Test
@@ -135,15 +135,14 @@ class TestIcebergPaginationHelper {
Assertions.assertThrows(
IllegalArgumentException.class,
() ->
- IcebergPaginationHelper.paginateNamespaces(
- response, Optional.empty(), Optional.of(-1)));
+ IcebergPaginationHelper.paginateNamespaces(response,
Optional.of(""), Optional.of(-1)));
}
@Test
void testPaginateNamespacesEmptyList() {
ListNamespacesResponse response = ListNamespacesResponse.builder().build();
ListNamespacesResponse result =
- IcebergPaginationHelper.paginateNamespaces(response, Optional.empty(),
Optional.of(10));
+ IcebergPaginationHelper.paginateNamespaces(response, Optional.of(""),
Optional.of(10));
Assertions.assertEquals(0, result.namespaces().size());
Assertions.assertNull(result.nextPageToken());
}
@@ -158,7 +157,7 @@ class TestIcebergPaginationHelper {
.add(TableIdentifier.of(ns, "t3"))
.build();
ListTablesResponse result =
- IcebergPaginationHelper.paginateTables(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateTables(response, Optional.of(""),
Optional.of(2));
Assertions.assertEquals(2, result.identifiers().size());
Assertions.assertNotNull(result.nextPageToken());
}
@@ -177,7 +176,7 @@ class TestIcebergPaginationHelper {
// Walk all pages with pageSize=2
ListTablesResponse page1 =
- IcebergPaginationHelper.paginateTables(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateTables(response, Optional.of(""),
Optional.of(2));
Assertions.assertEquals(2, page1.identifiers().size());
Assertions.assertNotNull(page1.nextPageToken());
@@ -209,8 +208,70 @@ class TestIcebergPaginationHelper {
.build();
// pageSize equals total items
ListTablesResponse result =
- IcebergPaginationHelper.paginateTables(response, Optional.empty(),
Optional.of(2));
+ IcebergPaginationHelper.paginateTables(response, Optional.of(""),
Optional.of(2));
Assertions.assertEquals(2, result.identifiers().size());
Assertions.assertNull(result.nextPageToken());
}
+
+ @Test
+ void testPageSizeWithoutPageTokenReturnsAllResults() {
+ // The REST spec: without a pageToken parameter the server returns every
result and no
+ // next-page token, whatever pageSize says.
+ ListNamespacesResponse namespaces =
+ ListNamespacesResponse.builder()
+ .addAll(Arrays.asList(Namespace.of("c"), Namespace.of("a"),
Namespace.of("b")))
+ .build();
+ ListNamespacesResponse all =
+ IcebergPaginationHelper.paginateNamespaces(namespaces,
Optional.empty(), Optional.of(1));
+ Assertions.assertEquals(3, all.namespaces().size());
+ Assertions.assertNull(all.nextPageToken());
+
+ ListTablesResponse tables =
+ ListTablesResponse.builder()
+ .add(TableIdentifier.of("db", "t1"))
+ .add(TableIdentifier.of("db", "t2"))
+ .build();
+ ListTablesResponse allTables =
+ IcebergPaginationHelper.paginateTables(tables, Optional.empty(),
Optional.of(1));
+ Assertions.assertEquals(2, allTables.identifiers().size());
+ Assertions.assertNull(allTables.nextPageToken());
+
+ // An empty pageToken starts paging.
+ ListNamespacesResponse firstPage =
+ IcebergPaginationHelper.paginateNamespaces(namespaces,
Optional.of(""), Optional.of(1));
+ Assertions.assertEquals(1, firstPage.namespaces().size());
+ Assertions.assertEquals("a", firstPage.nextPageToken());
+ }
+
+ @Test
+ void testEmptyPageTokenWithoutPageSizeReturnsAllUnsorted() {
+ ListNamespacesResponse namespaces =
+ ListNamespacesResponse.builder()
+ .addAll(Arrays.asList(Namespace.of("c"), Namespace.of("a"),
Namespace.of("b")))
+ .build();
+ ListNamespacesResponse all =
+ IcebergPaginationHelper.paginateNamespaces(namespaces,
Optional.of(""), Optional.empty());
+ Assertions.assertEquals(
+ Arrays.asList(Namespace.of("c"), Namespace.of("a"),
Namespace.of("b")), all.namespaces());
+ Assertions.assertNull(all.nextPageToken());
+ }
+
+ @Test
+ void testInvalidPageSizeWithoutPageTokenIsRejected() {
+ ListNamespacesResponse namespaces =
+ ListNamespacesResponse.builder().add(Namespace.of("db")).build();
+ ListTablesResponse tables =
+ ListTablesResponse.builder().add(TableIdentifier.of("db",
"table")).build();
+ for (int size : new int[] {0, -1}) {
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergPaginationHelper.paginateNamespaces(
+ namespaces, Optional.empty(), Optional.of(size)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ IcebergPaginationHelper.paginateTables(tables, Optional.empty(),
Optional.of(size)));
+ }
+ }
}
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
index 528c9ee699..16641ffc41 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java
@@ -411,11 +411,22 @@ public class TestIcebergTableOperations extends
IcebergNamespaceTestBase {
dummyEventListener.clearEvent();
- // First page: pageSize=2
String tablePath =
IcebergRestTestUtil.NAMESPACE_PATH + "/" +
RESTUtil.encodeNamespace(namespace) + "/tables";
+ // pageSize alone must not truncate the list.
+ try (Response response =
+ getIcebergClientBuilder(tablePath,
Optional.of(ImmutableMap.of("pageSize", "2"))).get()) {
+ Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+ ListTablesResponse all = response.readEntity(ListTablesResponse.class);
+ Assertions.assertEquals(3, all.identifiers().size());
+ Assertions.assertNull(all.nextPageToken());
+ }
+
+ // An empty pageToken starts paging.
Response firstPageResponse =
- getIcebergClientBuilder(tablePath,
Optional.of(ImmutableMap.of("pageSize", "2"))).get();
+ getIcebergClientBuilder(
+ tablePath, Optional.of(ImmutableMap.of("pageToken", "",
"pageSize", "2")))
+ .get();
Assertions.assertEquals(Status.OK.getStatusCode(),
firstPageResponse.getStatus());
ListTablesResponse firstPage =
firstPageResponse.readEntity(ListTablesResponse.class);
Assertions.assertEquals(2, firstPage.identifiers().size());
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergViewOperations.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergViewOperations.java
index e178df8567..7a92b7a112 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergViewOperations.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergViewOperations.java
@@ -152,11 +152,22 @@ public class TestIcebergViewOperations extends
IcebergNamespaceTestBase {
dummyEventListener.clearEvent();
- // First page: pageSize=2
String viewPath =
IcebergRestTestUtil.NAMESPACE_PATH + "/" +
RESTUtil.encodeNamespace(namespace) + "/views";
+ // pageSize alone must not truncate the list.
+ try (Response response =
+ getIcebergClientBuilder(viewPath,
Optional.of(ImmutableMap.of("pageSize", "2"))).get()) {
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
response.getStatus());
+ ListTablesResponse all = response.readEntity(ListTablesResponse.class);
+ Assertions.assertEquals(3, all.identifiers().size());
+ Assertions.assertNull(all.nextPageToken());
+ }
+
+ // An empty pageToken starts paging.
Response firstPageResponse =
- getIcebergClientBuilder(viewPath,
Optional.of(ImmutableMap.of("pageSize", "2"))).get();
+ getIcebergClientBuilder(
+ viewPath, Optional.of(ImmutableMap.of("pageToken", "",
"pageSize", "2")))
+ .get();
Assertions.assertEquals(Response.Status.OK.getStatusCode(),
firstPageResponse.getStatus());
ListTablesResponse firstPage =
firstPageResponse.readEntity(ListTablesResponse.class);
Assertions.assertEquals(2, firstPage.identifiers().size());
