fix: Update sqlExecute method to return structured result maps

- Modified the `sqlExecute` method to return a map containing success status, affected rows, and error messages for both DDL and DML operations.
- Updated documentation to reflect the new return structure.
- Adjusted unit tests to validate the new response format and ensure proper handling of success and error cases.

Signed-off-by: Edmund Miller <[email protected]>

Author: edmundmiller

plugins/nf-sqldb/src/main/nextflow/sql/ChannelSqlExtension.groovy

@@ -145,21 +145,21 @@ class ChannelSqlExtension extends PluginExtensionPoint {
 
     /**
      * Execute a SQL statement that does not return a result set (DDL/DML statements)
-     * For DML statements (INSERT, UPDATE, DELETE), it returns the number of affected rows
-     * For DDL statements (CREATE, ALTER, DROP), it returns null
+     * For DML statements (INSERT, UPDATE, DELETE), it returns a result map with success status and number of affected rows
+     * For DDL statements (CREATE, ALTER, DROP), it returns a result map with success status
      *
      * @param params A map containing 'db' (database alias) and 'statement' (SQL string to execute)
-     * @return The number of rows affected by the SQL statement for DML operations, null for DDL operations
+     * @return A map containing 'success' (boolean), 'result' (rows affected or null) and optionally 'error' (message)
      */
     @Function
-    Integer sqlExecute(Map params) {
+    Map sqlExecute(Map params) {
         CheckHelper.checkParams('sqlExecute', params, EXECUTE_PARAMS)
 
         final String dbName = params.db as String ?: 'default'
         final String statement = params.statement as String
 
         if (!statement)
-            throw new IllegalArgumentException("Missing required parameter 'statement'")
+            return [success: false, error: "Missing required parameter 'statement'"]
 
         final sqlConfig = new SqlConfig((Map) session.config.navigate('sql.db'))
         final SqlDataSource dataSource = sqlConfig.getDataSource(dbName)
@@ -171,28 +171,27 @@ class ChannelSqlExtension extends PluginExtensionPoint {
                 msg += " - Did you mean: ${choices.get(0)}?"
             else if (choices)
                 msg += " - Did you mean any of these?\n" + choices.collect { "  $it" }.join('\n') + '\n'
-            throw new IllegalArgumentException(msg)
+            return [success: false, error: msg]
         }
 
         try (Connection conn = groovy.sql.Sql.newInstance(dataSource.toMap()).getConnection()) {
             try (Statement stm = conn.createStatement()) {
                 String normalizedStatement = normalizeStatement(statement)
 
-                // For DDL statements (CREATE, ALTER, DROP, etc.), execute() returns true if the first result is a ResultSet
-                // For DML statements (INSERT, UPDATE, DELETE), executeUpdate() returns the number of rows affected
                 boolean isDDL = normalizedStatement.trim().toLowerCase().matches("^(create|alter|drop|truncate).*")
 
                 if (isDDL) {
                     stm.execute(normalizedStatement)
-                    return null
+                    return [success: true, result: null]
                 } else {
-                    return stm.executeUpdate(normalizedStatement)
+                    Integer rowsAffected = stm.executeUpdate(normalizedStatement)
+                    return [success: true, result: rowsAffected]
                 }
             }
         }
         catch (Exception e) {
             log.error("Error executing SQL statement: ${e.message}", e)
-            throw e
+            return [success: false, error: e.message]
         }
     }
 
@@ -204,7 +203,7 @@ class ChannelSqlExtension extends PluginExtensionPoint {
      */
     private static String normalizeStatement(String statement) {
         if (!statement)
-            throw new IllegalArgumentException("Missing SQL statement")
+            return null
         def result = statement.trim()
         if (!result.endsWith(';'))
             result += ';'

plugins/nf-sqldb/src/test/nextflow/sql/SqlExecutionTest.groovy

@@ -40,7 +40,7 @@ class SqlExecutionTest extends Specification {
         Global.session = null
     }
 
-    def 'should execute DDL statements successfully and return null'() {
+    def 'should execute DDL statements successfully and return success map'() {
         given:
         def JDBC_URL = 'jdbc:h2:mem:test_ddl_' + Random.newInstance().nextInt(1_000_000)
         def sql = Sql.newInstance(JDBC_URL, 'sa', null)
@@ -58,29 +58,32 @@ class SqlExecutionTest extends Specification {
             statement: 'CREATE TABLE test_table(id INT PRIMARY KEY, name VARCHAR(255))'
         ])
 
-        then: 'Table should be created and result should be null'
+        then: 'Table should be created and result should indicate success'
         sql.rows('SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = \'TEST_TABLE\'').size() > 0
-        createResult == null
+        createResult.success == true
+        createResult.result == null
 
         when: 'Altering the table'
         def alterResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'ALTER TABLE test_table ADD COLUMN description VARCHAR(255)'
         ])
 
-        then: 'Column should be added and result should be null'
+        then: 'Column should be added and result should indicate success'
         sql.rows('SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = \'TEST_TABLE\' AND COLUMN_NAME = \'DESCRIPTION\'').size() > 0
-        alterResult == null
+        alterResult.success == true
+        alterResult.result == null
 
         when: 'Dropping the table'
         def dropResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'DROP TABLE test_table'
         ])
 
-        then: 'Table should be dropped and result should be null'
+        then: 'Table should be dropped and result should indicate success'
         sql.rows('SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = \'TEST_TABLE\'').size() == 0
-        dropResult == null
+        dropResult.success == true
+        dropResult.result == null
     }
 
     def 'should execute DML statements successfully and return affected row count'() {
@@ -104,30 +107,33 @@ class SqlExecutionTest extends Specification {
             statement: 'INSERT INTO test_dml (id, name, value) VALUES (1, \'item1\', 100)'
         ])
 
