DS-1644: Fix bug in FeatureWriter related to writing to composite layers with enabled conflictDetection

Additionally, improve the following: - Fix error message of retryable write conflicts in DatabaseWriter - Add new field "internalDetails" to ErrorResponseException to be used by connector implementations optionally - Enable logging internal details of connector exceptions (at least as WARNs) - Fix plv8 emulation: - enable the proper search_path so that DB functions are working - Fully resubstitute all SQL placeholders to not use prepared statements, as NPM module "pg" does not support using multiple statements within one prepared statement (e.g., as it's needed for #insertHistoryRow()) Signed-off-by: Benjamin Rögner <benjamin.roegner@here.com>

Author: roegi

xyz-connectors/src/main/java/com/here/xyz/connectors/EntryConnectorHandler.java

@@ -103,7 +103,6 @@ public void handleRequest(InputStream input, OutputStream output, Context contex
  String className = "com.here.xyz.psql.PSQLXyzConnector";
  if (event.getConnectorParams() != null && event.getConnectorParams().containsKey("className"))
  className = event.getConnectorParams().get("className").toString();
- //TBD: read from connector config
  final Class<?> mainClass = Class.forName(className);
  final AbstractConnectorHandler reqHandler = (AbstractConnectorHandler) mainClass.getDeclaredConstructor().newInstance();
 
@@ -118,8 +117,12 @@ public void handleRequest(InputStream input, OutputStream output, Context contex
  }
  catch (ErrorResponseException e) {
  switch (e.getErrorResponse().getError()) {
- case EXCEPTION, BAD_GATEWAY, TIMEOUT -> logger.error("{} Exception in Connector:", streamId, e);
+ case EXCEPTION, BAD_GATEWAY, TIMEOUT -> logger.error("[{}] Unexpected exception in connector:", streamId, e);
+ default -> logger.warn("[{}] Exception in connector:", streamId, e);
  }
+ if (e.getInternalDetails() != null)
+ logger.warn("[{}] Internal details of exception: {}", streamId, e.getInternalDetails());
+
  e.getErrorResponse().setStreamId(streamId);
  dataOut = e.getErrorResponse();
  }

xyz-connectors/src/main/java/com/here/xyz/connectors/ErrorResponseException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2017-2023 HERE Europe B.V.
+ * Copyright (C) 2017-2025 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.
@@ -31,6 +31,11 @@ public class ErrorResponseException extends Exception {
 
  private ErrorResponse errorResponse;
 
+ /**
+ * Additional details about the error, that will be logged but won't be added to the (user-facing) error message.
+ */
+ private String internalDetails;
+
  public ErrorResponseException(XyzError xyzError, String errorMessage) {
  super(errorMessage);
  createErrorResponse(xyzError, errorMessage);
@@ -78,4 +83,17 @@ private void createErrorResponse(XyzError xyzError, String errorMessage) {
  public ErrorResponse getErrorResponse() {
  return errorResponse;
  }
+
+ public String getInternalDetails() {
+ return internalDetails;
+ }
+
+ public void setInternalDetails(String internalDetails) {
+ this.internalDetails = internalDetails;
+ }
+
+ public ErrorResponseException withInternalDetails(String internalDetails) {
+ setInternalDetails(internalDetails);
+ return this;
+ }
 }

xyz-hub-test/src/test/java/com/here/xyz/hub/rest/TestSpaceWithFeature.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2017-2023 HERE Europe B.V.
+ * Copyright (C) 2017-2025 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.
@@ -19,19 +19,25 @@
 
 package com.here.xyz.hub.rest;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import static com.here.xyz.hub.auth.TestAuthenticator.AuthProfile.ACCESS_ALL;
-import com.here.xyz.models.geojson.coordinates.PointCoordinates;
-import com.here.xyz.models.geojson.implementation.Feature;
-import com.here.xyz.models.geojson.implementation.FeatureCollection;
-import com.here.xyz.models.geojson.implementation.Point;
-import com.here.xyz.models.geojson.implementation.Properties;
 import static com.here.xyz.util.service.BaseHttpServerVerticle.HeaderValues.APPLICATION_GEO_JSON;
 import static com.here.xyz.util.service.BaseHttpServerVerticle.HeaderValues.APPLICATION_JSON;
 import static com.here.xyz.util.service.BaseHttpServerVerticle.HeaderValues.APPLICATION_VND_HERE_FEATURE_MODIFICATION_LIST;
 import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
 import static io.restassured.RestAssured.given;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.isIn;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.here.xyz.models.geojson.coordinates.PointCoordinates;
+import com.here.xyz.models.geojson.implementation.Feature;
+import com.here.xyz.models.geojson.implementation.FeatureCollection;
+import com.here.xyz.models.geojson.implementation.Point;
+import com.here.xyz.models.geojson.implementation.Properties;
 import io.restassured.response.ValidatableResponse;
 import io.vertx.core.json.JsonObject;
 import java.util.ArrayList;
@@ -47,11 +53,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.commons.lang3.RandomStringUtils;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.everyItem;
-import static org.hamcrest.Matchers.isIn;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
 
 public class TestSpaceWithFeature extends TestWithSpaceCleanup {
  protected static String embeddedStorageId = "psql";
@@ -63,8 +64,8 @@ protected static void remove() {
  remove("x-psql-test");
  }
 
- protected static void remove(String spaceId) { 
- TestWithSpaceCleanup.removeSpace(spaceId); 
+ protected static void remove(String spaceId) {
+ TestWithSpaceCleanup.removeSpace(spaceId);
  }
 
  public static ValidatableResponse createSpace(String content) {
@@ -498,11 +499,11 @@ public static void postFeatures(String spaceId, FeatureCollection features, Auth
  .post(getSpacesPath() + "/"+ spaceId +"/features");
  }
 
- protected static void makeComposite(String spaceId, String newExtendingId) {
+ protected static void makeComposite(String spaceId, String extendedSpaceId) {
  given()
  .contentType(APPLICATION_JSON)
  .headers(getAuthHeaders(AuthProfile.ACCESS_OWNER_1_ADMIN))
- .body("{\"extends\":{\"spaceId\":\"" + newExtendingId + "\"}}")
+ .body(new JsonObject().put("extends", new JsonObject().put("spaceId", extendedSpaceId)).toString())
  .when()
  .patch("/spaces/" + spaceId)
  .then()

xyz-hub-test/src/test/java/com/here/xyz/hub/rest/VersioningCompositeGetFeaturesIT.java

@@ -26,6 +26,7 @@
 import com.here.xyz.models.geojson.coordinates.PointCoordinates;
 import com.here.xyz.models.geojson.implementation.Point;
 import com.here.xyz.models.geojson.implementation.Properties;
+import com.here.xyz.models.geojson.implementation.XyzNamespace;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.FixMethodOrder;
@@ -167,4 +168,14 @@ public void testGetFeatureVersionStarCompositeVersionsBeforeAndAfterEdit() {
  .body("features[0].properties.@ns:com:here:xyz.version", equalTo(0))
  .body("features[1].properties.@ns:com:here:xyz.version", equalTo(3));
  }
+
+ @Test
+ public void testWriteFeatureWithConflictDetection() {
+ postFeature(BASE, newFeature().withId("f2"), ACCESS_ALL);
+ postFeature(DELTA, newFeature()
+ .withId("f2")
+ .withProperties(new Properties()
+ .withXyzNamespace(new XyzNamespace().withVersion(0))
+ .with("otherKey", "otherValue")), ACCESS_ALL, true);
+ }
 }

xyz-psql-connector/src/main/java/com/here/xyz/psql/query/WriteFeatures.java

@@ -23,6 +23,8 @@
 import static com.here.xyz.responses.XyzError.CONFLICT;
 import static com.here.xyz.responses.XyzError.EXCEPTION;
 import static com.here.xyz.responses.XyzError.NOT_FOUND;
+import static com.here.xyz.util.db.pg.SQLError.FEATURE_EXISTS;
+import static com.here.xyz.util.db.pg.SQLError.FEATURE_NOT_EXISTS;
 import static com.here.xyz.util.db.pg.SQLError.RETRYABLE_VERSION_CONFLICT;
 import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.PARTITION_SIZE;
 
@@ -113,12 +115,16 @@ protected FeatureCollection run(DataSourceProvider dataSourceProvider) throws Er
  catch (SQLException e) {
  final String message = e.getMessage();
  String cleanMessage = message.contains("\n") ? message.substring(0, message.indexOf("\n")) : message;
- throw switch (SQLError.fromErrorCode(e.getSQLState())) {
- case FEATURE_EXISTS, VERSION_CONFLICT_ERROR, MERGE_CONFLICT_ERROR, RETRYABLE_VERSION_CONFLICT -> new ErrorResponseException(CONFLICT, cleanMessage, e);
- case DUPLICATE_KEY -> new ErrorResponseException(CONFLICT, "Conflict while writing features.", e); //TODO: Handle all conflicts in FeatureWriter properly
- case FEATURE_NOT_EXISTS -> new ErrorResponseException(NOT_FOUND, cleanMessage, e);
- case ILLEGAL_ARGUMENT -> new ErrorResponseException(XyzError.ILLEGAL_ARGUMENT, cleanMessage, e);
- case XYZ_EXCEPTION, UNKNOWN -> new ErrorResponseException(EXCEPTION, e.getMessage(), e);
+ SQLError sqlError = SQLError.fromErrorCode(e.getSQLState());
+ String details = message.contains("\n") && sqlError != FEATURE_EXISTS && sqlError != FEATURE_NOT_EXISTS
+ ? message.substring(message.indexOf("\n") + 1)
+ : null;
+ throw switch (sqlError) {
+ case FEATURE_EXISTS, VERSION_CONFLICT_ERROR, MERGE_CONFLICT_ERROR, RETRYABLE_VERSION_CONFLICT -> new ErrorResponseException(CONFLICT, cleanMessage, e).withInternalDetails(details);
+ case DUPLICATE_KEY -> new ErrorResponseException(CONFLICT, "Conflict while writing features.", e).withInternalDetails(details); //TODO: Handle all conflicts in FeatureWriter properly
+ case FEATURE_NOT_EXISTS -> new ErrorResponseException(NOT_FOUND, cleanMessage, e).withInternalDetails(details);
+ case ILLEGAL_ARGUMENT -> new ErrorResponseException(XyzError.ILLEGAL_ARGUMENT, cleanMessage, e).withInternalDetails(details);
+ case XYZ_EXCEPTION, UNKNOWN -> new ErrorResponseException(EXCEPTION, e.getMessage(), e).withInternalDetails(details);
  };
  }
  }

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

@@ -81,17 +81,17 @@ class DatabaseWriter {
  * Executes a prepared query using the specified parameterSet and returns the results after parsing them using the
  * specified resultParser.
  * @param {PreparedPlan} plan
- * @param {object[][]} parameterSet
+ * @param {object[][]} parameterSets
  * @param {(function(object[]) : FeatureModificationExecutionResult)[]} resultParsers
  * @param {object} queryOptions
  * @returns {FeatureModificationExecutionResult[]}
  * @private
  */
- _executePlan(plan, parameterSet, resultParsers, queryOptions) {
+ _executePlan(plan, parameterSets, resultParsers, queryOptions) {
  let results = [];
- for (let index in parameterSet) {
+ for (let index in parameterSets) {
  try {
- let result = resultParsers[index](plan.execute(parameterSet[index]));
+ let result = resultParsers[index](plan.execute(parameterSets[index]));
  if (result != null)
  results.push(result);
  }
@@ -100,9 +100,9 @@ class DatabaseWriter {
  let exceptionToThrow;
  let onVersionConflict = queryOptions?.onVersionConflict;
  if (!onVersionConflict || onVersionConflict == "ERROR")
- exceptionToThrow = new VersionConflictError(`Version conflict while trying to write feature with ID ${parameterSet[0]} in version ${parameterSet[1]}.`);
+ exceptionToThrow = new VersionConflictError(`Version conflict while trying to write feature with ID ${parameterSets[index][0]} in version ${parameterSets[index][1]}.`);
  else
- exceptionToThrow = new RetryableVersionConflictError(`Retryable version conflict while trying to write feature with ID ${parameterSet[0]} in version ${parameterSet[1]}.`);
+ exceptionToThrow = new RetryableVersionConflictError(`Retryable version conflict while trying to write feature with ID ${parameterSets[index][0]} in version ${parameterSets[index][1]}.`);
  throw exceptionToThrow.withHint("The feature has been modified in the meantime.");
  }
  else

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

@@ -586,15 +586,18 @@ class FeatureWriter {
  else
  feature = resultSet[0].jsondata;
 
+ let dataset = resultSet[0].dataset;
+ let version = FeatureWriter._isComposite() && dataset < this.tables.length - 1 ? 0 : Number(resultSet[0].version);
+
  feature.id = resultSet[0].id;
  feature.geometry = resultSet[0].geo;
- feature.properties[XYZ_NS].version = Number(resultSet[0].version);
+ feature.properties[XYZ_NS].version = version;
  Object.defineProperty(feature, "operation", {
  value: resultSet[0].operation,
  enumerable: false
  });
  Object.defineProperty(feature, "dataset", {
- value: resultSet[0].dataset,
+ value: dataset,
  enumerable: false
  });
  Object.defineProperty(feature, "containingDatasets", {
@@ -825,7 +828,7 @@ class FeatureWriter {
 
  _loadFeature(id, version, tables) {
  let tableAliases = tables.map((table, i) => "t" + (tables.length - i - 1));
-  let branchTableMaxVersion = i => i == tables.length - 1 || FeatureWriter._isComposite() ? "" : `AND version <= ${this.tableBaseVersions[i + 1] - this.tableBaseVersions[i]}`;
+ let branchTableMaxVersion = i => i == tables.length - 1 || FeatureWriter._isComposite() ? "" : `AND version <= ${this.tableBaseVersions[i + 1] - this.tableBaseVersions[i]}`;
  let whereConditions = tables.map((table, i) => `WHERE id = $1 AND ${version == "HEAD" ? `next_version = ${MAX_BIG_INT}` : `version = ${version - this.tableBaseVersions[i]}`} ${branchTableMaxVersion(i)} AND operation != $2`).reverse();
  let tableBaseVersions = tables.map((table, i) => this.tableBaseVersions[i]).reverse();
 

xyz-util/src/test/js/plv8.js

@@ -24,14 +24,14 @@ let pgClient = new Client({
  host: "localhost",
  user: "postgres",
  password: "password",
- database: "postgres"
+ database: "postgres",
+ options: `-c search_path=public,topology,hub.common,hub.geo,hub.feature_writer,hub.h3,hub.ext`
 });
 
 let clientConnected = false;
-pgClient.connect().then(err => {
- //pgClient.query(`SET search_path = "public", "topology", "hub.common", "hub.geo", "hub.feature_writer", "hub.h3", "hub.ext"`);
+pgClient.connect().then(result => {
  clientConnected = true;
- console.log(err);
+ console.log(result);
 });
 
 class PreparedPlan {
@@ -84,12 +84,42 @@ global.plv8 = {
  deasync.loopWhile(() => !clientConnected);
 
  let queryResult = null;
- pgClient.query(sql, params).then(result => queryResult = result);
+ pgClient.query(substituteSQL(sql, params)).then(result => queryResult = result);
  deasync.loopWhile(() => queryResult == null);
  return queryResult.rows;
  },
  prepare(sql, typeNames = []) {
  return new PreparedPlan(sql, typeNames).prepare();
  }
  //TODO: Support prepared / batch queries
-};
+};
+
+//NOTE: The following helper function is necessary, because the NPN pg module does not support to have multiple SQL statements within prepared statements or queries.
+/**
+ * Substitute $1, $2, ... placeholders in a SQL string with values from params array
+ * @param {string} sql - SQL string with $1, $2, ... placeholders
+ * @param {Array} params - Array of parameter values
+ * @returns {string} SQL string with parameters substituted
+ */
+function substituteSQL(sql, params) {
+ return sql.replace(/\$(\d+)/g, (match, number) => {
+ const index = parseInt(number, 10) - 1;
+ if (index < 0 || index >= params.length) {
+ throw new Error(`No parameter provided for placeholder ${match}`);
+ }
+ const value = params[index];
+
+ if (value === null || value === undefined) return 'NULL';
+ if (typeof value === 'number') return value.toString();
+ if (typeof value === 'boolean') return value ? 'TRUE' : 'FALSE';
+ if (value instanceof Date) return `'${value.toISOString()}'`;
+
+ // Escape single quotes for strings
+ if (typeof value === 'string') {
+ return `'${value.replace(/'/g, "''")}'`;
+ }
+
+ // For objects/arrays, JSON stringify
+ return `'${JSON.stringify(value).replace(/'/g, "''")}'`;
+ });
+}