Merge remote-tracking branch 'origin/master'

# Conflicts:
#	src/Transport/ClickhouseCLIClientTransport.php

Author: FacedSID

src/Client.php

@@ -406,6 +406,35 @@ public function insertFiles(string $table, array $columns, array $files, string
         return $this->getTransport()->sendAsyncFilesWithQuery($this->getServer(), $query, $files, $concurrency);
     }
 
+    /**
+     * Performs insert query using all given files as one block of data
+     *
+     * @param string      $table
+     * @param array       $columns
+     * @param array       $files
+     * @param string|null $format
+     * @param array       $options
+     *
+     * @return mixed
+     * @throws ClientException
+     */
+    public function insertFilesAsOne(string $table, array $columns, array $files, string $format = null, $options = [])
+    {
+        if (is_null($format)) {
+            $format = Format::CSV;
+        }
+
+        $query = 'INSERT INTO '.$table.' ('.implode(', ', $columns).') FORMAT '.strtoupper($format);
+
+        foreach ($files as $file) {
+            if (!is_file($file)) {
+                throw ClientException::insertFileNotFound($file);
+            }
+        }
+
+        return $this->getTransport()->sendFilesAsOneWithQuery($this->getServer(), $query, $files, $options);
+    }
+
     /**
      * Executes query.
      *

src/Exceptions/ClientException.php

@@ -37,4 +37,9 @@ public static function insertFileNotFound($file)
     {
         return new static('File '.$file.' is not found');
     }
+
+    public static function notSupported($operation)
+    {
+        return new static('Operation '.$operation.' is not supported');
+    }
 }

src/Interfaces/TransportInterface.php

@@ -22,7 +22,7 @@ interface TransportInterface
      * @return bool
      */
     public function send(Server $server, string $query): bool;
-
+    
     /**
      * Sends async insert queries with given files.
      *
@@ -34,7 +34,19 @@ public function send(Server $server, string $query): bool;
      * @return array
      */
     public function sendAsyncFilesWithQuery(Server $server, string $query, array $files, int $concurrency = 5): array;
-
+    
+    /**
+     * Sends files as one block of data.
+     *
+     * @param \Tinderbox\Clickhouse\Server $server
+     * @param string                       $query
+     * @param array                        $files
+     * @param array                        $settings
+     *
+     * @return bool
+     */
+    public function sendFilesAsOneWithQuery(Server $server, string $query, array $files, array $settings = []): bool;
+    
     /**
      * Executes SELECT queries and returns result.
      *
@@ -47,7 +59,7 @@ public function sendAsyncFilesWithQuery(Server $server, string $query, array $fi
      * @return \Tinderbox\Clickhouse\Query\Result
      */
     public function get(Server $server, string $query, $tables = null): Result;
-
+    
     /**
      * Executes async SELECT queries and returns result.
      *

src/Transport/ClickhouseCLIClientTransport.php

@@ -21,21 +21,39 @@ class ClickhouseCLIClientTransport implements TransportInterface
      */
     protected $executablePath;
 
+    /**
+     * Path to executable cat command
+     *
+     * @var string
+     */
+    protected $catExecutablePath;
+
     /**
      * Last execute query
      *
      * @var string
      */
     protected $lastQuery = '';
 
+    /**
+     * Use special ccat command to cat files into one without ARG_MAX limits
+     *
+     * @var bool
+     */
+    protected $useCcat;
+
     /**
      * ClickhouseCLIClientTransport constructor.
      *
      * @param string|null $executablePath
+     * @param string|null $catExecutablePath
+     * @param bool        $useCcat
      */