-        then: 'Row should be inserted and result should be 1'
+        then: 'Row should be inserted and result should indicate success with 1 row affected'
         sql.rows('SELECT * FROM test_dml').size() == 1
         sql.firstRow('SELECT * FROM test_dml WHERE id = 1').name == 'item1'
-        insertResult == 1
+        insertResult.success == true
+        insertResult.result == 1
 
         when: 'Updating data'
         def updateResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'UPDATE test_dml SET value = 200 WHERE id = 1'
         ])
 
-        then: 'Row should be updated and result should be 1'
+        then: 'Row should be updated and result should indicate success with 1 row affected'
         sql.firstRow('SELECT value FROM test_dml WHERE id = 1').value == 200
-        updateResult == 1
+        updateResult.success == true
+        updateResult.result == 1
 
         when: 'Deleting data'
         def deleteResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'DELETE FROM test_dml WHERE id = 1'
         ])
 
-        then: 'Row should be deleted and result should be 1'
+        then: 'Row should be deleted and result should indicate success with 1 row affected'
         sql.rows('SELECT * FROM test_dml').size() == 0
-        deleteResult == 1
+        deleteResult.success == true
+        deleteResult.result == 1
     }
 
     def 'should return correct affected row count for multiple row operations'() {
@@ -149,40 +155,42 @@ class SqlExecutionTest extends Specification {
         sqlExtension.init(session)
 
         when: 'Inserting data'
-        def insertCount = sqlExtension.sqlExecute([
+        def insertResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'INSERT INTO test_update (id, name, value) VALUES (4, \'item4\', 100)'
         ])
 
-        then: 'Should return 1 affected row'
-        insertCount == 1
+        then: 'Should return success with 1 affected row'
+        insertResult.success == true
+        insertResult.result == 1
         sql.rows('SELECT * FROM test_update').size() == 4
 
         when: 'Updating multiple rows'
-        def updateCount = sqlExtension.sqlExecute([
+        def updateResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'UPDATE test_update SET value = 200 WHERE value = 100'
         ])
 
-        then: 'Should return 4 affected rows'
-        updateCount == 4
+        then: 'Should return success with 4 affected rows'
+        updateResult.success == true
+        updateResult.result == 4
         sql.rows('SELECT * FROM test_update WHERE value = 200').size() == 4
 
         when: 'Deleting multiple rows'
-        def deleteCount = sqlExtension.sqlExecute([
+        def deleteResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'DELETE FROM test_update WHERE value = 200'
         ])
 
-        then: 'Should return 4 affected rows'
-        deleteCount == 4
+        then: 'Should return success with 4 affected rows'
+        deleteResult.success == true
+        deleteResult.result == 4
         sql.rows('SELECT * FROM test_update').size() == 0
     }
 
     def 'should handle invalid SQL correctly'() {
         given:
         def JDBC_URL = 'jdbc:h2:mem:test_error_' + Random.newInstance().nextInt(1_000_000)
-        def sql = Sql.newInstance(JDBC_URL, 'sa', null)
 
         and:
         def session = Mock(Session) {
@@ -192,22 +200,24 @@ class SqlExecutionTest extends Specification {
         sqlExtension.init(session)
 
         when: 'Executing invalid SQL'
-        sqlExtension.sqlExecute([
+        def invalidResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'INVALID SQL STATEMENT'
         ])
 
-        then: 'Should throw an exception'
-        thrown(Exception)
+        then: 'Should return failure with error message'
+        invalidResult.success == false
+        invalidResult.error != null
 
         when: 'Executing query with invalid table name'
-        sqlExtension.sqlExecute([
+        def noTableResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'SELECT * FROM non_existent_table'
         ])
 
-        then: 'Should throw an exception'
-        thrown(Exception)
+        then: 'Should return failure with error message'
+        noTableResult.success == false
+        noTableResult.error != null
     }
 
     def 'should handle invalid database configuration correctly'() {
@@ -219,30 +229,36 @@ class SqlExecutionTest extends Specification {
         sqlExtension.init(session)
 
         when: 'Using non-existent database alias'
-        sqlExtension.sqlExecute([
+        def nonExistentDbResult = sqlExtension.sqlExecute([
             db: 'non_existent_db',
             statement: 'SELECT 1'
         ])
 
-        then: 'Should throw an IllegalArgumentException'
-        thrown(IllegalArgumentException)
+        then: 'Should return failure with error message'
+        nonExistentDbResult.success == false
+        nonExistentDbResult.error != null
+        nonExistentDbResult.error.contains('Unknown db name')
 
         when: 'Missing statement parameter'
-        sqlExtension.sqlExecute([
+        def missingStatementResult = sqlExtension.sqlExecute([
             db: 'test'
         ])
 
-        then: 'Should throw an IllegalArgumentException'
-        thrown(IllegalArgumentException)
+        then: 'Should return failure with error message'
+        missingStatementResult.success == false
+        missingStatementResult.error != null
+        missingStatementResult.error.contains('Missing required parameter')
 
         when: 'Empty statement parameter'
-        sqlExtension.sqlExecute([
+        def emptyStatementResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: ''
         ])
 
-        then: 'Should throw an IllegalArgumentException'
-        thrown(IllegalArgumentException)
+        then: 'Should return failure with error message'
+        emptyStatementResult.success == false
+        emptyStatementResult.error != null
+        emptyStatementResult.error.contains('Missing required parameter')
     }
 
     def 'should handle statement normalization correctly'() {
@@ -258,23 +274,25 @@ class SqlExecutionTest extends Specification {
         sqlExtension.init(session)
 
         when: 'Executing statement without semicolon'
-        def result = sqlExtension.sqlExecute([
+        def createResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'CREATE TABLE test_norm(id INT PRIMARY KEY)'
         ])
 
-        then: 'Statement should be executed successfully and result should be null'
+        then: 'Statement should be executed successfully'
         sql.rows('SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = \'TEST_NORM\'').size() > 0
-        result == null
+        createResult.success == true
+        createResult.result == null
 
         when: 'Executing statement with trailing whitespace'
         def dropResult = sqlExtension.sqlExecute([
             db: 'test',
             statement: 'DROP TABLE test_norm  '
         ])
 
-        then: 'Statement should be executed successfully and result should be null'
+        then: 'Statement should be executed successfully'
         sql.rows('SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = \'TEST_NORM\'').size() == 0
-        dropResult == null
+        dropResult.success == true
+        dropResult.result == null
     }
 } 

plugins/nf-sqldb/src/testResources/testDir/test_sql_db.nf

@@ -3,7 +3,7 @@ nextflow.enable.dsl=2
 include { fromQuery; sqlInsert; sqlExecute } from 'plugin/nf-sqldb'
 
 workflow {
-    // Setup: create table (DDL operation returns null)
+    // Setup: create table (DDL operation)
     def createResult = sqlExecute(
         db: 'foo',
         statement: '''
@@ -14,7 +14,13 @@ workflow {
             )
         '''
     )
-    println "Create result: $createResult" // null
+    println "Create table success: ${createResult.success}" // Should be true
+    
+    // Handle potential failure
+    if (!createResult.success) {
+        println "Failed to create table: ${createResult.error}"
+        return
+    }
 
     // Insert data using sqlInsert
     Channel
@@ -29,10 +35,15 @@ workflow {
     fromQuery('SELECT * FROM sample_table', db: 'foo')
         .view()
 
-    // Update data using sqlExecute (DML operation returns affected row count)
-    def updated = sqlExecute(
+    // Update data using sqlExecute (DML operation returns affected row count in result field)
+    def updateResult = sqlExecute(
         db: 'foo',
         statement: "UPDATE sample_table SET value = 30.5 WHERE name = 'beta'"
     )
-    println "Updated $updated row(s)"
+    
+    if (updateResult.success) {
+        println "Updated ${updateResult.result} row(s)"
+    } else {
+        println "Update failed: ${updateResult.error}"
+    }
 }