Added ability to iterate over large queries without exceeding memory

README.md

@@ -46,7 +46,7 @@ $client = $database->getClient();
 
 **Important:** don't forget to `urlencode()` the password (and username for that matter) when using a DSN,
 especially if it contains non-alphanumeric characters. Not doing so might cause exceptions to be thrown.
-  
+
 ### Reading data
 
 To fetch records from InfluxDB you can do a query directly on a database:
@@ -103,6 +103,16 @@ In production if you are querying InfluxDB to generate a response to a web or AP
 $database = InfluxDB\Client::fromDSN(sprintf('influxdb://user:pass@%s:%s/%s', $host, $port, $dbname), 5);
 ```
 
+### Reading large data sets
+
+Reading large data sets may exceed available memory. To overcome this obstacle, use the iterate() method to return each point without loading them all first into memory.  This method supports only a single measurement, however, you can manually target another measurement if necessary.
+
+```php
+foreach($result->iterate() as $index=>$point) {
+    // Process each $point as needed.
+}
+```
+
 ### Writing data
 
 Writing data is done by providing an array of points to the writePoints method on a database:
@@ -280,6 +290,27 @@ $result = $client->listUsers();
 $result = $client->listDatabases();
 ```
 
+### Other Database functions
+
+Several additional database functions are available:
+
+```php
+// list measurements
+$result = $database->listMeasurements();
+
+// list all field keys
+$result = $database->listFieldKeys();
+
+// list field keys for a single measurement
+$result = $database->listFieldKeys('yourMeasurement');
+
+// list all tag keys
+$result = $database->listTagKeys();
+
+// list tag keys for a single measurement
+$result = $database->listTagKeys('yourMeasurement');
+```
+
 ### Admin functionality
 
 You can use the client's $client->admin functionality to administer InfluxDB via the API.
@@ -371,10 +402,10 @@ $client->admin->revoke(\InfluxDB\Client\Admin::PRIVILEGE_ALL, 'admin_user');
 * Fixed tag with Boolean/Null value trigger parse error
 
 #### 1.4.1
-* Fixed bug: Escape field values as per line protocol. 
+* Fixed bug: Escape field values as per line protocol.
 
 #### 1.4.0
-* Updating Influx Database with support for writing direct payloads, thanks @virgofx 
+* Updating Influx Database with support for writing direct payloads, thanks @virgofx
 
 #### 1.3.1
 * Added ability to write data to a specific retention policy, thanks @virgofx !

composer.json

@@ -27,6 +27,7 @@
     ],
     "require": {
         "php": "^5.5 || ^7.0",
+        "halaxa/json-machine": "0.3.2",
         "guzzlehttp/guzzle": "^6.0"
     },
     "require-dev": {

src/InfluxDB/Client.php

@@ -349,6 +349,7 @@ public function getVerifySSL()
     public function setDriver(DriverInterface $driver)
     {
         $this->driver = $driver;
+        return $this;
     }
 
     /**

src/InfluxDB/Database.php

@@ -8,6 +8,7 @@
 use InfluxDB\Database\Exception as DatabaseException;
 use InfluxDB\Database\RetentionPolicy;
 use InfluxDB\Query\Builder as QueryBuilder;
+use InfluxDB\Query\QueryBuilder as QueryBuilder2;
 
 /**
  * Class Database
@@ -205,6 +206,63 @@ public function listRetentionPolicies()
         return $this->query(sprintf('SHOW RETENTION POLICIES ON "%s"', $this->name))->getPoints();
     }
 
+    /**
+     * @return array
+     */
+    public function listMeasurements():array
+    {
+        return array_map(function($m){
+            return $m[0];
+        }, $this->query('SHOW MEASUREMENTS')->getSeries()[0]['values']);
+    }
+
+    /**
+     * @param $measurement
+     * @return array
+     * @throws Exception
+     */
+    public function listFieldKeys(?string $measurement=null):array
+    {
+        $rs = $measurement
+        ?$this->query(sprintf('SHOW FIELD KEYS FROM "%s"', $measurement))->getSeries()
+        :$this->query('SHOW FIELD KEYS')->getSeries();
+        $arr=[];
+        foreach($rs as $m) {
+            $arr[$m['name']]=[];
+            foreach($m['values'] as $v) {
+                $arr[$m['name']][$v[0]]=$v[1];
+            }
+        }
+        if($measurement) {
+            if(!isset($arr[$measurement])) throw new Exception("Measurement $measurement does not exist.");
+            return $arr[$measurement];
+        }
+        else return $arr;
+    }
+
+    /**
+     * @param $measurement
+     * @return array
+     * @throws Exception
+     */
+    public function listTagKeys(?string $measurement=null):array
+    {
+        $rs = $measurement
+        ?$this->query(sprintf('SHOW TAG KEYS FROM "%s"', $measurement))->getSeries()
+        :$this->query('SHOW TAG KEYS')->getSeries();
+        $arr=[];
+        foreach($rs as $m) {
+            $arr[$m['name']]=array_map(function($v){
+                return $v[0];
+                }, $m['values']);
+        }
+        if($measurement) {
+            if(!isset($arr[$measurement])) throw new Exception("Measurement $measurement does not exist.");
+            return $arr[$measurement];
+        }
+        else return $arr;
+    }
+
     /**
      * Drop this database
      */
@@ -223,6 +281,16 @@ public function getQueryBuilder()
         return new QueryBuilder($this);
     }
 
