This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 995f6a0640 [#10986][#10985] fix(iceberg): Fix the issue of creating
remote IRC catalog tables using Trino (#10989)
995f6a0640 is described below
commit 995f6a06403b1d90e80941d1464e26a512a5ae73
Author: roryqi <[email protected]>
AuthorDate: Mon May 11 09:57:44 2026 +0800
[#10986][#10985] fix(iceberg): Fix the issue of creating remote IRC catalog
tables using Trino (#10989)
### What changes were proposed in this pull request?
Fix the issue of creating remote IRC catalog tables using Trino
### Why are the changes needed?
Fix:
#10986
#10985
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests.
---
.../iceberg/service/CatalogWrapperForREST.java | 150 +++++++++++++-
.../iceberg/service/TestCatalogWrapperForREST.java | 222 ++++++++++++++++++++-
2 files changed, 365 insertions(+), 7 deletions(-)
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 18980ea857..bc0e456410 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
@@ -20,6 +20,7 @@
package org.apache.gravitino.iceberg.service;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
@@ -55,17 +56,23 @@ import org.apache.gravitino.utils.MapUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.iceberg.BaseMetadataTable;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.BaseTransaction;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.IncrementalAppendScan;
+import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Scan;
import org.apache.iceberg.ScanTaskParser;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateRequirement;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -99,6 +106,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
private static final String DATA_ACCESS_VENDED_CREDENTIALS =
"vended-credentials";
private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing";
+ private static final Schema EMPTY_SCHEMA = new Schema();
private static final Set<String> catalogPropertiesToClientKeys =
ImmutableSet.of(
@@ -176,7 +184,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
public LoadTableResponse updateTable(
TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
if (isRESTCatalog()) {
- return CatalogHandlers.updateTable(getCatalog(), tableIdentifier,
updateTableRequest);
+ return tableUpdateInternal(tableIdentifier, updateTableRequest);
} else {
return super.updateTable(tableIdentifier, updateTableRequest);
}
@@ -735,6 +743,51 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return
LoadTableResponse.builder().withTableMetadata(metadata).addAllConfig(config).build();
}
+ private LoadTableResponse tableUpdateInternal(TableIdentifier ident,
UpdateTableRequest request) {
+ if (isCreate(request)) {
+ // this is a hacky way to get TableOperations for an uncommitted table
+ Optional<Integer> formatVersion =
+ request.updates().stream()
+ .filter(update -> update instanceof
MetadataUpdate.UpgradeFormatVersion)
+ .map(update -> ((MetadataUpdate.UpgradeFormatVersion)
update).formatVersion())
+ .findFirst();
+
+ Schema schema =
+ request.updates().stream()
+ .filter(update -> update instanceof MetadataUpdate.AddSchema)
+ .map(update -> ((MetadataUpdate.AddSchema) update).schema())
+ .findFirst()
+ .orElse(EMPTY_SCHEMA);
+
+ Catalog.TableBuilder tableBuilder = getCatalog().buildTable(ident,
schema);
+
+ TableMetadata.Builder changedMetadata =
+
formatVersion.map(TableMetadata::buildFromEmpty).orElse(TableMetadata.buildFromEmpty());
+ request.updates().forEach(update -> update.applyTo(changedMetadata));
+
+ TableMetadata changedTableMeta = changedMetadata.build();
+ tableBuilder.withPartitionSpec(changedTableMeta.spec());
+ tableBuilder.withSortOrder(changedTableMeta.sortOrder());
+ tableBuilder.withLocation(changedTableMeta.location());
+ tableBuilder.withProperties(changedTableMeta.properties());
+
+ Transaction transaction = tableBuilder.createOrReplaceTransaction();
+ if (transaction instanceof BaseTransaction) {
+ BaseTransaction baseTransaction = (BaseTransaction) transaction;
+
+ return LoadTableResponse.builder()
+ .withTableMetadata(create(baseTransaction, request))
+ .build();
+ } else {
+ throw new IllegalStateException(
+ "Cannot wrap catalog that does not produce BaseTransaction");
+ }
+
+ } else {
+ return CatalogHandlers.updateTable(getCatalog(), ident, request);
+ }
+ }
+
private LoadTableResponse loadTableInternal(TableIdentifier ident) {
Table table = getCatalog().loadTable(ident);
@@ -756,6 +809,101 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
}
+ private static boolean isCreate(UpdateTableRequest request) {
+ boolean isCreate =
+ request.requirements().stream()
+
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance);
+
+ if (isCreate) {
+ List<UpdateRequirement> invalidRequirements =
+ request.requirements().stream()
+ .filter(req -> !(req instanceof
UpdateRequirement.AssertTableDoesNotExist))
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ invalidRequirements.isEmpty(), "Invalid create requirements: %s",
invalidRequirements);
+ }
+
+ return isCreate;
+ }
+
+ private static TableMetadata create(BaseTransaction baseTransaction,
UpdateTableRequest request) {
+ // the only valid requirement is that the table will be created
+ TableOperations ops = baseTransaction.underlyingOps();
+ request.requirements().forEach(requirement ->
requirement.validate(ops.current()));
+
+ TableMetadata.Builder builder =
TableMetadata.buildFrom(baseTransaction.currentMetadata());
+ request
+ .updates()
+ .forEach(
+ update -> {
+ if (shouldApplyMetadataUpdateAfterBuilder(update)) {
+ update.applyTo(builder);
+ }
+ });
+
+ // create transactions do not retry. if the table exists, retrying is not
a solution
+ ops.commit(null, builder.build());
+
+ return ops.current();
+ }
+
+ /**
+ * Returns {@code false} for updates already reflected through {@link
Catalog.TableBuilder} during
+ * staged create; those must not be applied again on {@link
TableMetadata.Builder}.
+ */
+ @VisibleForTesting
+ static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) {
+ if (update instanceof MetadataUpdate.UpgradeFormatVersion) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddSchema) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetCurrentSchema) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemoveSchemas) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetLocation) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetProperties) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemoveProperties) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddSortOrder) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetDefaultSortOrder) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddPartitionSpec) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemovePartitionSpecs) {
+ return false;
+ }
+
+ return true;
+ }
+
private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) {
return fileIO instanceof InMemoryFileIO ? Maps.newHashMap() :
fileIO.properties();
}
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 1652548096..88acbec5f7 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
@@ -21,25 +21,41 @@ package org.apache.gravitino.iceberg.service;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyMap;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableMap;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateRequirement;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -99,7 +115,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testCatalogConfigToClientsForRestBackendUsesMergedRemoteProperties() {
+ void testRestCatalogClientConfigMergesRemote() {
IcebergConfig config =
new IcebergConfig(
ImmutableMap.of(
@@ -135,7 +151,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testCatalogConfigToClientsForNonRestBackend() {
+ void testNonRestCatalogClientConfig() {
Catalog catalog = mock(Catalog.class);
IcebergConfig config =
new IcebergConfig(
@@ -155,7 +171,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testCatalogConfigToClientsRejectsInvalidDataAccessValue() {
+ void testCatalogClientConfigRejectsBadDataAccess() {
Catalog catalog = mock(Catalog.class);
IcebergConfig config =
new IcebergConfig(
@@ -171,7 +187,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testConstructorDoesNotLoadCatalogEagerly() {
+ void testWrapperLazyLoadsCatalog() {
IcebergConfig config =
new IcebergConfig(
ImmutableMap.of(
@@ -189,7 +205,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testStageTableCreateWithLocationIncludesFileIoProperties() throws
Exception {
+ void testStageCreateWithLocationIncludesFileIo() throws Exception {
RESTCatalog catalog = mock(RESTCatalog.class);
Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class);
Transaction transaction = mock(Transaction.class);
@@ -240,7 +256,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testStageTableCreateWithNullLocationDoesNotCallWithLocation() {
+ void testStageCreateNullLocationSkipsWithLocation() {
RESTCatalog catalog = mock(RESTCatalog.class);
Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class);
Transaction transaction = mock(Transaction.class);
@@ -284,6 +300,200 @@ public class TestCatalogWrapperForREST {
verify(tableBuilder, never()).withLocation(any());
}
+ @Test
+ void testStagedCreateRejectsExtraRequirements() {
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test",
config, catalog);
+
+ UpdateRequirement extraRequirement = mock(UpdateRequirement.class);
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ List.of(new UpdateRequirement.AssertTableDoesNotExist(),
extraRequirement),
+ Collections.emptyList());
+
+ Assertions.assertThrowsExactly(
+ IllegalArgumentException.class,
+ () -> wrapper.updateTable(TableIdentifier.of("db", "tbl"), request));
+ }
+
+ @Test
+ void testPostBuilderMetadataSkipsHandledKinds() {
+ Schema schema = new Schema(Types.NestedField.required(1, "c",
Types.LongType.get()));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.AddSchema(schema)));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.UpgradeFormatVersion(2)));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.SetCurrentSchema(-1)));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.SetLocation("file:///tmp/loc")));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.SetProperties(ImmutableMap.of("k", "v"))));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.RemoveProperties(Collections.singleton("k"))));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new
MetadataUpdate.AddPartitionSpec(PartitionSpec.unpartitioned())));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new
MetadataUpdate.SetDefaultPartitionSpec(PartitionSpec.unpartitioned().specId())));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.AddSortOrder(SortOrder.unsorted())));
+ Assertions.assertFalse(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new
MetadataUpdate.SetDefaultSortOrder(SortOrder.unsorted().orderId())));
+ }
+
+ @Test
+ void testPostBuilderMetadataAllowsAssignUuid() {
+ Assertions.assertTrue(
+ CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ new MetadataUpdate.AssignUUID(UUID.randomUUID().toString())));
+ }
+
+ @Test
+ void testStagedCreateBuilderUsesDerivedMetadataV3() {
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class);
+ BaseTransaction baseTransaction = mock(BaseTransaction.class);
+ TableOperations ops = mock(TableOperations.class);
+ when(catalog.buildTable(any(TableIdentifier.class), any(Schema.class)))
+ .thenReturn(tableBuilder);
+ when(tableBuilder.withPartitionSpec(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withSortOrder(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withLocation(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withProperties(any())).thenReturn(tableBuilder);
+
when(tableBuilder.createOrReplaceTransaction()).thenReturn(baseTransaction);
+ when(baseTransaction.underlyingOps()).thenReturn(ops);
+
when(baseTransaction.currentMetadata()).thenReturn(minimalTableMetadataForStagedCreateTest());
+
+ AtomicReference<TableMetadata> opsCurrent = new AtomicReference<>();
+ when(ops.current()).thenAnswer(invocation -> opsCurrent.get());
+ doAnswer(
+ invocation -> {
+ opsCurrent.set(invocation.getArgument(1));
+ return null;
+ })
+ .when(ops)
+ .commit(any(), any());
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test",
config, catalog);
+
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ List.of(new UpdateRequirement.AssertTableDoesNotExist()),
+ stagedCreateMetadataUpdates(schema, Optional.of(3)));
+
+ Assertions.assertDoesNotThrow(
+ () -> wrapper.updateTable(TableIdentifier.of("db", "tbl"), request));
+
+ verify(tableBuilder).withPartitionSpec(any());
+ verify(tableBuilder).withSortOrder(any());
+ verify(tableBuilder).withProperties(any());
+ verify(tableBuilder, never()).withProperty(anyString(), anyString());
+ verify(tableBuilder).createOrReplaceTransaction();
+ }
+
+ @Test
+ void testStagedCreateOmitsFormatWithProperty() {
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class);
+ BaseTransaction baseTransaction = mock(BaseTransaction.class);
+ TableOperations ops = mock(TableOperations.class);
+ when(catalog.buildTable(any(TableIdentifier.class), any(Schema.class)))
+ .thenReturn(tableBuilder);
+ when(tableBuilder.withPartitionSpec(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withSortOrder(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withLocation(any())).thenReturn(tableBuilder);
+ when(tableBuilder.withProperties(any())).thenReturn(tableBuilder);
+
when(tableBuilder.createOrReplaceTransaction()).thenReturn(baseTransaction);
+ when(baseTransaction.underlyingOps()).thenReturn(ops);
+
when(baseTransaction.currentMetadata()).thenReturn(minimalTableMetadataForStagedCreateTest());
+
+ AtomicReference<TableMetadata> opsCurrent = new AtomicReference<>();
+ when(ops.current()).thenAnswer(invocation -> opsCurrent.get());
+ doAnswer(
+ invocation -> {
+ opsCurrent.set(invocation.getArgument(1));
+ return null;
+ })
+ .when(ops)
+ .commit(any(), any());
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test",
config, catalog);
+
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ List.of(new UpdateRequirement.AssertTableDoesNotExist()),
+ stagedCreateMetadataUpdates(schema, Optional.empty()));
+
+ Assertions.assertDoesNotThrow(
+ () -> wrapper.updateTable(TableIdentifier.of("db", "tbl"), request));
+
+ verify(tableBuilder, never()).withProperty(anyString(), anyString());
+ }
+
+ /**
+ * Minimal valid Iceberg staged-create update sequence so {@link
TableMetadata.Builder#build()}
+ * succeeds in {@link CatalogWrapperForREST#tableUpdateInternal}.
+ */
+ private static List<MetadataUpdate> stagedCreateMetadataUpdates(
+ Schema schema, Optional<Integer> formatVersion) {
+ List<MetadataUpdate> updates = new ArrayList<>();
+ updates.add(new MetadataUpdate.AssignUUID(UUID.randomUUID().toString()));
+ formatVersion.ifPresent(v -> updates.add(new
MetadataUpdate.UpgradeFormatVersion(v)));
+ updates.add(new MetadataUpdate.AddSchema(schema));
+ updates.add(new MetadataUpdate.SetCurrentSchema(-1));
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ updates.add(new MetadataUpdate.AddPartitionSpec(spec));
+ updates.add(new MetadataUpdate.SetDefaultPartitionSpec(spec.specId()));
+ SortOrder sortOrder = SortOrder.unsorted();
+ updates.add(new MetadataUpdate.AddSortOrder(sortOrder));
+ updates.add(new MetadataUpdate.SetDefaultSortOrder(sortOrder.orderId()));
+ updates.add(new MetadataUpdate.SetLocation("file:///tmp/t"));
+ return updates;
+ }
+
+ private static TableMetadata minimalTableMetadataForStagedCreateTest() {
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ return TableMetadata.newTableMetadata(
+ schema,
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "file:///tmp/t",
+ Collections.emptyMap());
+ }
+
private static class LazyCheckCatalogWrapperForREST extends
CatalogWrapperForREST {
LazyCheckCatalogWrapperForREST(String catalogName, IcebergConfig config) {
