keboola · zajca · Nov 12, 2019 · Nov 13, 2019 · Nov 13, 2019 · Nov 13, 2019
diff --git a/Dockerfile b/Dockerfile
@@ -24,6 +24,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
         debsig-verify \
         dirmngr \
         gpg-agent \
+        libicu-dev \
 	&& rm -r /var/lib/apt/lists/* \
 	&& sed -i 's/^# *\(en_US.UTF-8\)/\1/' /etc/locale.gen \
 	&& locale-gen \
@@ -64,6 +65,9 @@ RUN mkdir -p ~/.gnupg \
     && gpg --batch --delete-key --yes $SNOWFLAKE_GPG_KEY \
     && dpkg -i /tmp/snowflake-odbc.deb
 
+RUN docker-php-ext-configure intl \
+    && docker-php-ext-install intl
+
 ## Composer - deps always cached unless changed
 # First copy only composer files
 COPY composer.* /code/

diff --git a/composer.json b/composer.json
@@ -7,6 +7,7 @@
         "keboola/db-adapter-snowflake": "^1.0.0",
         "keboola/php-component": "^7.0.1",
         "microsoft/azure-storage-blob": "^1.4",
+        "pcrov/jsonreader": "^1.0",
         "symfony/stopwatch": "^4.3"
     },
     "require-dev": {
@@ -32,6 +33,7 @@
         }
     },
     "scripts": {
+        "testHuge": "php ./tests/testHugeManifest.php",
         "tests-unit": "phpunit tests/unit",
         "tests-functional": "phpunit tests/functional",
         "tests": [

diff --git a/src/Backend/BackendImportAdapterInterface.php b/src/Backend/BackendImportAdapterInterface.php
@@ -4,6 +4,7 @@
 
 namespace Keboola\Db\ImportExport\Backend;
 
+use Generator;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Storage\DestinationInterface;
 use Keboola\Db\ImportExport\Storage\SourceInterface;
@@ -12,12 +13,9 @@ interface BackendImportAdapterInterface
 {
     public function __construct(SourceInterface $source);
 
-    /**
-     * @return string[]
-     */
     public function getCopyCommands(
         DestinationInterface $destination,
         ImportOptions $importOptions,
         string $stagingTableName
-    ): array;
+    ): Generator;
 }
diff --git a/src/Backend/Snowflake/SnowflakeImportAdapterInterface.php b/src/Backend/Snowflake/SnowflakeImportAdapterInterface.php
@@ -4,6 +4,7 @@
 
 namespace Keboola\Db\ImportExport\Backend\Snowflake;
 
+use Generator;
 use Keboola\Db\ImportExport\Backend\ImportState;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Backend\BackendImportAdapterInterface;
