refactor: Rewrite execute functions into on one function

- Enhanced logging in examples to provide feedback on the number of affected rows and results of DDL operations.

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

Co-authored-by: Paolo Di Tommaso <[email protected]>

Author: edmundmiller

README.md

@@ -4,13 +4,13 @@ This plugin provides support for interacting with SQL databases in Nextflow scri
 
 The following databases are currently supported:
 
-* [AWS Athena](https://aws.amazon.com/athena/) (Setup guide [here](docs/aws-athena.md))
-* [DuckDB](https://duckdb.org/)
-* [H2](https://www.h2database.com)
-* [MySQL](https://www.mysql.com/)
-* [MariaDB](https://mariadb.org/)
-* [PostgreSQL](https://www.postgresql.org/)
-* [SQLite](https://www.sqlite.org/index.html)
+- [AWS Athena](https://aws.amazon.com/athena/) (Setup guide [here](docs/aws-athena.md))
+- [DuckDB](https://duckdb.org/)
+- [H2](https://www.h2database.com)
+- [MySQL](https://www.mysql.com/)
+- [MariaDB](https://mariadb.org/)
+- [PostgreSQL](https://www.postgresql.org/)
+- [SQLite](https://www.sqlite.org/index.html)
 
 NOTE: THIS IS A PREVIEW TECHNOLOGY, FEATURES AND CONFIGURATION SETTINGS CAN CHANGE IN FUTURE RELEASES.
 
@@ -24,7 +24,6 @@ plugins {
 }
 ```
 
-
 ## Configuration
 
 You can configure any number of databases under the `sql.db` configuration scope. For example:
@@ -79,7 +78,7 @@ The following options are available:
 
 `batchSize`
 : Query the data in batches of the given size. This option is recommended for queries that may return large a large result set, so that the entire result set is not loaded into memory at once.
-: *NOTE:* this feature requires that the underlying SQL database supports `LIMIT` and `OFFSET`.
+: _NOTE:_ this feature requires that the underlying SQL database supports `LIMIT` and `OFFSET`.
 
 `emitColumns`
 : When `true`, the column names in the `SELECT` statement are emitted as the first tuple in the resulting channel.
@@ -104,7 +103,7 @@ INSERT INTO SAMPLE (NAME, LEN) VALUES ('HELLO', 5);
 INSERT INTO SAMPLE (NAME, LEN) VALUES ('WORLD!', 6);
 ```
 
-*NOTE:* the target table (e.g. `SAMPLE` in the above example) must be created beforehand.
+_NOTE:_ the target table (e.g. `SAMPLE` in the above example) must be created beforehand.
 
 The following options are available:
 
@@ -125,21 +124,23 @@ The following options are available:
 
 `setup`
 : A SQL statement that is executed before inserting the data, e.g. to create the target table.
-: *NOTE:* the underlying database should support the *create table if not exist* idiom, as the plugin will execute this statement every time the script is run.
+: _NOTE:_ the underlying database should support the _create table if not exist_ idiom, as the plugin will execute this statement every time the script is run.
 
 ## SQL Execution Functions
 
-This plugin provides the following functions for executing SQL statements that don't return data, such as DDL (Data Definition Language) and DML (Data Manipulation Language) operations.
+This plugin provides the following function for executing SQL statements that don't return data, such as DDL (Data Definition Language) and DML (Data Manipulation Language) operations.
 
 ### sqlExecute
 
-The `sqlExecute` function executes a SQL statement that doesn't return a result set, such as `CREATE`, `ALTER`, `DROP`, `INSERT`, `UPDATE`, or `DELETE` statements. For example:
+The `sqlExecute` function executes a SQL statement that doesn't return a result set, such as `CREATE`, `ALTER`, `DROP`, `INSERT`, `UPDATE`, or `DELETE` statements. For DML statements (`INSERT`, `UPDATE`, `DELETE`), it returns the number of rows affected. For DDL statements (`CREATE`, `ALTER`, `DROP`), it returns `null`.
+
+For example:
 
 ```nextflow
 include { sqlExecute } from 'plugin/nf-sqldb'
 
-// Create a table
-sqlExecute(
+// Create a table (returns null for DDL operations)
+def createResult = sqlExecute(
     db: 'foo',
     statement: '''
         CREATE TABLE IF NOT EXISTS sample_table (
@@ -149,51 +150,24 @@ sqlExecute(
         )
     '''
 )
+println "Create result: $createResult" // null
 
-// Insert data
-sqlExecute(
+// Insert data (returns 1 for number of rows affected)
+def insertedRows = sqlExecute(
     db: 'foo',
     statement: "INSERT INTO sample_table (id, name, value) VALUES (1, 'alpha', 10.5)"
 )
-
-// Delete data
-sqlExecute(
-    db: 'foo',
-    statement: "DELETE FROM sample_table WHERE id = 1"
-)
-```
-
-The following options are available:
-
-`db`
-: The database handle. It must be defined under `sql.db` in the Nextflow configuration.
-
-`statement`
-: The SQL statement to execute. This can be any DDL or DML statement that doesn't return a result set.
-
-### executeUpdate
-
-The `executeUpdate` function is similar to `sqlExecute`, but it returns the number of rows affected by the SQL statement. This is particularly useful for DML operations like `INSERT`, `UPDATE`, and `DELETE` where you need to know how many rows were affected. For example:
-
-```nextflow
-include { executeUpdate } from 'plugin/nf-sqldb'
-
-// Insert data and get the number of rows inserted
-def insertedRows = executeUpdate(
-    db: 'foo',
-    statement: "INSERT INTO sample_table (id, name, value) VALUES (2, 'beta', 20.5)"
-)
 println "Inserted $insertedRows row(s)"
 
-// Update data and get the number of rows updated
-def updatedRows = executeUpdate(
+// Update data (returns number of rows updated)
+def updatedRows = sqlExecute(
     db: 'foo',
-    statement: "UPDATE sample_table SET value = 30.5 WHERE name = 'beta'"
+    statement: "UPDATE sample_table SET value = 30.5 WHERE name = 'alpha'"
 )
 println "Updated $updatedRows row(s)"
 
-// Delete data and get the number of rows deleted
-def deletedRows = executeUpdate(
+// Delete data (returns number of rows deleted)
+def deletedRows = sqlExecute(
     db: 'foo',
     statement: "DELETE FROM sample_table WHERE value > 25"
 )
@@ -206,25 +180,27 @@ The following options are available:
 : The database handle. It must be defined under `sql.db` in the Nextflow configuration.
 
 `statement`
-: The SQL statement to execute. This should be a DML statement that can return a count of affected rows.
+: The SQL statement to execute. This can be any DDL or DML statement that doesn't return a result set.
 
-## Differences Between Dataflow Operators and Execution Functions
+## Differences Between Dataflow Operators and Execution Function
 
 The plugin provides two different ways to interact with databases:
 
 1. **Dataflow Operators** (`fromQuery` and `sqlInsert`): These are designed to integrate with Nextflow's dataflow programming model, operating on channels.
+
    - `fromQuery`: Queries data from a database and returns a channel that emits the results.
    - `sqlInsert`: Takes data from a channel and inserts it into a database.
 
-2. **Execution Functions** (`sqlExecute` and `executeUpdate`): These are designed for direct SQL statement execution that doesn't require channel integration.
-   - `sqlExecute`: Executes a SQL statement without returning any data.
-   - `executeUpdate`: Executes a SQL statement and returns the count of affected rows.
+2. **Execution Function** (`sqlExecute`): This is designed for direct SQL statement execution that doesn't require channel integration.
+   - `sqlExecute`: Executes a SQL statement. For DML operations, it returns the count of affected rows. For DDL operations, it returns null.
 
 Use **Dataflow Operators** when you need to:
+
 - Query data that will flow into your pipeline processing
 - Insert data from your pipeline processing into a database
 
-Use **Execution Functions** when you need to:
+Use **Execution Function** when you need to:
+
 - Perform database setup (creating tables, schemas, etc.)
 - Execute administrative commands
 - Perform one-off operations (deleting all records, truncating a table, etc.)
@@ -262,8 +238,7 @@ The `CSVREAD` function provided by the H2 database engine allows you to query an
 
 Like all dataflow operators in Nextflow, the operators provided by this plugin are executed asynchronously.
 
-In particular, data inserted using the `sqlInsert` operator is *not* guaranteed to be available to any subsequent queries using the `fromQuery` operator, as it is not possible to make a channel factory operation dependent on some upstream operation.
-
+In particular, data inserted using the `sqlInsert` operator is _not_ guaranteed to be available to any subsequent queries using the `fromQuery` operator, as it is not possible to make a channel factory operation dependent on some upstream operation.
 
 ## Developtment 
 

examples/sql-execution/main.nf

@@ -1,10 +1,10 @@
 #!/usr/bin/env nextflow
 
 /*
- * Example script demonstrating how to use the SQL sqlExecute and executeUpdate functions
+ * Example script demonstrating how to use the SQL sqlExecute function
  */
 
-include { sqlExecute; executeUpdate } from 'plugin/nf-sqldb'
+include { sqlExecute } from 'plugin/nf-sqldb'
 include { fromQuery } from 'plugin/nf-sqldb'
 
 // Define database configuration in nextflow.config file
@@ -13,13 +13,13 @@ include { fromQuery } from 'plugin/nf-sqldb'
 workflow {
     log.info """
     =========================================
-    SQL Execution Functions Example
+    SQL Execution Function Example
     =========================================
     """
 
-    // Step 1: Create a table
+    // Step 1: Create a table (DDL operation returns null)
     log.info "Creating a sample table..."
-    def createResult = executeUpdate(
+    def createResult = sqlExecute(
         db: 'demo',
         statement: '''
             CREATE TABLE IF NOT EXISTS TEST_TABLE (
@@ -31,9 +31,9 @@ workflow {
     )
     log.info "Create table result: $createResult"
 
-    // Step 2: Insert some data
+    // Step 2: Insert some data (DML operation returns affected row count)
     log.info "Inserting data..."
-    executeUpdate(
+    def insertCount = sqlExecute(
         db: 'demo',
         statement: '''
             INSERT INTO TEST_TABLE (ID, NAME, VALUE) VALUES
@@ -43,27 +43,30 @@ workflow {
             (4, 'delta', 40.9);
         '''
     )
+    log.info "Inserted $insertCount rows"
 
-    // Step 3: Update some data
+    // Step 3: Update some data (DML operation returns affected row count)
     log.info "Updating data..."
-    executeUpdate(
+    def updateCount = sqlExecute(
         db: 'demo',
         statement: '''
             UPDATE TEST_TABLE
             SET VALUE = VALUE * 2
             WHERE ID = 2;
         '''
     )
+    log.info "Updated $updateCount rows"
 
-    // Step 4: Delete some data
+    // Step 4: Delete some data (DML operation returns affected row count)
     log.info "Deleting data..."
-    executeUpdate(
+    def deleteCount = sqlExecute(
         db: 'demo',
         statement: '''
             DELETE FROM TEST_TABLE
             WHERE ID = 4;
         '''
     )
+    log.info "Deleted $deleteCount rows"
 
     // Step 5: Query results
     log.info "Querying results..."

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

@@ -145,11 +145,14 @@ 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
      *
      * @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
      */
     @Function
-    void sqlExecute(Map params) {
+    Integer sqlExecute(Map params) {
         CheckHelper.checkParams('sqlExecute', params, EXECUTE_PARAMS)
 
         final String dbName = params.db as String ?: 'default'
@@ -173,7 +176,18 @@ class ChannelSqlExtension extends PluginExtensionPoint {
 
         try (Connection conn = groovy.sql.Sql.newInstance(dataSource.toMap()).getConnection()) {
             try (Statement stm = conn.createStatement()) {
-                stm.execute(normalizeStatement(statement))
+                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
+                } else {
+                    return stm.executeUpdate(normalizedStatement)
+                }
             }
         }
         catch (Exception e) {
@@ -182,47 +196,6 @@ class ChannelSqlExtension extends PluginExtensionPoint {
         }
     }
 
-    /**
-     * Execute a SQL statement that does not return a result set (DDL/DML statements)
-     * and returns the number of affected rows
-     *
-     * @param params A map containing 'db' (database alias) and 'statement' (SQL string to execute)
-     * @return The number of rows affected by the SQL statement
-     */
-    @Function
-    int executeUpdate(Map params) {
-        CheckHelper.checkParams('executeUpdate', 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'")
-            
-        final sqlConfig = new SqlConfig((Map) session.config.navigate('sql.db'))
-        final SqlDataSource dataSource = sqlConfig.getDataSource(dbName)
-        
-        if (dataSource == null) {
-            def msg = "Unknown db name: $dbName"
-            def choices = sqlConfig.getDataSourceNames().closest(dbName) ?: sqlConfig.getDataSourceNames()
-            if (choices?.size() == 1)
-                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)
-        }
-        
-        try (Connection conn = groovy.sql.Sql.newInstance(dataSource.toMap()).getConnection()) {
-            try (Statement stm = conn.createStatement()) {
-                return stm.executeUpdate(normalizeStatement(statement))
-            }
-        }
-        catch (Exception e) {
-            log.error("Error executing SQL update statement: ${e.message}", e)
-            throw e
-        }
-    }
-    
     /**
      * Normalizes a SQL statement by adding a semicolon if needed
      *