Add fallback in case the VERSION table is not present (#2653)

* initial commit

* wire up

* pastefix

* change to postgres specific code

Author: eric-maynard

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java

@@ -53,6 +53,7 @@ public class DatasourceOperations {
   private static final Logger LOGGER = LoggerFactory.getLogger(DatasourceOperations.class);
 
   private static final String CONSTRAINT_VIOLATION_SQL_CODE = "23505";
+  private static final String RELATION_DOES_NOT_EXIST = "42P01";
 
   // POSTGRES RETRYABLE EXCEPTIONS
   private static final String SERIALIZATION_FAILURE_SQL_CODE = "40001";
@@ -394,6 +395,10 @@ public boolean isConstraintViolation(SQLException e) {
     return CONSTRAINT_VIOLATION_SQL_CODE.equals(e.getSQLState());
   }
 
+  public boolean isRelationDoesNotExist(SQLException e) {
+    return RELATION_DOES_NOT_EXIST.equals(e.getSQLState());
+  }
+
   private Connection borrowConnection() throws SQLException {
     return datasource.getConnection();
   }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -725,7 +725,8 @@ public boolean hasChildren(
     }
   }
 
-  static int loadSchemaVersion(DatasourceOperations datasourceOperations) {
+  static int loadSchemaVersion(
+      DatasourceOperations datasourceOperations, boolean fallbackOnDoesNotExist) {
     PreparedQuery query = QueryGenerator.generateVersionQuery();
     try {
       List<SchemaVersion> schemaVersion =
@@ -736,6 +737,9 @@ static int loadSchemaVersion(DatasourceOperations datasourceOperations) {
       return schemaVersion.getFirst().getValue();
     } catch (SQLException e) {
       LOGGER.error("Failed to load schema version due to {}", e.getMessage(), e);
+      if (fallbackOnDoesNotExist && datasourceOperations.isRelationDoesNotExist(e)) {
+        return SchemaVersion.MINIMUM.getValue();
+      }
       throw new IllegalStateException("Failed to retrieve schema version", e);
     }
   }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -32,6 +32,8 @@
 import javax.sql.DataSource;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.config.BehaviorChangeConfiguration;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -74,6 +76,7 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject Instance<DataSource> dataSource;
   @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
+  @Inject PolarisConfigurationStore configurationStore;
 
   protected JdbcMetaStoreManagerFactory() {}
 
@@ -98,7 +101,11 @@ private void initializeForRealm(
     // RealmContext (request-scoped bean) can still create a JdbcBasePersistenceImpl
     String realmId = realmContext.getRealmIdentifier();
     // determine schemaVersion once per realm
-    final int schemaVersion = JdbcBasePersistenceImpl.loadSchemaVersion(datasourceOperations);
+    final int schemaVersion =
+        JdbcBasePersistenceImpl.loadSchemaVersion(
+            datasourceOperations,
+            configurationStore.getConfiguration(
+                realmContext, BehaviorChangeConfiguration.SCHEMA_VERSION_FALL_BACK_ON_DNE));
     sessionSupplierMap.put(
         realmId,
         () ->

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/models/SchemaVersion.java

@@ -25,6 +25,8 @@
 import org.apache.polaris.persistence.relational.jdbc.DatabaseType;
 
 public class SchemaVersion implements Converter<SchemaVersion> {
+  public static final SchemaVersion MINIMUM = new SchemaVersion(0);
+
   private final Integer value;
 
   public SchemaVersion() {

polaris-core/src/main/java/org/apache/polaris/core/config/BehaviorChangeConfiguration.java

@@ -84,4 +84,14 @@ protected BehaviorChangeConfiguration(
                   + " credential vending.")
           .defaultValue(false)
           .buildBehaviorChangeConfiguration();
+
+  public static final BehaviorChangeConfiguration<Boolean> SCHEMA_VERSION_FALL_BACK_ON_DNE =
+      PolarisConfiguration.<Boolean>builder()
+          .key("SCHEMA_VERSION_FALL_BACK_ON_DNE")
+          .description(
+              "If set to true, exceptions encountered while loading the VERSION table which appear to be"
+                  + " caused by the VERSION table not existing will be interpreted as meaning that the"
+                  + " schema version is currently 0.")
+          .defaultValue(true)
+          .buildBehaviorChangeConfiguration();
 }