CopySpacePost: decrease skipCount threshold 50k (#1812)

* CopySpacePost: decrease skipCount threshold 50k

Signed-off-by: qGYdXbY2 <47661341+qGYdXbY2@users.noreply.github.com>

* startTask: withRetryableErrorCodes

Signed-off-by: qGYdXbY2 <47661341+qGYdXbY2@users.noreply.github.com>

* rm unused import

Signed-off-by: qGYdXbY2 <47661341+qGYdXbY2@users.noreply.github.com>

---------

Signed-off-by: qGYdXbY2 <47661341+qGYdXbY2@users.noreply.github.com>

Author: qGYdXbY2

xyz-jobs/xyz-job-service/src/main/java/com/here/xyz/jobs/steps/compiler/SpaceCopy.java

@@ -160,18 +160,18 @@ public static CompilationStepGraph compile(String jobId, DatasetDescription.Spac
         outputMetadata = Map.of(target.getClass().getSimpleName().toLowerCase(), targetSpaceId);
 
       long DROPCREATEINDEX_THRESHOLD = 4_000_000;
-      boolean skipCountStatistics = sourceFeatureCount >= DROPCREATEINDEX_THRESHOLD;
+      long SKIPCOUNT_THRESHOLD = 50_000;
 
       CopySpacePost postCopySpace = new CopySpacePost()
           .withSpaceId(targetSpaceId)
           .withJobId(jobId)
           .withOutputMetadata(outputMetadata)
           .withInputSets(List.of(new InputSet(nextSpaceVersion.getOutputSet(VERSION))))
-          .withSkipCounts( skipCountStatistics );
+          .withSkipCounts( sourceFeatureCount >= SKIPCOUNT_THRESHOLD );
 
       startGraph.addExecution(postCopySpace);
 
-      boolean useDropIndexOptimization =    skipCountStatistics
+      boolean useDropIndexOptimization =    sourceFeatureCount >= DROPCREATEINDEX_THRESHOLD
                                             // target is empty and no filtering
                                          && targetFeatureCount <= 0
                                          && filters == null;

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

@@ -440,7 +440,8 @@ private void startTask(TaskProgress<I> taskProgressAndItem) throws TooManyResour
       //We can`t use the default callback here because we are reporting success during onAsyncUpdate in Java.
       //The failureCallback is still needed to report failures from the DB-side
       String failureCallback = buildFailureCallbackQuery().substitute().text().replaceAll("'", "''");
-      runReadQueryAsync(buildTaskQuery(taskProgressAndItem.getTaskId(), (I) taskProgressAndItem.getTaskInput(), failureCallback),
+      runReadQueryAsync(buildTaskQuery(taskProgressAndItem.getTaskId(), (I) taskProgressAndItem.getTaskInput(), failureCallback)
+                         .withRetryableErrorCodes(RETRYABLE_SQL_CODES),
                 queryRunsOnWriter() ? dbWriter() : dbReader(), 0d/*perItemAcus.doubleValue()*/,  false);
     }
   }
@@ -708,7 +709,7 @@ private TaskProgress getTaskProgress() throws WebClientException, SQLException,
   }
 
   private static SQLQuery buildTaskTableStatement(String schema, Step step) {
-    return new SQLQuery("""          
+    return new SQLQuery("""
             CREATE TABLE ${schema}.${table}
             (
             	task_id SERIAL,
@@ -727,7 +728,7 @@ private static SQLQuery buildTaskTableStatement(String schema, Step step) {
 
   private SQLQuery resetTaskItemWhichAreNotFinalized(String schema, String stepId) {
     infoLog(STEP_EXECUTE, "Reset task items for restart.");
-    return new SQLQuery("""             
+    return new SQLQuery("""
             UPDATE ${schema}.${table} t
                 SET started = false
                 WHERE started = true AND finalized = false;
@@ -804,7 +805,7 @@ private boolean insertTaskItemsInTaskTable(String schema, Step step, List<I> tas
       for (I taskInput : taskInputs) {
         String taskItem = taskInput.serialize();
 
-        insertQueries.add(new SQLQuery("""             
+        insertQueries.add(new SQLQuery("""
             INSERT INTO  ${schema}.${table} AS t (task_input)
                 VALUES (#{taskItem}::JSONB);
         """)