This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 93778817fa [Cherry-pick to branch-1.3] [#14438] fix(iceberg-rest):
validate commits and fix pagination and purge (#14533)
93778817fa is described below
commit 93778817fad5e05a4da5064b49c0b72fe5d1a6cc
Author: roryqi <[email protected]>
AuthorDate: Mon Nov 2 21:29:52 2026 +0800
[Cherry-pick to branch-1.3] [#14438] fix(iceberg-rest): validate commits
and fix pagination and purge (#14533)
### What changes were proposed in this pull request?
Backport #14530 (merged commit `c150be2456`) to `branch-1.3`, covering
#14439, #14455, #14458, and #14466.
- Validate namespace names, registration UUIDs, snapshots,
schema/partition evolution, and default transforms before committing.
Preserve staged-create validation and accept partition specs without
`spec-id`.
- Return all list results without a page token, serialize direct/staged
table creation with namespace drop, and make explicit purge clean up
reachable files while respecting `gc.enabled=false`.
- Preserve import errors without automatically deleting tables or
namespaces.
Resolve conflicts while retaining branch-1.3's scan-filter deserializer,
deletion-vector checks, audit/ownership updates, and namespace identity
handling. Adapt the registration test to the OSS JDBC backend's existing
rejection of overwrite requests.
### Why are the changes needed?
Bring the consolidated validation, pagination, and purge fixes from
#14530 to the 1.3 release branch.
### Does this PR introduce _any_ user-facing change?
Yes, the same behavior changes as #14530: invalid metadata is rejected
before commit, lists without a token are complete, and import failures
do not trigger deletion. Explicit purge preserves content files when GC
is disabled.
### How was this patch tested?
- `./gradlew :iceberg:iceberg-common:spotlessApply
:iceberg:iceberg-rest-server:spotlessApply`
- `./gradlew :iceberg:iceberg-common:test
:iceberg:iceberg-rest-server:test -PskipITs`
- 68 common tests and 485 REST server tests passed, with no failures or
skips.
- `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 | 54 +-
.../iceberg/service/IcebergRESTUtils.java | 23 +-
.../dispatcher/IcebergNamespaceHookDispatcher.java | 34 +-
.../dispatcher/IcebergTableHookDispatcher.java | 25 +-
.../service/rest/IcebergPaginationHelper.java | 30 +-
.../iceberg/service/TestCatalogWrapperForREST.java | 72 +++
.../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, 1813 insertions(+), 39 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 231bc58424..d60bb881b2 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
@@ -44,11 +44,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;
@@ -286,7 +288,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 b5b877564e..b9f7be6fa3 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.ServiceFailureException;
import org.apache.iceberg.exceptions.ServiceUnavailableException;
import org.apache.iceberg.exceptions.ValidationException;
@@ -68,6 +69,7 @@ import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.requests.CreateTableRequest;
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.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -138,9 +140,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
@@ -697,4 +731,28 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
// Iceberg's binding errors append the whole table struct; it is not
useful to a client.
return message == null ? null : message.replaceAll(" in struct: .*$", "");
}
+
+ 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 ed530ecef3..d909ffb6c1 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
@@ -31,11 +31,17 @@ import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.databind.node.BooleanNode;
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.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.PlanTableScanRequestParser;
+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() {}
@@ -46,10 +52,12 @@ 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 it takes precedence for the
same type.
- mapper.registerModule(
- new SimpleModule("gravitino-iceberg-scan-request")
- .addDeserializer(PlanTableScanRequest.class, new
PlanTableScanRequestDeserializer()));
+ // Registered after Iceberg's serializers, so these take precedence for
the same types.
+ SimpleModule module = new SimpleModule("gravitino-iceberg-rest");
+ module.addDeserializer(PlanTableScanRequest.class, new
PlanTableScanRequestDeserializer());
+ module.addDeserializer(UnboundPartitionSpec.class, new
PartitionSpecDeserializer());
+ module.addDeserializer(UpdateTableRequest.class, new
UpdateTableRequestDeserializer());
+ mapper.registerModule(module);
return mapper;
}
@@ -92,6 +100,44 @@ public class IcebergObjectMapper {
}
}
+ /**
+ * 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 7b2998caaa..1568696a52 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
@@ -472,12 +472,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 6cc7169a82..3210fb06f8 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,10 +18,12 @@
*/
package org.apache.gravitino.iceberg.service.dispatcher;
+import com.google.common.annotations.VisibleForTesting;
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.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.SchemaDispatcher;
@@ -34,6 +36,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.UpdateNamespacePropertiesRequest;
@@ -64,6 +67,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
@@ -81,7 +85,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
@@ -103,6 +108,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 9a6689bdc4..5da70a00e2 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
@@ -31,6 +31,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());
} else if (changesSchema(updateTableRequest)) {
@@ -221,9 +235,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);
IcebergEntityAuditUtils.recordCreator(
IcebergIdentifierUtils.toGravitinoTableIdentifier(
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 9d94ef4942..1ba31b8f05 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.Collections;
import java.util.List;
@@ -45,6 +46,7 @@ import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
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;
@@ -59,6 +61,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;
@@ -77,6 +80,7 @@ import
org.apache.iceberg.rest.responses.LoadTableResponseParser;
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 {
@@ -433,6 +437,74 @@ public class TestCatalogWrapperForREST {
withProvider.loadTable(table, false, CredentialPrivilege.READ),
true));
}
+ @Test
+ void testRegisterOverwriteWithAnotherTablesMetadataIsRefused(@TempDir Path
warehouse)
+ throws Exception {
+ 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());
+ try (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());
+
+ // Matching metadata passes UUID validation. The OSS JDBC backend on
branch-1.3 still rejects
+ // overwrite, so preserve that backend response rather than adding
overwrite support here.
+ RegisterTableRequest sameTable =
+ ImmutableRegisterTableRequest.builder()
+ .name("target")
+ .metadataLocation(targetMetadata)
+ .overwrite(true)
+ .build();
+ UnsupportedOperationException unsupported =
+ Assertions.assertThrows(
+ UnsupportedOperationException.class,
+ () -> wrapper.registerTable(namespace, sameTable));
+ Assertions.assertEquals(
+ "Registering tables with overwrite is not supported",
unsupported.getMessage());
+ Assertions.assertEquals(
+ targetMetadata,
+ ((BaseTable)
catalog.loadTable(target)).operations().current().metadataFileLocation());
+ }
+ }
+
@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 6253f57edc..41973deebc 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.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -452,6 +453,22 @@ public class TestIcebergRESTUtils {
rewritten.credentials().get(0).config().get("client.refresh-credentials-endpoint"));
}
+ @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 3a7b4ea66e..45f31c6713 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
@@ -50,6 +50,7 @@ import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.lock.TreeLock;
import org.apache.gravitino.meta.SchemaEntity;
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.RegisterTableRequest;
import org.apache.iceberg.rest.responses.CreateNamespaceResponse;
@@ -204,6 +205,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 e0ad27dfc6..9b873c47ce 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;
@@ -48,6 +50,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.MetadataUpdate;
@@ -66,6 +71,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 {
@@ -76,6 +82,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;
@@ -129,6 +138,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);
@@ -165,6 +181,8 @@ public class TestIcebergTableHookDispatcher {
previousInternalOwnerDispatcher,
true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
previousLockManager, true);
+
// Reset IcebergRESTServerContext singleton
Class<?> holderClass =
Arrays.stream(IcebergRESTServerContext.class.getDeclaredClasses())
@@ -174,6 +192,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");
@@ -203,6 +262,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 156539ffa0..dc4c890ae3 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 5226ee10ef..8800da59d2 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
@@ -336,11 +336,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 4cb65b64ce..a1f2994009 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
@@ -147,11 +147,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());