@@ -15,11 +16,10 @@ interface SnowflakeImportAdapterInterface extends BackendImportAdapterInterface
     /**
      * Snowflake import is handled differently for copy table2table and file2table
      *
-     * @param string[] $commands - sql commands array
      * @return int - number of imported rows
      */
     public function executeCopyCommands(
-        array $commands,
+        Generator $commands,
         Connection $connection,
         DestinationInterface $destination,
         ImportOptions $importOptions,

diff --git a/src/Storage/ABS/SnowflakeImportAdapter.php b/src/Storage/ABS/SnowflakeImportAdapter.php
@@ -4,6 +4,7 @@
 
 namespace Keboola\Db\ImportExport\Storage\ABS;
 
+use Generator;
 use Keboola\Csv\CsvOptions;
 use Keboola\Db\ImportExport\Backend\ImporterInterface;
 use Keboola\Db\ImportExport\Backend\ImportState;
@@ -35,7 +36,7 @@ public function __construct(SourceInterface $source)
      * @param Table $destination
      */
     public function executeCopyCommands(
-        array $commands,
+        Generator $commands,
         Connection $connection,
         DestinationInterface $destination,
         ImportOptions $importOptions,
@@ -62,33 +63,60 @@ public function getCopyCommands(
         DestinationInterface $destination,
         ImportOptions $importOptions,
         string $stagingTableName
-    ): array {
-        $filesToImport = $this->source->getManifestEntries();
-        $commands = [];
-        foreach (array_chunk($filesToImport, ImporterInterface::SLICED_FILES_CHUNK_SIZE) as $entries) {
-            $commands[] = sprintf(
-                'COPY INTO %s.%s 
+    ): Generator {
+        $entriesInChunk = [];
+        foreach ($this->source->getManifestEntries() as $entry) {
+            $entriesInChunk[] = $entry;
+            if (count($entriesInChunk) === ImporterInterface::SLICED_FILES_CHUNK_SIZE) {
+                yield $this->getCopyCommand(
+                    $destination,
+                    $importOptions,
+                    $stagingTableName,
+                    $entriesInChunk
+                );
+                $entriesInChunk = [];
+            }
+        }
+        if (!empty($entriesInChunk)) {
+            yield $this->getCopyCommand(
+                $destination,
+                $importOptions,
+                $stagingTableName,
+                $entriesInChunk
+            );
+        }
+    }
+
+    /**
+     * @param Table $destination
+     */
+    private function getCopyCommand(
+        DestinationInterface $destination,
+        ImportOptions $importOptions,
+        string $stagingTableName,
+        array $entriesInChunk
+    ): string {
+        return sprintf(
+            'COPY INTO %s.%s 
 FROM %s
 CREDENTIALS=(AZURE_SAS_TOKEN=\'%s\')
 FILE_FORMAT = (TYPE=CSV %s)
 FILES = (%s)',
-                QueryBuilder::quoteIdentifier($destination->getSchema()),
-                QueryBuilder::quoteIdentifier($stagingTableName),
-                QueryBuilder::quote($this->source->getContainerUrl()),
-                $this->source->getSasToken(),
-                implode(' ', $this->getCsvCopyCommandOptions($importOptions, $this->source->getCsvOptions())),
-                implode(
-                    ', ',
-                    array_map(
-                        function ($entry) {
-                            return QueryBuilder::quote(strtr($entry, [$this->source->getContainerUrl() => '']));
-                        },
-                        $entries
-                    )
+            QueryBuilder::quoteIdentifier($destination->getSchema()),
+            QueryBuilder::quoteIdentifier($stagingTableName),
+            QueryBuilder::quote($this->source->getContainerUrl()),
+            $this->source->getSasToken(),
+            implode(' ', $this->getCsvCopyCommandOptions($importOptions, $this->source->getCsvOptions())),
+            implode(
+                ', ',
+                array_map(
+                    function ($entry) {
+                        return QueryBuilder::quote(strtr($entry['url'], [$this->source->getContainerUrl() => '']));
+                    },
+                    $entriesInChunk
                 )
-            );
-        }
-        return $commands;
+            )
+        );
     }
 
     private function getCsvCopyCommandOptions(

diff --git a/src/Storage/ABS/SourceFile.php b/src/Storage/ABS/SourceFile.php
@@ -4,10 +4,12 @@
 
 namespace Keboola\Db\ImportExport\Storage\ABS;
 
+use Generator;
 use Keboola\Csv\CsvOptions;
 use Keboola\Db\ImportExport\Backend\BackendImportAdapterInterface;
 use Keboola\Db\ImportExport\Backend\ImporterInterface;
 use Keboola\Db\ImportExport\Backend\Snowflake\Importer as SnowflakeImporter;
+use Keboola\Db\ImportExport\Storage\ManifestReader;
 use Keboola\Db\ImportExport\Storage\NoBackendAdapterException;
 use Keboola\Db\ImportExport\Storage\SourceInterface;
 use MicrosoftAzure\Storage\Blob\BlobRestProxy;
@@ -54,10 +56,11 @@ public function getCsvOptions(): CsvOptions
         return $this->csvOptions;
     }
 
-    public function getManifestEntries(): array
+    public function getManifestEntries(): Generator
     {
         if (!$this->isSliced) {
-            return [$this->getContainerUrl() . $this->filePath];
+            yield from [['url' => $this->getContainerUrl() . $this->filePath]];
+            return;
         }
 
         $SASConnectionString = sprintf(
@@ -74,9 +77,6 @@ public function getManifestEntries(): array
         );
 
         $getResult = $blobClient->getBlob($this->container, $this->filePath);
-        $manifest = \GuzzleHttp\json_decode(stream_get_contents($getResult->getContentStream()), true);
-        return array_map(function ($entry) {
-            return $entry['url'];
-        }, $manifest['entries']);
+        yield from ManifestReader::readEntries($getResult->getContentStream());
     }
 }
diff --git a/src/Storage/ManifestReader.php b/src/Storage/ManifestReader.php
@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\Db\ImportExport\Storage;
+
+use Generator;
+use pcrov\JsonReader\JsonReader;
+
+class ManifestReader
+{
+    /**
+     * @param resource $stream
+     */
+    public static function readEntries(
+        $stream
+    ): Generator {
+        $reader = new JsonReader();
+        $reader->stream($stream);
+
+        $reader->read('entries');
+        $depth = $reader->depth(); // Check in a moment to break when the array is done.
+
+        $reader->read(); // Step to the first element.
+        do {
+            if ($reader->value() === null) {
+                // on empty entries lib will return null item
+                continue;
+            }
+            yield $reader->value();
+        } while ($reader->next() && $reader->depth() > $depth); // Read each sibling.
+
+        $reader->close();
+    }
+}
diff --git a/src/Storage/Snowflake/SnowflakeImportAdapter.php b/src/Storage/Snowflake/SnowflakeImportAdapter.php
@@ -4,9 +4,9 @@
 
 namespace Keboola\Db\ImportExport\Storage\Snowflake;
 
+use Generator;
 use Keboola\Db\ImportExport\Backend\ImportState;
 use Keboola\Db\ImportExport\ImportOptions;
-use Keboola\Db\ImportExport\Backend\Snowflake\Helper\QuoteHelper;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportAdapterInterface;
 use Keboola\Db\ImportExport\Storage\DestinationInterface;
 use Keboola\Db\ImportExport\Storage\SourceInterface;
@@ -32,14 +32,14 @@ public function __construct(SourceInterface $source)
      * @param Table $destination
      */
     public function executeCopyCommands(
-        array $commands,
+        Generator $commands,
         Connection $connection,
         DestinationInterface $destination,
         ImportOptions $importOptions,
         ImportState $importState
     ): int {
         $importState->startTimer('copyToStaging');
-        $connection->query($commands[0]);
+        $connection->query($commands->current());
         $rows = $connection->fetchAll(sprintf(
             'SELECT COUNT(*) AS "count" FROM %s.%s',
             QueryBuilder::quoteIdentifier($destination->getSchema()),
@@ -56,7 +56,7 @@ public function getCopyCommands(
         DestinationInterface $destination,
         ImportOptions $importOptions,
         string $stagingTableName
-    ): array {
+    ): Generator {
         $quotedColumns = array_map(function ($column) {
             return QueryBuilder::quoteIdentifier($column);
         }, $importOptions->getColumns());
@@ -75,6 +75,6 @@ public function getCopyCommands(
             QueryBuilder::quoteIdentifier($this->source->getTableName()),
         );
 
-        return [$sql];
+        yield $sql;
     }
 }
diff --git a/src/Storage/Snowflake/Table.php b/src/Storage/Snowflake/Table.php
@@ -8,7 +8,6 @@
 use Keboola\Db\ImportExport\Backend\BackendImportAdapterInterface;
 use Keboola\Db\ImportExport\Backend\ExporterInterface;
 use Keboola\Db\ImportExport\Backend\ImporterInterface;
-use Keboola\Db\ImportExport\Backend\Snowflake\Helper\QuoteHelper;
 use Keboola\Db\ImportExport\Backend\Snowflake\Importer as SnowflakeImporter;
 use Keboola\Db\ImportExport\Storage\DestinationInterface;
 use Keboola\Db\ImportExport\Storage\NoBackendAdapterException;