cursor() returns a LazyCollection, and survives a fresh connection

Both Query\Builder::cursor() and Eloquent\Builder::cursor() were generator
functions handing back a bare Generator, where Laravel's own builders return
a LazyCollection. Anything chaining off the framework contract broke, most
visibly cursor()->chunk(), which fataled with "Call to undefined method
Generator::chunk()".

getRawResponse() also dereferenced a null raw response. Only select, insert
and aggregate processing assign one and a scroll runs none of them, so a
cursor that was the first query on its connection died on asArray(). Every
existing cursor test inserted first, which is why nothing caught it.

Ported from pdphilip/opensearch v3.2.0. The scroll rewrite from that release
is not needed here: the Elasticsearch client already sends scroll_id in the
request body for both scroll and clearScroll, so the OpenSearch 3.x
"Cannot parse scroll id" failure has no equivalent.

Five tests cover it, each verified to fail without the fix: chunking off a
cursor, a cursor on a connection that has not run a query, laziness (three
hydrations out of ten rows, so an eager collect() fails loudly), and
re-iteration. The existing "returns lazy collection via cursor" test was
asserting Generator::class, which is now true to its name.

Author: pdphilip

src/Eloquent/Builder.php

@@ -13,7 +13,7 @@
 use Illuminate\Pagination\CursorPaginator;
 use Illuminate\Pagination\Paginator;
 use Illuminate\Support\Collection;
-use Iterator;
+use Illuminate\Support\LazyCollection;
 use PDPhilip\Elasticsearch\Data\MetaDTO;
 use PDPhilip\Elasticsearch\Exceptions\BuilderException;
 use PDPhilip\Elasticsearch\Exceptions\DynamicIndexException;
@@ -378,15 +378,17 @@ protected function searchAfterPaginator($items, $perPage, $cursor, $options)
     }
 
     /**
-     * Get a generator for the given query.
+     * Get a lazy collection for the given query.
      *
-     * @return Iterator
+     * @return LazyCollection
      */
     public function cursor($scrollTimeout = '30s')
     {
-        foreach ($this->applyScopes()->query->cursor($scrollTimeout) as $record) {
-            yield $this->model->newFromBuilder($record);
-        }
+        return new LazyCollection(function () use ($scrollTimeout) {
+            foreach ($this->applyScopes()->query->cursor($scrollTimeout) as $record) {
+                yield $this->model->newFromBuilder($record);
+            }
+        });
     }
 
     /**

src/Query/Builder.php

@@ -8,11 +8,11 @@
 use DateTimeInterface;
 use Elastic\Elasticsearch\Response\Elasticsearch;
 use Exception;
-use Generator;
 use Illuminate\Contracts\Support\Arrayable;
 use Illuminate\Database\Query\Builder as BaseBuilder;
 use Illuminate\Database\Query\Expression;
 use Illuminate\Support\Arr;
+use Illuminate\Support\LazyCollection;
 use Illuminate\Support\Str;
 use InvalidArgumentException;
 use PDPhilip\Elasticsearch\Connection;
@@ -676,19 +676,21 @@ protected function runPaginationCountQuery($columns = ['_id'])
     }
 
     /**
-     * Get a generator for the given query.
+     * Get a lazy collection for the given query.
      *
-     * @return Generator
+     * @return LazyCollection
      */
     public function cursor($scrollTimeout = '30s')
     {
-        if (is_null($this->columns)) {
-            $this->columns = ['*'];
-        }
+        return new LazyCollection(function () {
+            if (is_null($this->columns)) {
+                $this->columns = ['*'];
+            }
 
-        foreach ($this->connection->cursor($this->toCompiledQuery()) as $document) {
-            yield $this->processor->documentFromResult($this, $document);
-        }
+            foreach ($this->connection->cursor($this->toCompiledQuery()) as $document) {
+                yield $this->processor->documentFromResult($this, $document);
+            }
+        });
     }
 
     /**

src/Query/Processor/Processor.php

@@ -44,9 +44,17 @@ public function getRawAggregationResults(): array
 
     /**
      * Get the raw Elasticsearch response as an array
+     *
+     * Null is legitimate: only select, insert and aggregate processing assign
+     * a raw response, and the scroll behind cursor() runs none of them. Per-hit
+     * meta reaches MetaDTO through documentFromResult's extras either way.
      */
     public function getRawResponse(): array
     {
+        if ($this->rawResponse === null) {
+            return [];
+        }
+
         return is_array($this->rawResponse) ? $this->rawResponse : $this->rawResponse->asArray();
     }
 

tests/ModelTest.php

