fix: avoid early release of PostgresQueryRunner (#7109) (#7185)

* fix: avoid early release of PostgresQueryRunner (#7109)

If the QueryRunner is not connected yet, release is a no-op

* fix: avoid early release of PostgresQueryRunner (#7109)

Revert no-op driver-wide early release, fix only for stream case.

* fix: avoid early release of PostgresQueryRunner (#7109)

Restrict stream unit test to drivers who implement the stream interface

* fix: avoid early release of PostgresQueryRunner (#7109)

Restrict unit test to drivers whose stream implementation works in master

package.json

@@ -101,6 +101,7 @@
     "mysql2": "^2.1.0",
     "oracledb": "^5.0.0",
     "pg": "^8.3.0",
+    "pg-query-stream": "^4.0.0",
     "redis": "^3.0.2",
     "remap-istanbul": "^0.13.0",
     "rimraf": "^3.0.2",

src/query-builder/SelectQueryBuilder.ts

@@ -1261,9 +1261,6 @@ export class SelectQueryBuilder<Entity> extends QueryBuilder<Entity> implements
             }
             throw error;
 
-        } finally {
-            if (queryRunner !== this.queryRunner) // means we created our own query runner
-                await queryRunner.release();
         }
     }
 

test/github-issues/7109/entity/Dummy.ts

@@ -0,0 +1,13 @@
+import {Entity, PrimaryGeneratedColumn} from "../../../../src";
+import {Column} from "../../../../src/decorator/columns/Column";
+
+@Entity()
+export class Dummy {
+
+    @PrimaryGeneratedColumn()
+    id: number;
+
+    @Column()
+    field: string;
+
+}

test/github-issues/7109/issue-7109.ts

@@ -0,0 +1,46 @@
+import "reflect-metadata";
+import {createTestingConnections, closeTestingConnections, reloadTestingDatabases} from "../../utils/test-utils";
+import {Connection} from "../../../src/connection/Connection";
+import {Dummy} from './entity/Dummy';
+import {ReadStream} from 'fs';
+import {expect} from "chai";
+
+function ingestStream (stream: ReadStream): Promise<any[]> {
+    let chunks: any[] = [];
+    return new Promise((resolve, reject) => {
+      stream.on('data', chunk => chunks.push(chunk))
+      stream.on('error', reject)
+      stream.on('end', () => resolve(chunks))
+    })
+  }
+
+describe("github issues > #7109 stream() bug from 0.2.25 to 0.2.26 with postgresql", () => {
+    let connections: Connection[];
+    before(async () => connections = await createTestingConnections({
+        entities: [__dirname + "/entity/*{.js,.ts}"],
+        schemaCreate: true,
+        dropSchema: true,
+        enabledDrivers: ["postgres", "mysql", "mariadb", "cockroachdb"]
+    }));
+    beforeEach(() => reloadTestingDatabases(connections));
+    after(() => closeTestingConnections(connections));
+
+    it("should release the QueryRunner created by a SelectQueryBuilder", () => Promise.all(connections.map(async connection => {
+        const values = [{field: "abc"}, {field: "def"}, {field: "ghi"}];
+        // First create some test data
+        await connection.createQueryBuilder()
+            .insert()
+            .into(Dummy)
+            .values(values)
+            .execute();
+
+       // Stream data:
+       const stream = await connection.createQueryBuilder().from(Dummy, 'dummy').select('field').stream();
+       const streamedEntities = await ingestStream(stream);
+
+       // If the runner is properly released, the test is already successful; this assert is just a sanity check.
+       const extractFields = (val: {field: string}) => val.field;
+       expect(streamedEntities.map(extractFields)).to.have.members(values.map(extractFields));
+
+    })));
+});