Fix Connection Config DPOs

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationParametersDpo.java

@@ -31,7 +31,10 @@
  * The internal persistence-object counterpart to AuthenticationParameters defined in the API model.
  * Important: JsonSubTypes must be kept in sync with {@link AuthenticationType}.
  */
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "authenticationTypeCode", visible = true)
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME,
+    include = JsonTypeInfo.As.EXISTING_PROPERTY,
+    property = "authenticationTypeCode")
 @JsonSubTypes({
   @JsonSubTypes.Type(value = OAuthClientCredentialsParametersDpo.class, name = "1"),
   @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
@@ -65,7 +68,6 @@ public static AuthenticationParametersDpo fromAuthenticationParametersModelWithS
             (OAuthClientCredentialsParameters) authenticationParameters;
         config =
             new OAuthClientCredentialsParametersDpo(
-                AuthenticationType.OAUTH.getCode(),
                 oauthClientCredentialsModel.getTokenUri(),
                 oauthClientCredentialsModel.getClientId(),
                 secretReferences.get(INLINE_CLIENT_SECRET_REFERENCE_KEY),
@@ -76,7 +78,6 @@ public static AuthenticationParametersDpo fromAuthenticationParametersModelWithS
             (BearerAuthenticationParameters) authenticationParameters;
         config =
             new BearerAuthenticationParametersDpo(
-                AuthenticationType.BEARER.getCode(),
                 secretReferences.get(INLINE_BEARER_TOKEN_REFERENCE_KEY));
         break;
       default:

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationType.java

@@ -18,6 +18,9 @@
  */
 package org.apache.polaris.core.connection;
 
+import jakarta.annotation.Nonnull;
+import java.util.Arrays;
+
 /**
  * The internal persistence-object counterpart to AuthenticationParameters.AuthenticationTypeEnum
  * defined in the API model. We define integer type codes in this enum for better compatibility
@@ -27,15 +30,53 @@
  * AuthenticationParametersDpo}.
  */
 public enum AuthenticationType {
+  NULL_TYPE(0),
   OAUTH(1),
-  BEARER(2);
+  BEARER(2),
+  ;
+
+  private static final AuthenticationType[] REVERSE_MAPPING_ARRAY;
+
+  static {
+    // find max array size
+    int maxCode =
+        Arrays.stream(AuthenticationType.values())
+            .mapToInt(AuthenticationType::getCode)
+            .max()
+            .orElse(0);
+
+    // allocate mapping array
+    REVERSE_MAPPING_ARRAY = new AuthenticationType[maxCode + 1];
+
+    // populate mapping array
+    for (AuthenticationType authType : AuthenticationType.values()) {
+      REVERSE_MAPPING_ARRAY[authType.code] = authType;
+    }
+  }
 
   private final int code;
 
   AuthenticationType(int code) {
     this.code = code;
   }
 
+  /**
+   * Given the code associated to the type, return the associated AuthenticationType. Return
+   * NULL_TYPE if not found
+   *
+   * @param authTypeCode code associated to the authentication type
+   * @return ConnectionType corresponding to that code or null if mapping not found
+   */
+  public static @Nonnull AuthenticationType fromCode(int authTypeCode) {
+    // ensure it is within bounds
+    if (authTypeCode < 0 || authTypeCode >= REVERSE_MAPPING_ARRAY.length) {
+      return NULL_TYPE;
+    }
+
+    // get value
+    return REVERSE_MAPPING_ARRAY[authTypeCode];
+  }
+
   public int getCode() {
     return this.code;
   }

polaris-core/src/main/java/org/apache/polaris/core/connection/BearerAuthenticationParametersDpo.java

@@ -38,10 +38,9 @@ public class BearerAuthenticationParametersDpo extends AuthenticationParametersD
   private final UserSecretReference bearerTokenReference;
 
   public BearerAuthenticationParametersDpo(
-      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode,
       @JsonProperty(value = "bearerTokenReference", required = true) @Nonnull
           UserSecretReference bearerTokenReference) {
-    super(authenticationTypeCode);
+    super(AuthenticationType.BEARER.getCode());
     this.bearerTokenReference = bearerTokenReference;
   }
 

polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java

@@ -42,7 +42,10 @@
  * The internal persistence-object counterpart to ConnectionConfigInfo defined in the API model.
  * Important: JsonSubTypes must be kept in sync with {@link ConnectionType}.
  */
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "connectionTypeCode", visible = true)
+@JsonTypeInfo(
+    use = JsonTypeInfo.Id.NAME,
+    include = JsonTypeInfo.As.EXISTING_PROPERTY,
+    property = "connectionTypeCode")
 @JsonSubTypes({
   @JsonSubTypes.Type(value = IcebergRestConnectionConfigInfoDpo.class, name = "1"),
 })
@@ -146,7 +149,6 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
                 icebergRestConfigModel.getAuthenticationParameters(), secretReferences);
         config =
             new IcebergRestConnectionConfigInfoDpo(
-                ConnectionType.ICEBERG_REST.getCode(),
                 icebergRestConfigModel.getUri(),
                 authenticationParameters,
                 icebergRestConfigModel.getRemoteCatalogName());

polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionType.java

@@ -18,7 +18,8 @@
  */
 package org.apache.polaris.core.connection;
 
-import jakarta.annotation.Nullable;
+import jakarta.annotation.Nonnull;
+import java.util.Arrays;
 
 /**
  * The internal persistence-object counterpart to ConnectionConfigInfo.ConnectionTypeEnum defined in
@@ -29,21 +30,22 @@
  * ConnectionConfigInfoDpo}.
  */
 public enum ConnectionType {
-  ICEBERG_REST(1);
+  NULL_TYPE(0),
+  ICEBERG_REST(1),
+  ;
 
   private static final ConnectionType[] REVERSE_MAPPING_ARRAY;
 
   static {
     // find max array size
-    int maxId = 0;
-    for (ConnectionType connectionType : ConnectionType.values()) {
-      if (maxId < connectionType.code) {
-        maxId = connectionType.code;
-      }
-    }
+    int maxCode =
+        Arrays.stream(AuthenticationType.values())
+            .mapToInt(AuthenticationType::getCode)
+            .max()
+            .orElse(0);
 
     // allocate mapping array
-    REVERSE_MAPPING_ARRAY = new ConnectionType[maxId + 1];
+    REVERSE_MAPPING_ARRAY = new ConnectionType[maxCode + 1];
 
     // populate mapping array
     for (ConnectionType connectionType : ConnectionType.values()) {
@@ -61,13 +63,13 @@ public enum ConnectionType {
    * Given the code associated to the type, return the associated ConnectionType. Return null if not
    * found
    *
-   * @param connectionTypeCode code associated to the entity type
+   * @param connectionTypeCode code associated to the connection type
    * @return ConnectionType corresponding to that code or null if mapping not found
    */
-  public static @Nullable ConnectionType fromCode(int connectionTypeCode) {
+  public static @Nonnull ConnectionType fromCode(int connectionTypeCode) {
     // ensure it is within bounds
-    if (connectionTypeCode >= REVERSE_MAPPING_ARRAY.length) {
-      return null;
+    if (connectionTypeCode < 0 || connectionTypeCode >= REVERSE_MAPPING_ARRAY.length) {
+      return ConnectionType.NULL_TYPE;
     }
 
     // get value

polaris-core/src/main/java/org/apache/polaris/core/connection/IcebergRestConnectionConfigInfoDpo.java

@@ -39,13 +39,12 @@ public class IcebergRestConnectionConfigInfoDpo extends ConnectionConfigInfoDpo
   private final String remoteCatalogName;
 
   public IcebergRestConnectionConfigInfoDpo(
-      @JsonProperty(value = "connectionTypeCode", required = true) int connectionTypeCode,
       @JsonProperty(value = "uri", required = true) @Nonnull String uri,
       @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
           AuthenticationParametersDpo authenticationParameters,
       @JsonProperty(value = "remoteCatalogName", required = false) @Nullable
           String remoteCatalogName) {
-    super(connectionTypeCode, uri, authenticationParameters);
+    super(ConnectionType.ICEBERG_REST.getCode(), uri, authenticationParameters);
     this.remoteCatalogName = remoteCatalogName;
   }
 

polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java

@@ -59,13 +59,12 @@ public class OAuthClientCredentialsParametersDpo extends AuthenticationParameter
   private final List<String> scopes;
 
   public OAuthClientCredentialsParametersDpo(
-      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode,
       @JsonProperty(value = "tokenUri", required = false) @Nullable String tokenUri,
       @JsonProperty(value = "clientId", required = true) @Nonnull String clientId,
       @JsonProperty(value = "clientSecretReference", required = true) @Nonnull
           UserSecretReference clientSecretReference,
       @JsonProperty(value = "scopes", required = false) @Nullable List<String> scopes) {
-    super(authenticationTypeCode);
+    super(AuthenticationType.OAUTH.getCode());
 
     this.tokenUri = tokenUri;
     this.clientId = clientId;

polaris-core/src/test/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpoTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.core.connection;
 
+import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -28,8 +29,12 @@
 import org.junit.jupiter.api.Test;
 
 public class ConnectionConfigInfoDpoTest {
-  PolarisDiagnostics polarisDiagnostics = new PolarisDefaultDiagServiceImpl();
-  ObjectMapper objectMapper = new ObjectMapper();
+  private static final PolarisDiagnostics polarisDiagnostics = new PolarisDefaultDiagServiceImpl();
+  private static final ObjectMapper objectMapper = new ObjectMapper();
+
+  static {
+    objectMapper.enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION);
+  }
 
   @Test
   void testOAuthClientCredentialsParameters() throws JsonProcessingException {