Nessie: Infer default API version from URI (apache#9459)

nessie/src/main/java/org/apache/iceberg/nessie/NessieCatalog.java

@@ -25,6 +25,8 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.TableOperations;
@@ -103,9 +105,12 @@ public void initialize(String name, Map<String, String> options) {
             .fallbackTo(x -> options.get(removePrefix.apply(x)));
     NessieClientBuilder nessieClientBuilder =
         NessieClientBuilder.createClientBuilderFromSystemSettings(configSource);
-    // default version is set to v1.
-    final String apiVersion =
-        options.getOrDefault(removePrefix.apply(NessieUtil.CLIENT_API_VERSION), "1");
+    // default version is inferred by uri.
+    String apiVersion = options.get(removePrefix.apply(NessieUtil.CLIENT_API_VERSION));
+    if (apiVersion == null) {
+      apiVersion = inferVersionFromURI(options.get(CatalogProperties.URI));
+    }
+
     NessieApiV1 api;
     switch (apiVersion) {
       case "1":
@@ -128,6 +133,25 @@ public void initialize(String name, Map<String, String> options) {
         catalogOptions);
   }
 
+  private static String inferVersionFromURI(String uri) {
+    if (uri == null) {
+      throw new IllegalArgumentException("URI is not specified in the catalog properties");
+    }
+
+    // match for uri ending with /v1, /v2 etc
+    Pattern pattern = Pattern.compile("/v(\\d+)$");
+    Matcher matcher = pattern.matcher(uri);
+    if (matcher.find()) {
+      return matcher.group(1);
+    } else {
+      throw new IllegalArgumentException(
+          String.format(
+              "URI doesn't end with the version: %s. "
+                  + "Please configure `client-api-version` in the catalog properties explicitly.",
+              uri));
+    }
+  }
+
   /**
    * An alternative way to initialize the catalog using a pre-configured {@link NessieIcebergClient}
    * and {@link FileIO} instance.

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java

@@ -35,7 +35,6 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 import org.projectnessie.client.api.NessieApiV1;
-import org.projectnessie.client.ext.NessieApiVersion;
 import org.projectnessie.client.ext.NessieApiVersions;
 import org.projectnessie.client.ext.NessieClientFactory;
 import org.projectnessie.client.ext.NessieClientUri;
@@ -65,7 +64,6 @@ public class TestNessieCatalog extends CatalogTests<NessieCatalog> {
 
   private NessieCatalog catalog;
   private NessieApiV1 api;
-  private NessieApiVersion apiVersion;
   private Configuration hadoopConfig;
   private String initialHashOfDefaultBranch;
   private String uri;
@@ -74,7 +72,6 @@ public class TestNessieCatalog extends CatalogTests<NessieCatalog> {
   public void setUp(NessieClientFactory clientFactory, @NessieClientUri URI nessieUri)
       throws NessieNotFoundException {
     api = clientFactory.make();
-    apiVersion = clientFactory.apiVersion();
     initialHashOfDefaultBranch = api.getDefaultBranch().getHash();
     uri = nessieUri.toASCIIString();
     hadoopConfig = new Configuration();
@@ -122,9 +119,7 @@ private NessieCatalog initNessieCatalog(String ref) {
             CatalogProperties.URI,
             uri,
             CatalogProperties.WAREHOUSE_LOCATION,
-            temp.toUri().toString(),
-            "client-api-version",
-            apiVersion == NessieApiVersion.V2 ? "2" : "1");
+            temp.toUri().toString());
     newCatalog.initialize("nessie", options);
     return newCatalog;
   }

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieIcebergClient.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
@@ -530,6 +531,44 @@ public void testInvalidClientApiVersion() throws IOException {
     }
   }
 
+  @Test
+  public void testInvalidClientApiVersionViaURI() throws IOException {
+    try (NessieCatalog newCatalog = new NessieCatalog()) {
+      newCatalog.setConf(hadoopConfig);
+      ImmutableMap.Builder<String, String> options =
+          ImmutableMap.<String, String>builder().put("uri", "some/uri/");
+      Assertions.assertThatIllegalArgumentException()
+          .isThrownBy(() -> newCatalog.initialize("nessie", options.buildOrThrow()))
+          .withMessage(
+              "URI doesn't end with the version: some/uri/. Please configure `client-api-version` in the catalog properties explicitly.");
+
+      ImmutableMap.Builder<String, String> newOptions =
+          ImmutableMap.<String, String>builder().put("uri", "some/uri/v3");
+      Assertions.assertThatIllegalArgumentException()
+          .isThrownBy(() -> newCatalog.initialize("nessie", newOptions.buildOrThrow()))
+          .withMessage("Unsupported client-api-version: 3. Can only be 1 or 2");
+    }
+  }
+
+  @Test
+  public void testClientApiVersionOverride() {
+    // for v1 URI use v2 version and vice versa.
+    String version = apiVersion.equals("1") ? "2" : "1";
+
+    NessieCatalog newCatalog = new NessieCatalog();
+    newCatalog.setConf(hadoopConfig);
+    ImmutableMap.Builder<String, String> options =
+        ImmutableMap.<String, String>builder()
+            .put(CatalogProperties.URI, uri)
+            .put(CatalogProperties.WAREHOUSE_LOCATION, temp.toUri().toString())
+            .put("client-api-version", version);
+    newCatalog.initialize("nessie", options.buildOrThrow());
+    // Since client-api-version is configured, API version should not be based on URI.
+    Assertions.assertThatRuntimeException()
+        .isThrownBy(() -> newCatalog.loadTable(TableIdentifier.of("foo", "t1")))
+        .withMessageStartingWith("API version mismatch, check URI prefix");
+  }
+
   private void commit(String branch, String message, Operation... operations)
       throws NessieNotFoundException, NessieConflictException {
     Branch ref = (Branch) api.getReference().refName(branch).get();