DS-1723,DS-1841:

* Introduce new Connector "PSQLXyzNLConnector".
* Add new TableLayout "OLD_LAYOUT_WITH_SEARCHABLE".
* Start implementing support for index creation on searchable columns.

Signed-off-by: mchrza <[email protected]>

Author: mchrza

xyz-hub-service/src/main/resources/connectors.json

@@ -75,7 +75,7 @@
         "type": "Embedded",
         "id": "xyz-psql-local",
         "warmUp": 1,
-        "className": "com.here.xyz.psql.PSQLXyzConnector",
+        "className": "com.here.xyz.psql.PSQLXyzNLConnector",
         "env": {
           "ECPS_PHRASE": "local"
         },

xyz-jobs/xyz-job-steps/src/main/java/com/here/xyz/jobs/steps/impl/CreateIndex.java

@@ -24,6 +24,7 @@
 import com.here.xyz.jobs.steps.impl.tools.ResourceAndTimeCalculator;
 import com.here.xyz.jobs.steps.resources.Load;
 import com.here.xyz.jobs.steps.resources.TooManyResourcesClaimed;
+import com.here.xyz.util.db.ConnectorParameters;
 import com.here.xyz.util.db.SQLQuery;
 import com.here.xyz.util.db.pg.IndexHelper;
 import com.here.xyz.util.db.pg.IndexHelper.Index;
@@ -131,9 +132,11 @@ private SQLQuery buildOnDemandIndexCreationQuery(OnDemandIndex onDemandIndex) th
     List<SQLQuery> indexCreationQueries = new ArrayList<>();
 
     indexCreationQueries.addAll(loadPartitionNamesOf(rootTableName).stream()
-        .map(partitionTableName -> new SQLQuery(IndexHelper.buildOnDemandIndexCreationQuery(schema, partitionTableName, onDemandIndex.getPropertyPath(), true).toExecutableQueryString()))
+        .map(partitionTableName -> new SQLQuery(IndexHelper.buildOnDemandIndexCreationQuery(schema, partitionTableName, onDemandIndex,
+                ConnectorParameters.TableLayout.OLD_LAYOUT, true).toExecutableQueryString()))
         .toList());
-    indexCreationQueries.add(new SQLQuery(IndexHelper.buildOnDemandIndexCreationQuery(schema, rootTableName, onDemandIndex.getPropertyPath(),true).toExecutableQueryString()));
+    indexCreationQueries.add(new SQLQuery(IndexHelper.buildOnDemandIndexCreationQuery(schema, rootTableName,
+            onDemandIndex, ConnectorParameters.TableLayout.OLD_LAYOUT,true).toExecutableQueryString()));
 
     return SQLQuery.join(indexCreationQueries, ";");
   }

xyz-jobs/xyz-job-steps/src/main/java/com/here/xyz/jobs/steps/impl/transport/TaskedSpaceBasedStep.java