@@ -5,6 +5,8 @@
 use Carbon\Carbon;
 use Illuminate\Database\Eloquent\Collection as EloquentCollection;
 use Illuminate\Database\Eloquent\ModelNotFoundException;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\LazyCollection;
 use Illuminate\Support\Str;
 use PDPhilip\Elasticsearch\Connection;
 use PDPhilip\Elasticsearch\Data\ModelMeta;
@@ -578,6 +580,75 @@
     expect($names)->toHaveCount(15000);
 });
 
+it('returns a chunkable lazy collection from cursor', function () {
+    User::insert([
+        ['name' => 'User 1'],
+        ['name' => 'User 2'],
+        ['name' => 'User 3'],
+    ]);
+
+    $cursor = User::query()->orderBy('name.keyword')->cursor();
+
+    expect($cursor)->toBeInstanceOf(LazyCollection::class);
+
+    $chunkSizes = $cursor->chunk(2)->map(fn ($chunk) => $chunk->count())->all();
+    expect($chunkSizes)->toBe([2, 1]);
+});
+
+it('cursors on a connection that has not run a query yet', function () {
+    User::insert([
+        ['name' => 'User 1'],
+        ['name' => 'User 2'],
+        ['name' => 'User 3'],
+    ]);
+
+    // A scroll populates no raw response, so a cursor that is the first query
+    // on its connection has no earlier select or insert to inherit one from.
+    DB::purge('elasticsearch');
+
+    $names = User::query()->orderBy('name.keyword')->cursor()->pluck('name')->all();
+
+    expect($names)->toBe(['User 1', 'User 2', 'User 3']);
+});
+
+it('keeps the cursor lazy, hydrating only what is consumed', function () {
+    User::insert(array_map(fn ($i) => ['name' => "User {$i}"], range(1, 10)));
+
+    $hydrated = 0;
+    User::retrieved(function () use (&$hydrated) {
+        $hydrated++;
+    });
+
+    $cursor = User::query()->orderBy('name.keyword')->cursor();
+
+    // Building the collection must not touch the cluster.
+    expect($hydrated)->toBe(0);
+
+    $taken = $cursor->take(3)->all();
+
+    // Three, not ten - an eagerly materialised collection would fail here while
+    // every other assertion about cursor() carried on passing.
+    expect($taken)->toHaveCount(3)
+        ->and($hydrated)->toBe(3);
+});
+
+it('re-runs the scroll when the cursor is iterated twice', function () {
+    User::insert([
+        ['name' => 'User 1'],
+        ['name' => 'User 2'],
+        ['name' => 'User 3'],
+    ]);
+
+    $cursor = User::query()->orderBy('name.keyword')->cursor();
+
+    // A bare generator throws "Cannot rewind a generator" on the second pass.
+    $first = $cursor->pluck('name')->all();
+    $second = $cursor->pluck('name')->all();
+
+    expect($first)->toBe(['User 1', 'User 2', 'User 3'])
+        ->and($second)->toBe($first);
+});
+
 it('tests truncate model', function () {
     User::create(['name' => 'John Doe']);
 

tests/QueryBuilderTest.php

@@ -4,6 +4,7 @@
 
 use Carbon\Carbon;
 use Illuminate\Support\Facades\Date;
+use Illuminate\Support\LazyCollection;
 use PDPhilip\Elasticsearch\Eloquent\Builder;
 use PDPhilip\Elasticsearch\Tests\Models\Item;
 use PDPhilip\Elasticsearch\Tests\Models\User;
@@ -596,12 +597,27 @@
 
     $results = Item::orderBy('name.keyword', 'asc')->cursor();
 
-    expect($results)->toBeInstanceOf(Generator::class);
+    expect($results)->toBeInstanceOf(LazyCollection::class);
     foreach ($results as $i => $result) {
         expect($result->name)->toBe($data[$i]['name']);
     }
 });
 
+it('chunks a cursor directly, without iterating it', function () {
+    Item::insert([
+        ['name' => 'fork'],
+        ['name' => 'spoon'],
+        ['name' => 'spork'],
+    ]);
+
+    // Chained straight off cursor() rather than iterated - the call shape that
+    // failed with "Call to undefined method Generator::chunk()".
+    $chunks = Item::orderBy('name.keyword', 'asc')->cursor()->chunk(2);
+
+    expect($chunks->map->count()->all())->toBe([2, 1])
+        ->and($chunks->first()->pluck('name')->all())->toBe(['fork', 'spoon']);
+});
+
 it('increments multiple fields simultaneously', function () {
     User::insert([
         ['name' => 'John Doe', 'age' => 30, 'note' => 5],