Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

PHPLIB-1617 Accept a Pipeline instance in aggregate and watch methods #1580

Merged
merged 1 commit into from
Feb 10, 2025
Merged

PHPLIB-1617 Accept a Pipeline instance in aggregate and watch methods #1580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions src/Collection.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,16 +209,16 @@ public function __toString(): string
* Executes an aggregation framework pipeline on the collection.
*
* @see Aggregate::__construct() for supported options
* @param array $pipeline Aggregation pipeline
* @param array $options Command options
* @param array|Pipeline $pipeline Aggregation pipeline
* @param array $options Command options
* @throws UnexpectedValueException if the command response was malformed
* @throws UnsupportedException if options are not supported by the selected server
* @throws InvalidArgumentException for parameter/option parsing errors
* @throws DriverRuntimeException for other driver errors (e.g. connection errors)
*/
public function aggregate(array $pipeline, array $options = []): CursorInterface
public function aggregate(array|Pipeline $pipeline, array $options = []): CursorInterface
{
if (is_builder_pipeline($pipeline)) {
if (is_array($pipeline) && is_builder_pipeline($pipeline)) {
$pipeline = new Pipeline(...$pipeline);
}

Expand Down Expand Up @@ -1014,13 +1014,13 @@ public function updateSearchIndex(string $name, array|object $definition, array
* Create a change stream for watching changes to the collection.
*
* @see Watch::__construct() for supported options
* @param array $pipeline Aggregation pipeline
* @param array $options Command options
* @param array|Pipeline $pipeline Aggregation pipeline
* @param array $options Command options
* @throws InvalidArgumentException for parameter/option parsing errors
*/
public function watch(array $pipeline = [], array $options = []): ChangeStream
public function watch(array|Pipeline $pipeline = [], array $options = []): ChangeStream
{
if (is_builder_pipeline($pipeline)) {
if (is_array($pipeline) && is_builder_pipeline($pipeline)) {
$pipeline = new Pipeline(...$pipeline);
}

Expand Down
16 changes: 8 additions & 8 deletions src/Database.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,16 +188,16 @@ public function __toString(): string
* and $listLocalSessions. Requires MongoDB >= 3.6
*
* @see Aggregate::__construct() for supported options
* @param array $pipeline Aggregation pipeline
* @param array $options Command options
* @param array|Pipeline $pipeline Aggregation pipeline
* @param array $options Command options
* @throws UnexpectedValueException if the command response was malformed
* @throws UnsupportedException if options are not supported by the selected server
* @throws InvalidArgumentException for parameter/option parsing errors
* @throws DriverRuntimeException for other driver errors (e.g. connection errors)
*/
public function aggregate(array $pipeline, array $options = []): CursorInterface
public function aggregate(array|Pipeline $pipeline, array $options = []): CursorInterface
{
if (is_builder_pipeline($pipeline)) {
if (is_array($pipeline) && is_builder_pipeline($pipeline)) {
$pipeline = new Pipeline(...$pipeline);
}

Expand Down Expand Up @@ -582,13 +582,13 @@ public function selectGridFSBucket(array $options = []): Bucket
* Create a change stream for watching changes to the database.
*
* @see Watch::__construct() for supported options
* @param array $pipeline Aggregation pipeline
* @param array $options Command options
* @param array|Pipeline $pipeline Aggregation pipeline
* @param array $options Command options
* @throws InvalidArgumentException for parameter/option parsing errors
*/
public function watch(array $pipeline = [], array $options = []): ChangeStream
public function watch(array|Pipeline $pipeline = [], array $options = []): ChangeStream
{
if (is_builder_pipeline($pipeline)) {
if (is_array($pipeline) && is_builder_pipeline($pipeline)) {
$pipeline = new Pipeline(...$pipeline);
}

Expand Down
21 changes: 15 additions & 6 deletions tests/Collection/BuilderCollectionFunctionalTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
use MongoDB\Builder\Pipeline;
use MongoDB\Builder\Query;
use MongoDB\Builder\Stage;
use PHPUnit\Framework\Attributes\TestWith;

use function iterator_to_array;

Expand All @@ -18,7 +19,9 @@ public function setUp(): void
$this->collection->insertMany([['x' => 1], ['x' => 2], ['x' => 2]]);
}

public function testAggregate(): void
#[TestWith([true])]
#[TestWith([false])]
public function testAggregate(bool $pipelineAsArray): void
{
$this->collection->insertMany([['x' => 10], ['x' => 10], ['x' => 10]]);
$pipeline = new Pipeline(
Expand All @@ -27,8 +30,10 @@ public function testAggregate(): void
buckets: 2,
),
);
// Extract the list of stages for arg type restriction
$pipeline = iterator_to_array($pipeline);

if ($pipelineAsArray) {
$pipeline = iterator_to_array($pipeline);
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't disagree that this tests the underlying behavior, but it seems a bit backwards that this operates by converting the Pipeline back into an array, instead of creating a list of stages and then conditionally converting that to a pipeline.

The latter approach seems more straightforward.

Copy link
Member Author

@GromNaN GromNaN Feb 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right, I hadn't thought of it like that. #1596


$results = $this->collection->aggregate($pipeline)->toArray();
$this->assertCount(2, $results);
Expand Down Expand Up @@ -257,7 +262,9 @@ public function testUpdateManyWithPipeline(): void
$this->assertEquals(3, $result[0]->x);
}

public function testWatch(): void
#[TestWith([true])]
#[TestWith([false])]
public function testWatch(bool $pipelineAsArray): void
{
$this->skipIfChangeStreamIsNotSupported();

Expand All @@ -268,8 +275,10 @@ public function testWatch(): void
$pipeline = new Pipeline(
Stage::match(operationType: Query::eq('insert')),
);
// Extract the list of stages for arg type restriction
$pipeline = iterator_to_array($pipeline);

if ($pipelineAsArray) {
$pipeline = iterator_to_array($pipeline);
}

$changeStream = $this->collection->watch($pipeline);
$this->collection->insertOne(['x' => 3]);
Expand Down
21 changes: 15 additions & 6 deletions tests/Database/BuilderDatabaseFunctionalTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
use MongoDB\Builder\Pipeline;
use MongoDB\Builder\Query;
use MongoDB\Builder\Stage;
use PHPUnit\Framework\Attributes\TestWith;

use function iterator_to_array;

Expand All @@ -18,7 +19,9 @@ public function tearDown(): void
parent::tearDown();
}

public function testAggregate(): void
#[TestWith([true])]
#[TestWith([false])]
public function testAggregate(bool $pipelineAsArray): void
{
$this->skipIfServerVersion('<', '6.0.0', '$documents stage is not supported');

Expand All @@ -33,14 +36,18 @@ public function testAggregate(): void
buckets: 2,
),
);
// Extract the list of stages for arg type restriction
$pipeline = iterator_to_array($pipeline);

if ($pipelineAsArray) {
$pipeline = iterator_to_array($pipeline);
}

$results = $this->database->aggregate($pipeline)->toArray();
$this->assertCount(2, $results);
}

public function testWatch(): void
#[TestWith([true])]
#[TestWith([false])]
public function testWatch(bool $pipelineAsArray): void
{
$this->skipIfChangeStreamIsNotSupported();

Expand All @@ -51,8 +58,10 @@ public function testWatch(): void
$pipeline = new Pipeline(
Stage::match(operationType: Query::eq('insert')),
);
// Extract the list of stages for arg type restriction
$pipeline = iterator_to_array($pipeline);

if ($pipelineAsArray) {
$pipeline = iterator_to_array($pipeline);
}

$changeStream = $this->database->watch($pipeline);
$this->database->selectCollection($this->getCollectionName())->insertOne(['x' => 3]);
Expand Down