-    public function __construct(string $executablePath = null)
+    public function __construct(string $executablePath = null, string $catExecutablePath = null, bool $useCcat = false)
     {
         $this->setExecutablePath($executablePath);
+        $this->setCatExecutablePath($catExecutablePath);
+        $this->useCcat = $useCcat;
     }
 
     /**
@@ -52,6 +70,20 @@ protected function setExecutablePath(string $executablePath = null)
         $this->executablePath = $executablePath;
     }
 
+    /**
+     * Set path to cat executable.
+     *
+     * @param string|null $catExecutablePath
+     */
+    protected function setCatExecutablePath(string $catExecutablePath = null)
+    {
+        if (is_null($catExecutablePath)) {
+            $catExecutablePath = 'cat';
+        }
+
+        $this->catExecutablePath = $catExecutablePath;
+    }
+
     /**
      * Sends query to given $server.
      *
@@ -120,15 +152,27 @@ public function sendFilesAsOneWithQuery(Server $server, string $query, array $fi
          * (IFS=$'\n'; cat $(< tmp-file))
          *
          * to iterate through file and cat each file into stdout
+         *
+         * ccat is used to ignore ARG_MAX constant when trying to push too many files
          */
-        $fileName = $this->writeTemporaryFile(implode(PHP_EOL, array_map(function(string $file) {
-            return 'cat '.$file;
-        }, $files)));
+        if ($this->useCcat) {
+            $files = implode(PHP_EOL, $files);
+        } else {
+            $files = implode(PHP_EOL, array_map(function(string $file) {
+                return 'cat '.$file;
+            }, $files));
+        }
+
+        $fileName = $this->writeTemporaryFile($files);
 
         $command = $this->buildCommandForWriteFilesAsOne($server, $query, $fileName, $settings);
 
         try {
             $this->executeCommand($command);
+
+            if (!is_null($fileName)) {
+                $this->removeTemporaryFile($fileName);
+            }
         } catch (\Throwable $e) {
             $this->removeTemporaryFile($fileName);
 
@@ -243,7 +287,7 @@ protected function buildCommandForWrite(Server $server, string $query, string $f
         $command = [];
 
         if (!is_null($file)) {
-            $command[] = "$(which cat) ".$file.' |';
+            $command[] = $this->catExecutablePath.' '.$file.' |';
         }
 
         $command = array_merge($command, [
@@ -275,14 +319,21 @@ protected function buildCommandForWriteFilesAsOne(Server $server, string $query,
     {
         $query = escapeshellarg($query);
 
-        $command = [
-            '$(cat '.$file.') | ',
+        $command = [];
+
+        if ($this->useCcat) {
+            $command[] = $this->catExecutablePath.' '.$file.' | ';
+        } else {
+            $command[] = '$('.$this->catExecutablePath.' '.$file.') | ';
+        }
+
+        $command = array_merge($command, [
             $this->executablePath,
             "--host='{$server->getHost()}'",
             "--port='{$server->getPort()}'",
             "--database='{$server->getDatabase()}'",
             "--query={$query}",
-        ];
+        ]);
 
         foreach ($settings as $key => $value) {
             $command[] = '--'.$key.'='.escapeshellarg($value);
@@ -306,7 +357,7 @@ protected function buildCommandForRead(Server $server, string $query, $tables =
         $fileName = $this->writeTemporaryFile($query);
 
         $command = [
-            "cat {$fileName} |",
+            $this->catExecutablePath." {$fileName} |",
             $this->executablePath,
             "--host='{$server->getHost()}'",
             "--port='{$server->getPort()}'",

src/Transport/HttpTransport.php

@@ -429,4 +429,21 @@ protected function buildRequestUri(Server $server, array $query = []): string
 
         return $uri.'?'.http_build_query($query);
     }
+
+    /**
+     * Sends files as one block of data.
+     *
+     * NOT SUPPORTED VIA HTTP INTERFACE
+     *
+     * @param \Tinderbox\Clickhouse\Server $server
+     * @param string                       $query
+     * @param array                        $files
+     * @param array                        $settings
+     *
+     * @return bool
+     */
+    public function sendFilesAsOneWithQuery(Server $server, string $query, array $files, array $settings = []): bool
+    {
+        throw ClientException::notSupported('inserting multiple files as one via http interface');
+    }
 }