AJDA-594 (input-mapping) job-based workspace loading for async input mapping

libs/input-mapping/src/Helper/PathHelper.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\InputMapping\Helper;
+
+use Keboola\InputMapping\Table\Options\InputTableOptions;
+use Keboola\StagingProvider\Staging\File\FileStagingInterface;
+
+class PathHelper
+{
+    public static function ensurePathDelimiter(string $path): string
+    {
+        return self::ensureNoPathDelimiter($path) . '/';
+    }
+
+    public static function ensureNoPathDelimiter(string $path): string
+    {
+        return rtrim($path, '\\/');
+    }
+
+    public static function getManifestPath(
+        FileStagingInterface $metadataStorage,
+        string $destination,
+        InputTableOptions $table,
+    ): string {
+        return self::ensurePathDelimiter($metadataStorage->getPath()) .
+            self::getDestinationFilePath($destination, $table) . '.manifest';
+    }
+
+    public static function getDataFilePath(
+        FileStagingInterface $dataStorage,
+        string $destination,
+        InputTableOptions $table,
+    ): string {
+        return self::ensurePathDelimiter($dataStorage->getPath()) .
+            self::getDestinationFilePath($destination, $table);
+    }
+
+    private static function getDestinationFilePath(string $destination, InputTableOptions $table): string
+    {
+        if (!$table->getDestination()) {
+            return $destination . '/' . $table->getSource();
+        } else {
+            return $destination . '/' . $table->getDestination();
+        }
+    }
+}

libs/input-mapping/src/Reader.php

@@ -4,6 +4,8 @@
 
 namespace Keboola\InputMapping;
 
+use InvalidArgumentException;
+use Keboola\InputMapping\Exception\InputOperationException;
 use Keboola\InputMapping\File\Options\InputFileOptions;
 use Keboola\InputMapping\File\Options\RewrittenInputFileOptions;
 use Keboola\InputMapping\Helper\InputBucketValidator;
@@ -14,8 +16,11 @@
 use Keboola\InputMapping\State\InputTableStateList;
 use Keboola\InputMapping\Table\Options\InputTableOptionsList;
 use Keboola\InputMapping\Table\Options\ReaderOptions;
+use Keboola\InputMapping\Table\Options\RewrittenInputTableOptionsList;
 use Keboola\InputMapping\Table\Result;
 use Keboola\InputMapping\Table\Strategy\AbstractStrategy as TableAbstractStrategy;
+use Keboola\InputMapping\Table\Strategy\AbstractWorkspaceStrategy;
+use Keboola\InputMapping\Table\Strategy\WorkspaceLoadQueue;
 use Keboola\InputMapping\Table\TableDefinitionResolver;
 use Keboola\StorageApiBranch\ClientWrapper;
 use Psr\Log\LoggerInterface;
@@ -47,32 +52,29 @@ public function downloadFiles(
         return $strategy->downloadFiles($configuration, $destination);
     }
 
