Improved partition creation handling

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java

@@ -26,6 +26,9 @@
 import static com.here.xyz.psql.DatabaseWriter.ModificationType.UPDATE;
 import static com.here.xyz.psql.query.XyzEventBasedQueryRunner.readBranchTableFromEvent;
 import static com.here.xyz.responses.XyzError.NOT_IMPLEMENTED;
+import static com.here.xyz.util.db.pg.LockHelper.advisoryLock;
+import static com.here.xyz.util.db.pg.LockHelper.advisoryUnlock;
+import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.PARTITION_SIZE;
 
 import com.amazonaws.services.lambda.runtime.Context;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -54,11 +57,13 @@
 import com.here.xyz.util.db.datasource.DataSourceProvider;
 import com.here.xyz.util.db.datasource.DatabaseSettings;
 import com.here.xyz.util.db.datasource.DatabaseSettings.ScriptResourcePath;
+import com.here.xyz.util.db.pg.XyzSpaceTableHelper;
 import com.here.xyz.util.runtime.FunctionRuntime;
 import com.here.xyz.util.runtime.LambdaFunctionRuntime;
 import java.sql.BatchUpdateException;
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -88,10 +93,13 @@ public abstract class DatabaseHandler extends StorageConnector {
     private static String INCLUDE_OLD_STATES = "includeOldStates"; // read from event params
 
     private boolean retryAttempted;
+    private boolean missingPartitionRecoveryAttempted;
 
     void reset() {
         //Clear the data sources cache
         CachedPooledDataSources.invalidateCache();
+        retryAttempted = false;
+        missingPartitionRecoveryAttempted = false;
     }
 
     protected DataSourceProvider dataSourceProvider;
@@ -126,6 +134,7 @@ public void initialize(DatabaseSettings dbSettings, Context context) {
 
             dataSourceProvider = new CachedPooledDataSources(dbSettings);
             retryAttempted = false;
+            missingPartitionRecoveryAttempted = false;
             DataSourceProvider.setDefaultProvider(dataSourceProvider);
             this.dbSettings = dbSettings;
         }
@@ -322,6 +331,14 @@ protected FeatureCollection executeModifyFeatures(ModifyFeaturesEvent event) thr
                 /** Add objects which are responsible for the failed operation */
                 event.setFailed(fails);
 
+                boolean missingHistoryPartition = DatabaseWriter.isMissingHistoryPartitionException(e);
+                boolean historyPartitionLockNotAvailable = DatabaseWriter.isLockNotAvailableException(e);
+                if ((missingHistoryPartition || historyPartitionLockNotAvailable)
+                    && recoverHistoryPartition(event, version, connection, historyPartitionLockNotAvailable)) {
+                    logger.warn("{} triggered retry after history partition recovery.", traceItem, e);
+                    return executeModifyFeatures(event);
+                }
+
                 if (retryCausedOnServerlessDB(e) && !retryAttempted) {
                     if(!connection.isClosed())
                     { connection.setAutoCommit(previousAutoCommitState);
@@ -335,10 +352,9 @@ protected FeatureCollection executeModifyFeatures(ModifyFeaturesEvent event) thr
                 if (event.getTransaction()) {
                     connection.rollback();
 
-                    if (e instanceof SQLException sqlException && sqlException.getSQLState() != null
-                            && sqlException.getSQLState().equalsIgnoreCase("42P01"))
-                        ;//Table does not exist yet - create it!
-                    else {
+                    boolean tableDoesNotExist = e instanceof SQLException sqlException && sqlException.getSQLState() != null
+                            && sqlException.getSQLState().equalsIgnoreCase("42P01");
+                    if (!tableDoesNotExist) {
 
                         logger.warn("{} Transaction has failed. ", traceItem, e);
                         connection.close();
@@ -423,6 +439,47 @@ protected FeatureCollection executeModifyFeatures(ModifyFeaturesEvent event) thr
         }
     }
 
+    private boolean recoverHistoryPartition(ModifyFeaturesEvent event, long version, Connection connection, boolean createNextPartition)
+        throws SQLException {
+        if (missingPartitionRecoveryAttempted)
+            return false;
+
+        int remainingSeconds = FunctionRuntime.getInstance().getRemainingTime() / 1000;
+        if (!isRemainingTimeSufficient(remainingSeconds))
+            throw new SQLException("No time left to create missing history partition.", "54000");
+
+        missingPartitionRecoveryAttempted = true;
+        retryAttempted = true;
+
+        boolean previousAutoCommitState = connection.getAutoCommit();
+        if (!previousAutoCommitState)
+            connection.rollback();
+
+        String table = readBranchTableFromEvent(event);
+        long partitionNo = Math.floorDiv(version, PARTITION_SIZE) + (createNextPartition ? 1 : 0);
+        SQLQuery createPartitionQuery = XyzSpaceTableHelper.buildCreateHistoryPartitionQuery(dbSettings.getSchema(), table, partitionNo, true);
+
+        logger.warn("{} History partition recovery triggered for {}.{} and version {}. Creating partition no {} before retry.",
+            traceItem, dbSettings.getSchema(), table, version, partitionNo);
+
+        try {
+            connection.setAutoCommit(true);
+            advisoryLock(table, connection);
+            try (Statement statement = connection.createStatement()) {
+                statement.setQueryTimeout(calculateTimeout());
+                statement.execute(createPartitionQuery.toExecutableQueryString());
+            }
+            finally {
+                advisoryUnlock(table, connection);
+            }
+        }
+        finally {
+            connection.setAutoCommit(previousAutoCommitState);
+        }
+
+        return true;
+    }
+
     private boolean checkUniqueTableConstraint(ModifyFeaturesEvent event) throws SQLException {
         return new SQLQuery("SELECT 1 FROM pg_catalog.pg_constraint "
             + "WHERE connamespace::regnamespace::text = #{schema} AND conname = #{constraintName}")

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseWriter.java

@@ -48,6 +48,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map.Entry;
+import java.util.function.Predicate;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.locationtech.jts.geom.Coordinate;
@@ -137,6 +138,10 @@ public String toString() {
 
     protected static final String TRANSACTION_ERROR_GENERAL = "Transaction has failed";
 
+    private static final String UNDEFINED_TABLE_SQL_STATE = "42P01";
+    private static final String LOCK_NOT_AVAILABLE_SQL_STATE = "55P03";
+    private static final String CHECK_VIOLATION_SQL_STATE = "23514";
+
     private static PGobject featureToPGobject(ModifyFeaturesEvent event, final Feature feature, long version) throws SQLException {
         final Geometry geometry = feature.getGeometry();
         feature.setGeometry(null); //Do not serialize the geometry in the JSON object
@@ -280,8 +285,9 @@ protected static void modifyFeatures(DatabaseHandler dbh, ModifyFeaturesEvent ev
                     if (transactional)
                         throw e;
 
-                    if (e instanceof SQLException sqlException && "42P01".equalsIgnoreCase(sqlException.getSQLState()))
-                        throw (SQLException) e;
+                    SQLException retryableException = getRetryableTableMaintenanceException(e);
+                    if (retryableException != null)
+                        throw retryableException;
 
                     fails.add(new FeatureCollection.ModificationFailure().withId(getIdFromInput(action, inputDatum))
                         .withMessage(e instanceof WriteFeatureException ? e.getMessage() : getFailedRowErrorMsg(action, event)));
@@ -413,11 +419,7 @@ private static void executeBatchesAndCheckOnFailures(List<String> idList, Prepar
             }
         }
         catch (Exception e) {
-            if (e instanceof SQLException sqlException && sqlException.getSQLState() != null
-                && (sqlException.getSQLState().equalsIgnoreCase("42P01") ||
-                    //Lock not available - e.g.: happens if a long-running query blocks the
-                    //history partition creation
-                    sqlException.getSQLState().equalsIgnoreCase("55P03")))
+            if (getRetryableTableMaintenanceException(e) != null)
                 //Re-throw, as a missing table will be handled by DatabaseHandler.
                 throw e;
 
@@ -443,4 +445,52 @@ private static void fillFailList(int[] batchResult, List<FeatureCollection.Modif
             if (batchResult[i] == 0)
                 fails.add(new FeatureCollection.ModificationFailure().withId(idList.get(i)).withMessage(getFailedRowErrorMsg(action, event)));
     }
+
+    static boolean isMissingHistoryPartitionException(Throwable e) {
+        return anySqlExceptionMatches(e, sqlException -> CHECK_VIOLATION_SQL_STATE.equalsIgnoreCase(sqlException.getSQLState())
+            && sqlException.getMessage() != null
+            && sqlException.getMessage().toLowerCase().contains("no partition of relation"));
+    }
+
+    static boolean isLockNotAvailableException(Throwable e) {
+        return anySqlExceptionMatches(e, sqlException -> LOCK_NOT_AVAILABLE_SQL_STATE.equalsIgnoreCase(sqlException.getSQLState()));
+    }
+
+    private static SQLException getRetryableTableMaintenanceException(Throwable e) {
+        return findSqlException(e, sqlException ->
+            UNDEFINED_TABLE_SQL_STATE.equalsIgnoreCase(sqlException.getSQLState())
+                || isLockNotAvailableException(sqlException)
+                || isMissingHistoryPartitionException(sqlException));
+    }
+
+    private static boolean anySqlExceptionMatches(Throwable e, Predicate<SQLException> matcher) {
+        return findSqlException(e, matcher) != null;
+    }
+
+    private static SQLException findSqlException(Throwable e, Predicate<SQLException> matcher) {
+        while (e != null) {
+            if (e instanceof SQLException sqlException) {
+                SQLException matchingException = findMatchingSqlException(sqlException, matcher);
+                if (matchingException != null)
+                    return matchingException;
+            }
+            e = e.getCause();
+        }
+        return null;
+    }
+
+    private static SQLException findMatchingSqlException(SQLException e, Predicate<SQLException> matcher) {
+        while (e != null) {
+            if (matcher.test(e))
+                return e;
+            Throwable cause = e.getCause();
+            if (cause != null && cause != e) {
+                SQLException matchingException = findSqlException(cause, matcher);
+                if (matchingException != null)
+                    return matchingException;
+            }
+            e = e.getNextException();
+        }
+        return null;
+    }
 }

xyz-psql-connector/src/main/resources/sql/ext.sql

@@ -1,3 +1,22 @@
+/*
+ * Copyright (C) 2017-2026 HERE Europe B.V.
+ *
+ * Licensed 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
 /*
  * Copyright (C) 2017-2025 HERE Europe B.V.
  *
@@ -1941,7 +1960,19 @@ $BODY$
             format('INSERT INTO %I.%I (id, version, operation, author, jsondata, geo) VALUES (%L, %L, %L, %L, %L, %L)',
                    schema, tableName, id, version, operation, author, jsondata, xyz_geoFromWkb(geo) );
 
-        -- If the current history partition is nearly full, create the next one already
+        -- Create the next history partition proactively. Start once the current partition is half full,
+        -- then repeat at regular intervals (~100 attempts per second half for the default partition size).
+        IF version % partitionSize >= partitionSize / 2 AND version % greatest(partitionSize / 200, 1) = 0 THEN
+            BEGIN
+                EXECUTE xyz_create_history_partition(schema, tableName, (floor(version / partitionSize) + 1)::BIGINT, partitionSize);
+            EXCEPTION WHEN OTHERS THEN
+                RAISE WARNING 'Best-effort history partition creation failed for %.% at version %: %',
+                    schema, tableName, version, SQLERRM;
+            END;
+        END IF;
+
+        -- Keep the near-full safety net strict. At this point a failure must abort the write so the
+        -- application-level recovery path can create the partition and retry the whole modification.
         IF version % partitionSize > partitionSize - 50 THEN
             EXECUTE xyz_create_history_partition(schema, tableName, (floor(version / partitionSize) + 1)::BIGINT, partitionSize);
         END IF;

xyz-util/src/main/resources/sql/DatabaseWriter.js

@@ -372,15 +372,32 @@ class DatabaseWriter {
   }
 
   /**
-   * If the current history partition is nearly full, create the next one already
+   * Create the next history partition proactively once the current partition is half full,
+   * then repeat at regular intervals. Early attempts are best-effort, while the near-full safety net is strict.
    * @param version The version that is about to be written
    * @private
    */
   _createHistoryPartition(version) {
     const PARTITION_SIZE = this._PARTITION_SIZE();
-    if (version % PARTITION_SIZE > PARTITION_SIZE - 50)
-      plv8.execute(`SELECT xyz_create_history_partition($1, $2, $3::BIGINT, $4::BIGINT);`,
-          [this.schema, this.table, Math.floor(version / PARTITION_SIZE) + 1, PARTITION_SIZE]);
+    const partitionOffset = version % PARTITION_SIZE;
+    const preCreationInterval = Math.max(Math.floor(PARTITION_SIZE / 200), 1);
+    const nearFull = partitionOffset > PARTITION_SIZE - 50;
+    const bestEffortPreCreation = partitionOffset >= Math.floor(PARTITION_SIZE / 2) && version % preCreationInterval === 0;
+    const parameters = [this.schema, this.table, Math.floor(version / PARTITION_SIZE) + 1, PARTITION_SIZE];
+
+    if (nearFull) {
+      plv8.execute(`SELECT xyz_create_history_partition($1, $2, $3::BIGINT, $4::BIGINT);`, parameters);
+      return;
+    }
+
+    if (bestEffortPreCreation) {
+      try {
+        plv8.execute(`SELECT xyz_create_history_partition($1, $2, $3::BIGINT, $4::BIGINT);`, parameters);
+      }
+      catch (e) {
+        plv8.elog(WARNING, `FW_LOG [${queryContext().queryId}] Best-effort history partition creation failed for ${this.schema}.${this.table} at version ${version}: ${e.message || e}`);
+      }
+    }
   }
 
   _purgeOldChangesets(version) {