(iceberg) 02/02: Flink: Forward-port REST catalog support in CatalogTestBase to 2.2

[pv...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git

commit 7858e8003eddda8af74e578f042b993fd32de6fc
Author: Talat Uyarer <ta...@apache.org>
AuthorDate: Fri Aug 28 12:09:42 2026 -0700

    Flink: Forward-port REST catalog support in CatalogTestBase to 2.2
---
 flink/v2.2/build.gradle                            |  1 +
 .../org/apache/iceberg/flink/CatalogTestBase.java  | 48 +++++++++++++++--
 .../flink/TestFlinkRESTCatalogDatabase.java        | 60 ++++++++++++++++++++++
 3 files changed, 104 insertions(+), 5 deletions(-)

diff --git a/flink/v2.2/build.gradle b/flink/v2.2/build.gradle
index b57d9383cf..ffd2f5dc2f 100644
--- a/flink/v2.2/build.gradle
+++ b/flink/v2.2/build.gradle
@@ -87,6 +87,7 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
     testImplementation project(path: ':iceberg-data', configuration: 'testArtifacts')
     testImplementation project(path: ':iceberg-orc', configuration: 'testArtifacts')
     testImplementation(testFixtures(project(':iceberg-parquet')))
+    testImplementation (project(path: ':iceberg-open-api', configuration: 'testFixturesRuntimeElements'))
 
     // By default, hive-exec is a fat/uber jar and it exports a guava library
     // that's really old. We use the core classifier to be able to override our guava
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
index bd34e62817..36abcf6a4c 100644
--- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
@@ -33,11 +33,17 @@ import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.RESTCatalogServer;
+import org.apache.iceberg.rest.RESTServerExtension;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 
 @ExtendWith(ParameterizedTestExtension.class)
@@ -46,7 +52,8 @@ public abstract class CatalogTestBase extends TestBase {
   /** Catalog implementations that SQL test suites can run against. */
   public enum CatalogType {
     HIVE("testhive"),
-    HADOOP("testhadoop");
+    HADOOP("testhadoop"),
+    REST("testrest");
 
     private final String catalogNamePrefix;
 
@@ -65,6 +72,21 @@ public abstract class CatalogTestBase extends TestBase {
   @TempDir protected File hiveWarehouse;
   @TempDir protected File hadoopWarehouse;
 
+  @RegisterExtension
+  private static final RESTServerExtension REST_SERVER_EXTENSION =
+      new RESTServerExtension(
+          ImmutableMap.of(
+              RESTCatalogServer.REST_PORT,
+              RESTServerExtension.FREE_PORT,
+              // In-memory sqlite database by default is private to the connection that created
+              // it. If more than 1 jdbc connection backs the catalog, the connections can see
+              // different database states, so limit the backend JdbcCatalog to a single
+              // connection.
+              CatalogProperties.CLIENT_POOL_SIZE,
+              "1"));
+
+  protected static RESTCatalog restCatalog;
+
   @Parameter(index = 0)
   protected CatalogType catalogType;
 
@@ -79,6 +101,7 @@ public abstract class CatalogTestBase extends TestBase {
   protected String flinkDatabase;
   protected Namespace icebergNamespace;
   protected boolean isHadoopCatalog;
+  protected boolean isRestCatalog;
 
   @Parameters(name = "catalogType={0}, baseNamespace={1}")
   protected static List<Object[]> parameters() {
@@ -88,14 +111,23 @@ public abstract class CatalogTestBase extends TestBase {
         new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1")});
   }
 
+  @BeforeAll
+  public static void initRestCatalog() {
+    restCatalog = REST_SERVER_EXTENSION.client();
+  }
+
   @BeforeEach
   public void before() {
     this.catalogName = catalogType.catalogName(baseNamespace);
     this.isHadoopCatalog = catalogType == CatalogType.HADOOP;
-    this.validationCatalog =
-        isHadoopCatalog
-            ? new HadoopCatalog(hiveConf, "file:" + hadoopWarehouse.getPath())
-            : catalog;
+    this.isRestCatalog = catalogType == CatalogType.REST;
+    if (isHadoopCatalog) {
+      this.validationCatalog = new HadoopCatalog(hiveConf, "file:" + hadoopWarehouse.getPath());
+    } else if (isRestCatalog) {
+      this.validationCatalog = restCatalog;
+    } else {
+      this.validationCatalog = catalog;
+    }
     this.validationNamespaceCatalog = (SupportsNamespaces) validationCatalog;
 
     config.put("type", "iceberg");
@@ -104,6 +136,12 @@ public abstract class CatalogTestBase extends TestBase {
     }
     if (isHadoopCatalog) {
       config.put(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE, "hadoop");
+    } else if (isRestCatalog) {
+      config.put(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE, "rest");
+      config.put(CatalogProperties.URI, restCatalog.properties().get(CatalogProperties.URI));
+      // disable Flink-side catalog caching so validations that write directly through
+      // restCatalog observe fresh metadata
+      config.put(CatalogProperties.CACHE_ENABLED, "false");
     } else {
       config.put(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE, "hive");
       config.put(CatalogProperties.URI, getURI(hiveConf));
diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkRESTCatalogDatabase.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkRESTCatalogDatabase.java
new file mode 100644
index 0000000000..988e179fdf
--- /dev/null
+++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkRESTCatalogDatabase.java
@@ -0,0 +1,60 @@
+/*
+ * 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.iceberg.flink;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.catalog.Namespace;
+import org.junit.jupiter.api.TestTemplate;
+
+/**
+ * Runs the {@link TestFlinkCatalogDatabase} suite against a REST catalog server backed by an
+ * in-memory JDBC catalog.
+ */
+public class TestFlinkRESTCatalogDatabase extends TestFlinkCatalogDatabase {
+
+  @Parameters(name = "catalogType={0}, baseNamespace={1}")
+  protected static List<Object[]> parameters() {
+    return Collections.singletonList(new Object[] {CatalogType.REST, Namespace.empty()});
+  }
+
+  @TestTemplate
+  @Override
+  public void testCreateNamespaceWithLocation() throws Exception {
+    // The JDBC-backed REST catalog stores the 'location' property verbatim instead of
+    // resolving it to a "file:" URI the way the Hive catalog does
+    assertThat(validationNamespaceCatalog.namespaceExists(icebergNamespace))
+        .as("Namespace should not already exist")
+        .isFalse();
+
+    Path location = temporaryDirectory;
+    sql("CREATE DATABASE %s WITH ('location'='%s')", flinkDatabase, location);
+    assertThat(validationNamespaceCatalog.namespaceExists(icebergNamespace))
+        .as("Namespace should exist")
+        .isTrue();
+    Map<String, String> nsMetadata =
+        validationNamespaceCatalog.loadNamespaceMetadata(icebergNamespace);
+    assertThat(nsMetadata).containsEntry("location", location.toString());
+  }
+}