[BUG] Stream import files in interpreters to reduce memory usage

composer.json

@@ -18,9 +18,11 @@
     "ext-json": "*",
     "doctrine/dbal": "^4.4",
     "dragonmantank/cron-expression": "^3.1",
+    "halaxa/json-machine": "^1.2",
     "league/flysystem-sftp-v3": "^3.0",
     "mtdowling/jmespath.php": "^2.8",
     "nesbot/carbon": "^2.72 || ^3.8.4",
+    "openspout/openspout": "^4.24",
     "phpoffice/phpspreadsheet": "^4.3 || ^5.1",
     "pimcore/data-hub": "^2026.1.3",
     "pimcore/pimcore": "^2026.1",

src/DataSource/Interpreter/AbstractInterpreter.php

@@ -12,8 +12,9 @@
 
 namespace Pimcore\Bundle\DataImporterBundle\DataSource\Interpreter;
 
+use League\Flysystem\FilesystemException;
+use League\Flysystem\UnableToWriteFile;
 use Pimcore\Bundle\ApplicationLoggerBundle\ApplicationLogger;
-use Pimcore\Bundle\ApplicationLoggerBundle\FileObject;
 use Pimcore\Bundle\DataImporterBundle\DataSource\Interpreter\DeltaChecker\DeltaChecker;
 use Pimcore\Bundle\DataImporterBundle\Exception\InvalidInputException;
 use Pimcore\Bundle\DataImporterBundle\PimcoreDataImporterBundle;
@@ -22,6 +23,7 @@
 use Pimcore\Bundle\DataImporterBundle\Resolver\Resolver;
 use Pimcore\Model\Tool\TmpStore;
 use Pimcore\Tool\Admin;
