Add support for streaming and cursor query execution in SQL templates

vertx-sql-client-templates/src/main/asciidoc/index.adoc

@@ -65,6 +65,35 @@ When your template must be executed inside a transaction, you might create a tem
 ----
 ====
 
+== Streaming
+
+When dealing with large result sets, you can use {@link io.vertx.sqlclient.templates.SqlTemplateStream} to read rows progressively using a cursor with a configurable fetch size, instead of loading all rows in memory at once.
+
+NOTE: Streaming requires a {@link io.vertx.sqlclient.SqlConnection}. Some databases (e.g. PostgreSQL) also require an active transaction for cursors.
+
+[source,$lang]
+----
+{@link examples.TemplateExamples#streamExample}
+----
+
+You can use `mapTo` to map each row emitted by the stream to a custom type:
+
+[source,$lang]
+----
+{@link examples.TemplateExamples#streamWithMapToExample}
+----
+
+== Cursor
+
+If you need finer control over row fetching, you can use a cursor-based template with {@link io.vertx.sqlclient.templates.SqlTemplate#forCursor}. This gives you a {@link io.vertx.sqlclient.Cursor} that allows you to read rows in batches.
+
+NOTE: Cursors require a {@link io.vertx.sqlclient.SqlConnection}. Some databases (e.g. PostgreSQL) also require an active transaction for cursors.
+
+[source,$lang]
+----
+{@link examples.TemplateExamples#cursorExample}
+----
+
 == Template syntax
 
 The template syntax uses `#{XXX}` syntax where `XXX` is a valid https://docs.oracle.com/javase/specs/jls/se8/html/jls-3.html#jls-3.8[java identifier] string

vertx-sql-client-templates/src/main/java/examples/TemplateExamples.java

@@ -9,6 +9,7 @@
 import io.vertx.sqlclient.*;
 import io.vertx.sqlclient.templates.RowMapper;
 import io.vertx.sqlclient.templates.SqlTemplate;
+import io.vertx.sqlclient.templates.SqlTemplateStream;
 import io.vertx.sqlclient.templates.TupleMapper;
 import io.vertx.sqlclient.templates.annotations.Column;
 import io.vertx.sqlclient.templates.annotations.ParametersMapped;
@@ -424,6 +425,56 @@ public Tuple map(Function<Integer, String> mapping, int size, UserDataObject par
     }
   }
 
+  public void streamExample(SqlConnection connection) {
+    SqlTemplateStream
+      .forStream(connection, "SELECT * FROM users WHERE age > #{age}", 50)
+      .execute(Collections.singletonMap("age", 18))
+      .onSuccess(stream -> {
+        stream.handler(row -> {
+          System.out.println(row.getString("first_name") + " " + row.getString("last_name"));
+        });
+        stream.endHandler(v -> {
+          System.out.println("End of stream");
+        });
+        stream.exceptionHandler(err -> {
+          System.out.println("Error: " + err.getMessage());
+        });
+      });
+  }
+
+  public void streamWithMapToExample(SqlConnection connection) {
+    SqlTemplateStream
+      .forStream(connection, "SELECT * FROM users WHERE age > #{age}", 50)
+      .mapTo(ROW_USER_MAPPER)
+      .execute(Collections.singletonMap("age", 18))
+      .onSuccess(stream -> {
+        stream.handler(user -> {
+          System.out.println(user.firstName + " " + user.lastName);
+        });
+        stream.endHandler(v -> {
+          System.out.println("End of stream");
+        });
+      });
+  }
+
+  public void cursorExample(SqlConnection connection) {
+    SqlTemplate
+      .forCursor(connection, "SELECT * FROM users WHERE age > #{age}")
+      .execute(Collections.singletonMap("age", 18))
+      .onSuccess(cursor -> {
+        cursor.read(100).onSuccess(rows -> {
+          rows.forEach(row -> {
+            System.out.println(row.getString("first_name") + " " + row.getString("last_name"));
+          });
+          if (cursor.hasMore()) {
+            // Read more
+          } else {
+            cursor.close();
+          }
+        });
+      });
+  }
+
   public void templateInTransaction(Pool pool) {
     SqlTemplate<Map<String, Object>, RowSet<UserDataObject>> template = SqlTemplate
       .forQuery(pool, "SELECT * FROM users WHERE id=#{id}")

vertx-sql-client-templates/src/main/java/io/vertx/sqlclient/templates/RowMapper.java

@@ -11,6 +11,7 @@
 package io.vertx.sqlclient.templates;
 
 import io.vertx.codegen.annotations.VertxGen;
+import io.vertx.core.json.JsonObject;
 import io.vertx.sqlclient.Row;
 
 /**
@@ -20,6 +21,25 @@
 @FunctionalInterface
 public interface RowMapper<T> {
 
+  /**
+   * Create a mapper that converts a {@link Row} to an instance of the given {@code type}.
+   *
+   * <p>This feature relies on {@link io.vertx.core.json.JsonObject#mapTo} feature. This likely requires
+   * to use Jackson databind in the project.
+   *
+   * @param type the target class
+   * @return the mapper
+   */
+  static <T> RowMapper<T> mapper(Class<T> type) {
+    return row -> {
+      JsonObject json = new JsonObject();
+      for (int i = 0; i < row.size(); i++) {
+        json.getMap().put(row.getColumnName(i), row.getValue(i));
+      }
+      return json.mapTo(type);
+    };
+  }
+
   /**
    * Build a {@code T} representation of the given {@code row}
    *

vertx-sql-client-templates/src/main/java/io/vertx/sqlclient/templates/SqlTemplate.java

@@ -14,15 +14,11 @@
 import io.vertx.codegen.annotations.GenIgnore;
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Future;
-import io.vertx.core.json.JsonObject;
-import io.vertx.sqlclient.Row;
-import io.vertx.sqlclient.RowSet;
-import io.vertx.sqlclient.SqlClient;
-import io.vertx.sqlclient.SqlResult;
+import io.vertx.sqlclient.*;
 import io.vertx.sqlclient.impl.SqlClientInternal;
+import io.vertx.sqlclient.templates.impl.CursorSqlTemplateImpl;
 import io.vertx.sqlclient.templates.impl.SqlTemplateImpl;
 
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -69,6 +65,32 @@ static SqlTemplate<Map<String, Object>, SqlResult<Void>> forUpdate(SqlClient cli
     return new SqlTemplateImpl<>(clientInternal, sqlTemplate, query -> query.collecting(SqlTemplateImpl.NULL_COLLECTOR), sqlTemplate::mapTuple);
   }
 
+  /**
+   * Create an SQL template for streaming query results consuming map parameters and returning {@link Row}.
+   *
+   * <p>Delegates to {@link SqlTemplateStream#forStream(SqlConnection, String, int)}.
+   *
+   * @param client the wrapped SQL connection
+   * @param template the template query string
+   * @param fetchSize the cursor fetch size
+   * @return the template
+   */
+  static SqlTemplateStream<Map<String, Object>, Row> forStream(SqlConnection client, String template, int fetchSize) {
+    return SqlTemplateStream.forStream(client, template, fetchSize);
+  }
+
+  /**
+   * Create an SQL template for cursor-based query execution consuming map parameters and returning a {@link Cursor}.
+   *
+   * @param client the wrapped SQL connection
+   * @param template the template query string
+   * @return the template
+   */
+  static SqlTemplate<Map<String, Object>, Cursor> forCursor(SqlConnection client, String template) {
+    SqlClientInternal clientInternal = (SqlClientInternal) client;
+    io.vertx.sqlclient.templates.impl.SqlTemplate sqlTemplate = io.vertx.sqlclient.templates.impl.SqlTemplate.create(clientInternal, template);
+    return new CursorSqlTemplateImpl<>(client, sqlTemplate, sqlTemplate::mapTuple);
+  }
 
   /**
    * @return the computed SQL for this template
@@ -99,14 +121,7 @@ static SqlTemplate<Map<String, Object>, SqlResult<Void>> forUpdate(SqlClient cli
    * @return a new template
    */
   default <T> SqlTemplate<T, R> mapFrom(Class<T> type) {
-    return mapFrom(TupleMapper.mapper(params -> {
-      JsonObject jsonObject = JsonObject.mapFrom(params);
-      Map<String, Object> map = new LinkedHashMap<>(jsonObject.size());
-      for (String fieldName : jsonObject.fieldNames()) {
-        map.put(fieldName, jsonObject.getValue(fieldName));
-      }
-      return map;
-    }));
+    return mapFrom(TupleMapper.mapper(type));
   }
 
   /**

vertx-sql-client-templates/src/main/java/io/vertx/sqlclient/templates/SqlTemplateStream.java

@@ -0,0 +1,115 @@
+package io.vertx.sqlclient.templates;
+
+import io.vertx.codegen.annotations.VertxGen;
+import io.vertx.core.Future;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowStream;
+import io.vertx.sqlclient.SqlConnection;
+import io.vertx.sqlclient.impl.SqlClientInternal;
+import io.vertx.sqlclient.templates.impl.SqlTemplateStreamImpl;
+
+import java.util.Map;
+
+/**
+ * An SQL template for streaming query results.
+ *
+ * <p>Stream templates execute queries using named instead of positional parameters and return results
+ * as a {@link RowStream} that reads rows progressively using a cursor with a configurable fetch size.
+ *
+ * @param <I> the input parameters type
+ * @param <T> the row output type
+ */
+@VertxGen
+public interface SqlTemplateStream<I, T> {
+
+  /**
+   * Create an SQL template for streaming query results consuming map parameters and returning {@link Row}.
+   *
+   * <p>The returned stream template uses a cursor with the given {@code fetchSize} to read rows progressively.
+   *
+   * @param client the wrapped SQL connection
+   * @param template the template query string
+   * @param fetchSize the cursor fetch size
+   * @return the template
+   */
+  static SqlTemplateStream<Map<String, Object>, Row> forStream(SqlConnection client, String template, int fetchSize) {
+    SqlClientInternal clientInternal = (SqlClientInternal) client;
+    io.vertx.sqlclient.templates.impl.SqlTemplate sqlTemplate = io.vertx.sqlclient.templates.impl.SqlTemplate.create(clientInternal, template);
+    return new SqlTemplateStreamImpl<>(client, sqlTemplate, sqlTemplate::mapTuple, null, fetchSize);
+  }
+
+  /**
+   * @return the computed SQL for this template
+   */
+  String sql();
+
+  /**
+   * Set a parameters user defined mapping function.
+   *
+   * <p> At query execution, the {@code mapper} is called to map the parameters object
+   * to a {@link io.vertx.sqlclient.Tuple} that configures the prepared query.
+   *
+   * @param mapper the mapping function
+   * @return a new template
+   */
+  <T2> SqlTemplateStream<T2, T> mapFrom(TupleMapper<T2> mapper);
+
+  /**
+   * Set a parameters user defined class mapping.
+   *
+   * <p> At query execution, the parameters object is mapped to a {@code Map<String, Object>} that
+   * configures the prepared query.
+   *
+   * <p> This feature relies on {@link io.vertx.core.json.JsonObject#mapFrom} feature. This likely requires
+   * to use Jackson databind in the project.
+   *
+   * @param type the mapping type
+   * @return a new template
+   */
+  default <T2> SqlTemplateStream<T2, T> mapFrom(Class<T2> type) {
+    return mapFrom(TupleMapper.mapper(type));
+  }
+
+  /**
+   * Set a row user defined mapping function.
+   *
+   * <p>When rows are emitted by the stream, the {@code mapper} function is called to map each {@link Row}
+   * to the target type.
+   *
+   * @param mapper the mapping function
+   * @return a new template
+   */
+  <U> SqlTemplateStream<I, U> mapTo(RowMapper<U> mapper);
+
+  /**
+   * Set a row user defined mapping function.
+   *
+   * <p>When rows are emitted by the stream, resulting rows are mapped to {@code type} instances.
+   *
+   * <p> This feature relies on {@link io.vertx.core.json.JsonObject#mapFrom} feature. This likely requires
+   * to use Jackson databind in the project.
+   *
+   * @param type the mapping type
+   * @return a new template
+   */
+  default <U> SqlTemplateStream<I, U> mapTo(Class<U> type) {
+    return mapTo(RowMapper.mapper(type));
+  }
+
+  /**
+   * Returns a new template, using the specified {@code connection}.
+   *
+   * @param connection the connection that will execute requests
+   * @return a new template
+   */
+  SqlTemplateStream<I, T> withConnection(SqlConnection connection);
+
+  /**
+   * Execute the query with the {@code parameters}
+   *
+   * @param params the query parameters
+   * @return a future notified with the result
+   */
+  Future<RowStream<T>> execute(I params);
+
+}

vertx-sql-client-templates/src/main/java/io/vertx/sqlclient/templates/TupleMapper.java

@@ -15,6 +15,7 @@
 import io.vertx.sqlclient.Tuple;
 import io.vertx.sqlclient.templates.impl.JsonTuple;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.function.Function;
 
@@ -44,6 +45,26 @@ static <T> TupleMapper<T> mapper(Function<T, Map<String, Object>> fn) {
     };
   }
 
+  /**
+   * Create a mapper that converts a parameters object of the given {@code type} to a map of named parameters.
+   *
+   * <p>This feature relies on {@link io.vertx.core.json.JsonObject#mapFrom} feature. This likely requires
+   * to use Jackson databind in the project.
+   *
+   * @param type the parameters class
+   * @return the mapper
+   */
+  static <T> TupleMapper<T> mapper(Class<T> type) {
+    return mapper(params -> {
+      JsonObject jsonObject = JsonObject.mapFrom(params);
+      Map<String, Object> map = new LinkedHashMap<>(jsonObject.size());
+      for (String fieldName : jsonObject.fieldNames()) {
+        map.put(fieldName, jsonObject.getValue(fieldName));
+      }
+      return map;
+    });
+  }
+
   /**
    * Map a {@link JsonObject} to a {@link Tuple}.
    */