-    /**
-     * @param InputTableOptionsList $tablesDefinition list of input mappings
-     * @param InputTableStateList $tablesState list of input mapping table states
-     * @param string $destination destination folder
-     * @param ReaderOptions $readerOptions
-     * @return Result
-     */
-    public function downloadTables(
-        InputTableOptionsList $tablesDefinition,
-        InputTableStateList $tablesState,
-        string $destination,
-        ReaderOptions $readerOptions,
-    ): Result {
-        $tableResolver = new TableDefinitionResolver(
+    private function createTableResolver(): TableDefinitionResolver
+    {
+        return new TableDefinitionResolver(
             $this->clientWrapper->getTableAndFileStorageClient(),
             $this->logger,
         );
-        $tablesState = TableRewriteHelperFactory::getTableRewriteHelper(
+    }
+
+    private function rewriteTableStatesDestinations(InputTableStateList $tablesState): InputTableStateList
+    {
+        return TableRewriteHelperFactory::getTableRewriteHelper(
             $this->clientWrapper->getClientOptionsReadOnly(),
         )->rewriteTableStatesDestinations(
             $tablesState,
             $this->clientWrapper,
             $this->logger,
         );
-        $tablesDefinition = $tableResolver->resolve($tablesDefinition);
-        $strategy = $this->strategyFactory->getTableInputStrategy($destination, $tablesState);
+    }
+
+    private function validateAndRewriteDevBuckets(
+        InputTableOptionsList $tablesDefinition,
+        ReaderOptions $readerOptions,
+    ): RewrittenInputTableOptionsList {
         if ($readerOptions->devInputsDisabled()
             && !$this->clientWrapper->getClientOptionsReadOnly()->useBranchStorage()
         ) {
@@ -83,17 +85,79 @@ public function downloadTables(
                 $this->clientWrapper,
             );
         }
-        $tablesDefinition = TableRewriteHelperFactory::getTableRewriteHelper(
+
+        return TableRewriteHelperFactory::getTableRewriteHelper(
             $this->clientWrapper->getClientOptionsReadOnly(),
         )->rewriteTableOptionsSources(
             $tablesDefinition,
             $this->clientWrapper,
             $this->logger,
         );
+    }
+
+
+    /**
+     * @param InputTableOptionsList $tablesDefinition list of input mappings
+     * @param InputTableStateList $tablesState list of input mapping table states
+     * @param string $destination destination folder
+     * @param ReaderOptions $readerOptions
+     * @return Result
+     */
+    public function downloadTables(
+        InputTableOptionsList $tablesDefinition,
+        InputTableStateList $tablesState,
+        string $destination,
+        ReaderOptions $readerOptions,
+    ): Result {
+        $tablesState = $this->rewriteTableStatesDestinations($tablesState);
+
+        $tablesDefinition =  $this->createTableResolver()->resolve($tablesDefinition);
+        $strategy = $this->strategyFactory->getTableInputStrategy($destination, $tablesState);
+        $tablesDefinition = $this->validateAndRewriteDevBuckets($tablesDefinition, $readerOptions);
+
         /** @var TableAbstractStrategy $strategy */
         return $strategy->downloadTables($tablesDefinition->getTables(), $readerOptions->preserveWorkspace());
     }
 
+    /**
+     * Execute only prepare and execute phases for workspace table loading
+     * Returns WorkspaceLoadQueue for later completion with waitForTableLoadCompletion()
+     *
+     * @param InputTableOptionsList $tablesDefinition list of input mappings
+     * @param InputTableStateList $tablesState list of input mapping table states
+     * @param string $destination destination folder
+     * @param ReaderOptions $readerOptions
+     * @return WorkspaceLoadQueue
+     * @throws InvalidArgumentException if strategy is not workspace-based
+     */
+    public function prepareAndExecuteTableLoads(
+        InputTableOptionsList $tablesDefinition,
+        InputTableStateList $tablesState,
+        string $destination,
+        ReaderOptions $readerOptions,
+    ): WorkspaceLoadQueue {
+        $tablesState = $this->rewriteTableStatesDestinations($tablesState);
+
+        $tablesDefinition = $this->createTableResolver()->resolve($tablesDefinition);
+        $strategy = $this->strategyFactory->getTableInputStrategy($destination, $tablesState);
+
+        // Ensure we have a workspace strategy. For file this method is not yet implemented.
+        if (!$strategy instanceof AbstractWorkspaceStrategy) {
+            throw new InputOperationException(
+                'prepareAndExecuteTableLoads() can only be used with workspace strategies',
+            );
+        }
+
+        $tablesDefinition = $this->validateAndRewriteDevBuckets($tablesDefinition, $readerOptions);
+
+        // Execute only Phase 1 & 2: Prepare and Execute
+        // Let the strategy handle the planning internally
+        return $strategy->prepareAndExecuteTableLoads(
+            $tablesDefinition->getTables(),
+            $readerOptions->preserveWorkspace(),
+        );
+    }
+
     /**
      * Get parent runId to the current runId (defined by SAPI client)
      * @param string $runId

libs/input-mapping/src/Table/Strategy/ABS.php

@@ -5,6 +5,7 @@
 namespace Keboola\InputMapping\Table\Strategy;
 
 use Keboola\InputMapping\Exception\InvalidInputException;
+use Keboola\InputMapping\Helper\PathHelper;
 use Keboola\InputMapping\Table\Options\RewrittenInputTableOptions;
 use Keboola\StorageApi\Options\GetFileOptions;
 
@@ -36,8 +37,11 @@ public function handleExports(array $exports, bool $preserve): array
         foreach ($exports as $export) {
             /** @var RewrittenInputTableOptions $table */
             [$jobId, $table] = $export;
-            $manifestPath = $this->ensurePathDelimiter($this->metadataStorage->getPath()) .
-                $this->getDestinationFilePath($this->destination, $table) . '.manifest';
+            $manifestPath = PathHelper::getManifestPath(
+                $this->metadataStorage,
+                $this->destination,
+                $table,
+            );
             $tableInfo = $table->getTableInfo();
             $fileInfo = $this->clientWrapper->getTableAndFileStorageClient()->getFile(
                 $keyedResults[$jobId]['results']['file']['id'],

libs/input-mapping/src/Table/Strategy/AbstractStrategy.php

@@ -5,7 +5,6 @@
 namespace Keboola\InputMapping\Table\Strategy;
 
 use Keboola\InputMapping\State\InputTableStateList;
-use Keboola\InputMapping\Table\Options\InputTableOptions;
 use Keboola\InputMapping\Table\Options\RewrittenInputTableOptions;
 use Keboola\InputMapping\Table\Result;
 use Keboola\InputMapping\Table\Result\TableInfo;
@@ -16,16 +15,6 @@ abstract class AbstractStrategy implements StrategyInterface
 {
     protected readonly LoggerInterface $logger; // @phpstan-ignore-line initialized in child classes
 
-    protected function ensurePathDelimiter(string $path): string
-    {
-        return $this->ensureNoPathDelimiter($path) . '/';
-    }
-
-    protected function ensureNoPathDelimiter(string $path): string
-    {
-        return rtrim($path, '\\/');
-    }
-
     /**
      * @param RewrittenInputTableOptions[] $tables
      * @param bool $preserve
@@ -52,13 +41,4 @@ public function downloadTables(array $tables, bool $preserve): Result
 
         return $result;
     }
-
-    protected function getDestinationFilePath(string $destination, InputTableOptions $table): string
-    {
-        if (!$table->getDestination()) {
-            return $destination . '/' . $table->getSource();
-        } else {
-            return $destination . '/' . $table->getDestination();
-        }
-    }
 }