fix: cleanup after streaming in sap hana

Author: alumni

src/driver/sap/SapQueryRunner.ts

@@ -336,20 +336,54 @@ export class SapQueryRunner extends BaseQueryRunner implements QueryRunner {
     ): Promise<ReadStream> {
         if (this.isReleased) throw new QueryRunnerAlreadyReleasedError()
 
-        const databaseConnection = await this.connect()
-        this.driver.connection.logger.logQuery(query, parameters, this)
+        const release = await this.lock.acquire()
+        let dropStatement: () => Promise<void>
 
-        const prepareAsync = promisify(databaseConnection.prepare).bind(
-            databaseConnection,
-        )
-        const statement = await prepareAsync(query)
-        const resultSet = statement.executeQuery(parameters)
-        const stream = this.driver.streamClient.createObjectStream(resultSet)
+        try {
+            const databaseConnection = await this.connect()
+            this.driver.connection.logger.logQuery(query, parameters, this)
 
-        if (onEnd) stream.on("end", onEnd)
-        if (onError) stream.on("error", onError)
+            const statement = await promisify(databaseConnection.prepare).call(
+                databaseConnection,
+                query,
+            )
+            const resultSet = await promisify(statement.executeQuery).call(
+                statement,
+                parameters,
+            )
+            const stream =
+                this.driver.streamClient.createObjectStream(resultSet)
+            dropStatement = promisify(statement.drop).bind(statement)
+
+            stream.on("end", async () => {
+                await dropStatement()
+                release()
+                onEnd?.()
+            })
+            stream.on("error", async (error: Error) => {
+                this.driver.connection.logger.logQueryError(
+                    error,
+                    query,
+                    parameters,
+                    this,
+                )
+                await dropStatement()
+                release()
+                onError?.(error)
+            })
 
-        return stream
+            return stream
+        } catch (error) {
+            this.driver.connection.logger.logQueryError(
+                error,
+                query,
+                parameters,
+                this,
+            )
+            await dropStatement?.()
+            release()
+            throw new QueryFailedError(query, parameters, error)
+        }
     }
 
     /**