@@ -406,6 +406,8 @@ private void startTask(TaskProgress<I> taskProgressAndItem) throws TooManyResour
 
   @Override
   public void execute(boolean resume) throws Exception {
+    infoLog(STEP_EXECUTE, "execute called with resume=" + resume);
+
     //The following code is running synchronously till the first task is getting started.
     String schema = getSchema(db());
     if (!resume) {

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

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ * 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
+ */
+
+package com.here.xyz.psql;
+
+import com.here.xyz.events.ModifySpaceEvent;
+import com.here.xyz.psql.query.ModifySpace;
+import com.here.xyz.responses.SuccessResponse;
+
+import static com.here.xyz.util.db.ConnectorParameters.TableLayout.OLD_LAYOUT_WITH_SEARCHABLE;
+
+public class PSQLXyzNLConnector extends PSQLXyzConnector {
+
+  @Override
+  protected SuccessResponse processModifySpaceEvent(ModifySpaceEvent event) throws Exception {
+    return write(new ModifySpace(event).withTableLayout(OLD_LAYOUT_WITH_SEARCHABLE));
+  }
+}

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

@@ -25,7 +25,6 @@
 import static com.here.xyz.events.ModifySpaceEvent.Operation.UPDATE;
 import static com.here.xyz.psql.query.helpers.versioning.GetNextVersion.VERSION_SEQUENCE_SUFFIX;
 import static com.here.xyz.responses.XyzError.ILLEGAL_ARGUMENT;
-import static com.here.xyz.util.db.ConnectorParameters.TableLayout.NEW_LAYOUT;
 import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.SCHEMA;
 import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.TABLE;
 import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.buildCreateSpaceTableQueries;
@@ -45,22 +44,8 @@
 import java.util.List;
 import static com.here.xyz.util.db.pg.XyzSpaceTableHelper.buildCleanUpQuery;
 import static com.here.xyz.util.db.pg.IndexHelper.getActivatedSearchableProperties;
-import static com.here.xyz.util.db.ConnectorParameters.TableLayout.OLD_LAYOUT;
-
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.here.xyz.connectors.ErrorResponseException;
-import com.here.xyz.events.ModifySpaceEvent;
-import com.here.xyz.events.ModifySpaceEvent.Operation;
-import com.here.xyz.responses.SuccessResponse;
-import com.here.xyz.util.db.ConnectorParameters;
-import com.here.xyz.util.db.SQLQuery;
-import com.here.xyz.util.db.datasource.DataSourceProvider;
-import com.here.xyz.util.db.pg.IndexHelper;
-import com.here.xyz.util.db.pg.XyzSpaceTableHelper;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.List;
+import static com.here.xyz.util.db.ConnectorParameters.TableLayout;
+import static com.here.xyz.util.db.ConnectorParameters.TableLayout.NEW_LAYOUT;
 
 public class ModifySpace extends ExtendedSpace<ModifySpaceEvent, SuccessResponse> {
 
@@ -140,7 +125,9 @@ private void validateModifySpaceEvent(ModifySpaceEvent event) throws ErrorRespon
 
     @Override
     protected SQLQuery buildQuery(ModifySpaceEvent event) throws SQLException {
-        if(getTableLayout().equals(ConnectorParameters.TableLayout.OLD_LAYOUT)){
+        TableLayout tableLayout = getTableLayout();
+
+        if(tableLayout.isOld()){
             if (event.getOperation() == CREATE || event.getOperation() == UPDATE) {
                 List<SQLQuery> queries = new ArrayList<>();
                 final String table = getDefaultTable(event);
@@ -151,7 +138,7 @@ protected SQLQuery buildQuery(ModifySpaceEvent event) throws SQLException {
                             = getActivatedSearchableProperties(event.getSpaceDefinition().getSearchableProperties());
 
                     queries.addAll(buildCreateSpaceTableQueries(getSchema(), table, activatedSearchableProperties,
-                            event.getSpace(), OLD_LAYOUT));
+                            event.getSpace(), tableLayout));
                 }
 
                 //Write metadata
@@ -160,20 +147,20 @@ protected SQLQuery buildQuery(ModifySpaceEvent event) throws SQLException {
                 return SQLQuery.batchOf(queries).withLock(table);
             }
             else if (event.getOperation() == DELETE) {
-                return SQLQuery.batchOf(buildCleanUpQuery(event, getSchema(), getDefaultTable(event), VERSION_SEQUENCE_SUFFIX, OLD_LAYOUT));
+                return SQLQuery.batchOf(buildCleanUpQuery(event, getSchema(), getDefaultTable(event), VERSION_SEQUENCE_SUFFIX, tableLayout));
             }
-        }else if(getTableLayout().equals(ConnectorParameters.TableLayout.NEW_LAYOUT)){
+        }else if(tableLayout.equals(NEW_LAYOUT)){
             if (event.getOperation() == CREATE) {
                 final String table = getDefaultTable(event);
                 List<SQLQuery> queries = new ArrayList<>(buildCreateSpaceTableQueries(getSchema(), table,
                         //No OnDemandIndices are supported in V2
-                        null, event.getSpace(), NEW_LAYOUT));
+                        null, event.getSpace(), tableLayout));
                 return SQLQuery.batchOf(queries).withLock(table);
             }
             else if (event.getOperation() == DELETE)
-                return SQLQuery.batchOf(buildCleanUpQuery(event, getSchema(), getDefaultTable(event), VERSION_SEQUENCE_SUFFIX, NEW_LAYOUT));
+                return SQLQuery.batchOf(buildCleanUpQuery(event, getSchema(), getDefaultTable(event), VERSION_SEQUENCE_SUFFIX, tableLayout));
         }
-        throw new IllegalArgumentException("Unsupported Table Layout: " + getTableLayout());
+        throw new IllegalArgumentException("Unsupported Table Layout: " + tableLayout);
     }
 
     @Override

xyz-psql-connector/src/test/java/com/here/xyz/psql/PSQLIndexIT.java

@@ -27,6 +27,7 @@
 import com.here.xyz.responses.StatisticsResponse;
 import com.here.xyz.responses.SuccessResponse;
 import com.here.xyz.responses.XyzError;
+import com.here.xyz.util.db.ConnectorParameters;
 import com.here.xyz.util.db.pg.IndexHelper.OnDemandIndex;
 import org.junit.After;
 import org.junit.Before;
@@ -182,7 +183,7 @@ public void testOnDemandIdxCreation() throws Exception {
             List<OnDemandIndex> activatedSearchableProperties = getActivatedSearchableProperties(searchableProperties);
 
             for(OnDemandIndex onDemandIndex : activatedSearchableProperties)
-                stmt.execute(buildOnDemandIndexCreationQuery("public", "foo", onDemandIndex.getPropertyPath(), false).toExecutableQueryString());
+                stmt.execute(buildOnDemandIndexCreationQuery("public", "foo", onDemandIndex, ConnectorParameters.TableLayout.OLD_LAYOUT, false).toExecutableQueryString());
 
             // Check which Indices are available
             ResultSet resultSet = stmt.executeQuery( DhString.format("select idx_property, src from xyz_index_list_all_available(%s, 'foo');",sqlSpaceSchema));

xyz-util/src/main/java/com/here/xyz/util/db/ConnectorParameters.java

@@ -40,7 +40,23 @@ public class ConnectorParameters {
 
   private TableLayout tableLayout;
   public enum TableLayout {
-    OLD_LAYOUT, NEW_LAYOUT;
+    OLD_LAYOUT(false),
+    OLD_LAYOUT_WITH_SEARCHABLE(true),
+    NEW_LAYOUT(false);
+
+    private final boolean hasSearchableColumn;
+
+    TableLayout(boolean hasSearchableColumn) {
+      this.hasSearchableColumn = hasSearchableColumn;
+    }
+
+    public boolean hasSearchableColumn() {
+      return hasSearchableColumn;
+    }
+
+    public boolean isOld() {
+      return this == OLD_LAYOUT || this == OLD_LAYOUT_WITH_SEARCHABLE;
+    }
   }
 
   public ConnectorParameters() {}