+use Pimcore\Tool\Storage;
 use Psr\Log\LoggerAwareTrait;
 
 /**
@@ -147,10 +149,16 @@ public function interpretFile(string $path): bool
         }
 
         if ($this->doArchiveImportFile) {
-            $this->applicationLogger->info($archiveLogMessage, [
+            $context = [
                 'component' => PimcoreDataImporterBundle::LOGGER_COMPONENT_PREFIX . $this->configName,
-                'fileObject' => new FileObject(file_get_contents($path))
-            ]);
+            ];
+
+            $archivedFilePath = $this->archiveImportFile($path);
+            if ($archivedFilePath !== null) {
+                $context['fileObject'] = $archivedFilePath;
+            }
+
+            $this->applicationLogger->info($archiveLogMessage, $context);
         }
 
         $this->updateExecutionPackageInformation();
@@ -160,6 +168,40 @@ public function interpretFile(string $path): bool
 
     abstract protected function doInterpretFileAndCallProcessRow(string $path): void;
 
+    /**
+     * Streams the import file into the application-log storage instead of loading it into
+     * memory via FileObject (large source files would otherwise exhaust the memory limit).
+     * Returns the storage path in the same format FileObject produces, so the log viewer
+     * can resolve it, or null when archiving failed.
+     */
+    private function archiveImportFile(string $path): ?string
+    {
+        $storagePath = date('/Y/m/d/') . uniqid('fileobject_', true);
+
+        $stream = fopen($path, 'rb');
+        if ($stream === false) {
+            $this->logger->warning(sprintf('Could not open import file `%s` for archiving.', $path));
+
+            return null;
+        }
+
+        try {
+            Storage::get('application_log')->writeStream($storagePath, $stream);
+        } catch (FilesystemException | UnableToWriteFile $exception) {
+            $this->logger->warning(
+                sprintf('Could not archive import file to `%s`: %s', $storagePath, $exception->getMessage())
+            );
+
+            return null;
+        } finally {
+            if (is_resource($stream)) {
+                fclose($stream);
+            }
+        }
+
+        return $storagePath;
+    }
+
     protected function processImportRow(array $data)
     {
         $this->assertValidRowEncoding($data);

src/DataSource/Interpreter/JsonFileInterpreter.php

@@ -15,6 +15,9 @@
 use JmesPath\Env as JmesPath;
 use JmesPath\Parser as Parser;
 use JmesPath\SyntaxErrorException;
+use JsonMachine\Exception\JsonMachineException;
+use JsonMachine\Items;
+use JsonMachine\JsonDecoder\ExtJsonDecoder;
 use Pimcore\Bundle\DataImporterBundle\Exception\InvalidConfigurationException;
 use Pimcore\Bundle\DataImporterBundle\PimcoreDataImporterBundle;
 use Pimcore\Bundle\DataImporterBundle\Preview\Model\PreviewData;
@@ -24,6 +27,8 @@
  */
 class JsonFileInterpreter extends AbstractInterpreter
 {
+    private const UTF8_BOM = "\xEF\xBB\xBF";
+
     protected string $path;
 
     protected ?array $cachedContent = null;
@@ -54,6 +59,14 @@
 
     protected function doInterpretFileAndCallProcessRow(string $path): void
     {
+        if ($this->getStreamingJsonPointer() !== null) {
+            foreach ($this->streamItems($path) as $dataRow) {
+                $this->processImportRow($dataRow);
+            }
+
+            return;
+        }
+
         $data = $this->loadData($path);
 
         foreach ($data as $dataRow) {
@@ -84,9 +97,8 @@
      */
     protected function prepareContent($content)
     {
-        $UTF8_BOM = chr(0xEF) . chr(0xBB) . chr(0xBF);
         $first3 = substr($content, 0, 3);
-        if ($first3 === $UTF8_BOM) {
+        if ($first3 === self::UTF8_BOM) {
             $content = substr($content, 3);
         }
 
@@ -106,6 +118,10 @@
             }
         }
 
+        if ($this->getStreamingJsonPointer() !== null) {
+            return $this->validateStreamed($path);
+        }
+
         $data = $this->loadDataRaw($path);
 
         if (json_last_error() === JSON_ERROR_NONE) {
@@ -129,18 +145,22 @@
         $readRecordNumber = 0;
 
         if ($this->fileValid($path)) {
-            $data = $this->loadData($path);
+            if ($this->getStreamingJsonPointer() !== null) {
+                [$previewDataRow, $readRecordNumber] = $this->readStreamedRecord($path, $recordNumber);
+            } else {
+                $data = $this->loadData($path);
 
-            $previewDataRow = $data[$recordNumber] ?? null;
+                $previewDataRow = $data[$recordNumber] ?? null;
 
-            if (empty($previewDataRow)) {
-                $previewDataRow = end($data);
-                $readRecordNumber = count($data) - 1;
-            } else {
-                $readRecordNumber = $recordNumber;
+                if (empty($previewDataRow)) {
+                    $previewDataRow = end($data);
+                    $readRecordNumber = count($data) - 1;
+                } else {
+                    $readRecordNumber = $recordNumber;
+                }
             }
 
-            foreach ($previewDataRow as $index => $columnData) {
+            foreach ($previewDataRow ?? [] as $index => $columnData) {
                 $previewData[$index] = $columnData;
             }
 
@@ -158,4 +178,106 @@
     {
         return JmesPath::search($this->path, $data);
     }
+
+    /**
+     * Returns the JSON pointer equivalent of the configured JMESPath expression when the
+     * expression is simple enough to stream (empty, or a plain dotted field path). Complex
+     * JMESPath expressions (filters, projections, functions, ...) need the whole document
+     * in memory and return null here, falling back to the full-load code path.
+     */
+    protected function getStreamingJsonPointer(): ?string
+    {
+        if (empty($this->path)) {
+            return '';
+        }
+
+        // any expression reaching this point already passed the JMESPath parser, so a
+        // loose word-character check is enough to recognize a plain dotted field path
+        if (preg_match('/^\w+(\.\w+)*$/', $this->path) === 1) {
+            return '/' . str_replace('.', '/', $this->path);
+        }
+
+        return null;
+    }
+
+    /**
+     * Streams the records below the configured path one by one so the whole file never
+     * has to be decoded into memory at once.
+     *
+     * @return \Generator<array>
+     */
+    protected function streamItems(string $path): \Generator
+    {
+        $handle = @fopen($path, 'rb');
+        if ($handle === false) {
+            // fail loud: silently yielding nothing would let the import report success
+            // without creating any queue items
+            throw new JsonMachineException(sprintf('Could not open JSON file `%s` for reading.', $path));
+        }
+
+        try {
+            $this->skipByteOrderMark($handle);
+
+            $items = Items::fromStream($handle, [
+                'pointer' => $this->getStreamingJsonPointer(),
+                'decoder' => new ExtJsonDecoder(true),
+            ]);
+
+            foreach ($items as $item) {
+                yield $item;
+            }
+        } finally {
+            fclose($handle);
+        }
+    }
+
+    /**
+     * Validates the file by streaming through all records, keeping memory usage bounded.
+     */
+    private function validateStreamed(string $path): bool
+    {
+        try {
+            // iterate to let the streaming parser see the whole document
+            iterator_count($this->streamItems($path));
+
+            return true;
+        } catch (JsonMachineException $exception) {
+            $this->applicationLogger->error('Reading file ERROR: ' . $exception->getMessage(), [
+                'component' => PimcoreDataImporterBundle::LOGGER_COMPONENT_PREFIX . $this->configName
+            ]);
+
+            return false;
+        }
+    }
+
+    /**
+     * Streams up to the requested record and returns it together with the record number that
+     * was actually read (the last record when the requested one is out of range).
+     *
+     * @return array{0: ?array, 1: int}
+     */
+    private function readStreamedRecord(string $path, int $recordNumber): array
+    {
+        $currentRecordNumber = -1;
+        $currentRow = null;
+
+        foreach ($this->streamItems($path) as $row) {
+            $currentRow = $row;
+            $currentRecordNumber++;
+
+            if ($currentRecordNumber === $recordNumber && !empty($currentRow)) {
+                return [$currentRow, $currentRecordNumber];
+            }
+        }
+
+        return [$currentRow, max(0, $currentRecordNumber)];
+    }
+
+    private function skipByteOrderMark($handle): void
+    {
+        $bom = fread($handle, strlen(self::UTF8_BOM));
+        if (0 !== strncmp(self::UTF8_BOM, (string)$bom, strlen(self::UTF8_BOM))) {
+            rewind($handle);
+        }
+    }
 }