From 6b132e9541fcd393371d3672db2bd6d14e81a2f8 Mon Sep 17 00:00:00 2001 From: Dimitar Goshev Date: Tue, 26 May 2026 16:18:22 +0200 Subject: [PATCH] Handle missing partition creation during versioned writes and proactively create partitions earlier. A failure to create partitions will not result in an operation failure during early attempts. Signed-off-by: Dimitar Goshev --- .../com/here/xyz/psql/DatabaseHandler.java | 65 +++++++++++++++++-- .../com/here/xyz/psql/DatabaseWriter.java | 64 ++++++++++++++++-- .../src/main/resources/sql/ext.sql | 33 +++++++++- .../src/main/resources/sql/DatabaseWriter.js | 25 +++++-- 4 files changed, 171 insertions(+), 16 deletions(-) diff --git a/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java b/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java index 3e686293d9..7d58a42e0c 100644 --- a/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java +++ b/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}") diff --git a/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseWriter.java b/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseWriter.java index c49dbf481b..19485e4343 100644 --- a/xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseWriter.java +++ b/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 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 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 matcher) { + return findSqlException(e, matcher) != null; + } + + private static SQLException findSqlException(Throwable e, Predicate 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 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; + } } diff --git a/xyz-psql-connector/src/main/resources/sql/ext.sql b/xyz-psql-connector/src/main/resources/sql/ext.sql index ecc5fefcbf..94fe5ba138 100644 --- a/xyz-psql-connector/src/main/resources/sql/ext.sql +++ b/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; diff --git a/xyz-util/src/main/resources/sql/DatabaseWriter.js b/xyz-util/src/main/resources/sql/DatabaseWriter.js index d3da4eb079..77177f6c74 100644 --- a/xyz-util/src/main/resources/sql/DatabaseWriter.js +++ b/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) {