+    /**
+     * Retrieve the optional second query builder
+     *
+     * @return QueryBuilder
+     */
+    public function getQueryBuilder2()
+    {
+        return new QueryBuilder2($this);
+    }
+
     /**
      * @return Client
      */

src/InfluxDB/Driver/Curl.php

@@ -136,34 +136,38 @@ public function isSuccess()
      */
     public function query()
     {
-        $response = $this->execute($this->parameters['url'], $this->getCurlOptions());
-        return new ResultSet($response);
+        $stream=fopen('php://temp', 'w+');
+        $this->execute($this->parameters['url'], $this->getCurlOptions() + [CURLOPT_FILE => $stream]);
+        rewind($stream);
+        return new ResultSet($stream);
     }
 
     protected function execute($url, $curlOptions = [])
     {
         $this->lastRequestInfo = null;
         $ch = curl_init();
 
-        foreach ($curlOptions as $option => $value) {
-            curl_setopt($ch, $option, $value);
-        }
-
-        curl_setopt($ch, CURLOPT_URL, $this->dsn . '/' . $url);
-        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
+        $curlOptions=[
+            CURLOPT_URL => $this->dsn . '/' . $url,
+            CURLOPT_RETURNTRANSFER => true, /* CURLOPT_FILE must be set after CURLOPT_RETURNTRANSFER */
+            CURLOPT_HEADER => 0,
+            CURLOPT_BUFFERSIZE => 256,
+        ] + $curlOptions;
 
-        $result = curl_exec($ch);
-        $this->lastRequestInfo = curl_getinfo($ch);
+        curl_setopt_array( $ch, $curlOptions);
 
-        if ($result === false) {
+        if(isset($curlOptions[CURLOPT_FILE])) {
+            curl_exec($ch);
+            rewind($curlOptions[CURLOPT_FILE]);
+        }
+        elseif (curl_exec($ch) === false) {
             // in case of total failure - socket/port is closed etc
             throw new Exception('Request failed! curl_errno: ' . curl_errno($ch));
         }
 
+        $this->lastRequestInfo = curl_getinfo($ch);
 
         curl_close($ch);
-
-        return $result;
     }
 
     /**

src/InfluxDB/Driver/Guzzle.php

@@ -7,6 +7,7 @@
 
 use GuzzleHttp\Client;
 use GuzzleHttp\Psr7\Response;
+use GuzzleHttp\Psr7\StreamWrapper;
 use InfluxDB\ResultSet;
 
 /**
@@ -86,25 +87,13 @@ public function write($data = null)
 
     /**
      * @throws \Exception
-     * @return ResultSet
+     * @return stream
      */
     public function query()
     {
         $response = $this->httpClient->get($this->parameters['url'], $this->getRequestParameters());
 
-        $raw = (string) $response->getBody();
-
-        return $this->asResultSet($raw);
-    }
-
-    /**
-     * @param $raw
-     * @return ResultSet
-     * @throws \InfluxDB\Client\Exception
-     */
-    protected function asResultSet($raw)
-    {
-        return new ResultSet($raw);
+        return new ResultSet(StreamWrapper::getResource($response->getBody()));
     }
 
     /**

src/InfluxDB/Driver/QueryDriverInterface.php

@@ -16,7 +16,8 @@ interface QueryDriverInterface
 {
 
     /**
-     * @return ResultSet
+     * @throws \Exception
+     * @return stream
      */
     public function query();
 }