Support parameter types arguments.

Author: paurkedal

examples/async.ml

@@ -54,14 +54,36 @@ let test (c : connection) =
         REFERENCES postgresql_ocaml_async ON DELETE CASCADE)";
   assert ((fetch_single_result c)#status = Command_ok);
 
+  begin
+    c#send_query
+      ~param_types:Postgresql.[|oid_of_ftype INT8; oid_of_ftype INT8|]
+      ~params:[|"4100100100"; "5100100100"|]
+      "SELECT $1 + $2";
+    let r = fetch_single_result c in
+    assert (r#status = Tuples_ok);
+    assert (r#nfields = 1);
+    assert (r#ntuples = 1);
+    assert (r#getvalue 0 0 = "9200200200");
+  end;
+
   (* Populate using a prepared statement. *)
-  c#send_prepare "test_ins"
-                 "INSERT INTO postgresql_ocaml_async (a, b) VALUES ($1, $2)";
-  assert ((fetch_single_result c)#status = Command_ok);
-  c#send_query_prepared ~params:[|"2"; "two"|] "test_ins";
-  assert ((fetch_single_result c)#status = Command_ok);
-  c#send_query_prepared ~params:[|"3"; "three"|] "test_ins";
-  assert ((fetch_single_result c)#status = Command_ok);
+  let shown_ntuples = 10 in
+  let expected_ntuples = 3 * 100 in
+  for i = 0 to 2 do
+    let stmt = sprintf "test_ins_%d" i in
+    let param_types =
+      Array.sub Postgresql.[|oid_of_ftype INT4; oid_of_ftype TEXT|] 0 i
+    in
+    c#send_prepare stmt ~param_types
+      "INSERT INTO postgresql_ocaml_async (a, b) VALUES ($1, $2)";
+    assert ((fetch_single_result c)#status = Command_ok);
+    for j = 1 to 100 do
+      let c0 = string_of_int (i + 3 * j) in
+      let c1 = sprintf "The number %d." (i + 3 * j) in
+      c#send_query_prepared ~params:[|c0; c1|] stmt;
+      assert ((fetch_single_result c)#status = Command_ok)
+    done
+  done;
 
   (* Prepare a select statement. *)
   c#send_prepare "test_sel" "SELECT * FROM postgresql_ocaml_async";
@@ -80,26 +102,27 @@ let test (c : connection) =
   c#send_query_prepared "test_sel";
   let r = fetch_single_result c in
   assert (r#status = Tuples_ok);
-  assert (r#ntuples = 2);
+  assert (r#ntuples = expected_ntuples);
   assert (r#nfields = 3);
-  for i = 0 to r#ntuples - 1 do
-    Printf.printf "%s %s %s\n"
-                  (r#getvalue i 0) (r#getvalue i 1) (r#getvalue i 2)
+  for i = 0 to min r#ntuples shown_ntuples - 1 do
+    printf "%s, %s, %s\n" (r#getvalue i 0) (r#getvalue i 1) (r#getvalue i 2)
   done;
+  printf "[...]\n";
 
   (* Run it in single-row mode. *)
   c#send_query_prepared "test_sel";
   c#set_single_row_mode;
-  for i = 0 to 2 do
+  for i = 0 to expected_ntuples do
     match fetch_result c with
     | None -> assert false
-    | Some r when i < 2 ->
+    | Some r when i < expected_ntuples ->
       assert (r#status = Single_tuple);
-      Printf.printf "%s %s %s\n"
-                    (r#getvalue 0 0) (r#getvalue 0 1) (r#getvalue 0 2)
+      if i < shown_ntuples then
+        printf "%s, %s, %s\n" (r#getvalue 0 0) (r#getvalue 0 1) (r#getvalue 0 2)
     | Some r ->
       assert (r#status = Tuples_ok)
   done;
+  printf "[...]\n";
   assert (fetch_result c = None);
 
   (* Drop the main table. *)

src/postgresql.ml

@@ -380,10 +380,13 @@ module Stub = struct
   external result_isnull : result -> bool = "PQres_isnull" [@@noalloc]
 
   external exec_params :
-    connection -> string -> string array -> bool array -> bool -> result
-    = "PQexecParams_stub"
+    connection -> string -> oid array -> string array -> bool array ->
+    bool -> result
+    = "PQexecParams_stub_bc" "PQexecParams_stub"
 
-  external prepare : connection -> string -> string -> result = "PQprepare_stub"
+  external prepare :
+    connection -> string -> string -> oid array -> result
+    = "PQprepare_stub"
 
   external exec_prepared :
     connection -> string -> string array -> bool array -> result
@@ -473,11 +476,12 @@ module Stub = struct
     connection -> bool = "PQisnonblocking_stub" [@@noalloc]
 
   external send_query_params :
-    connection -> string -> string array -> bool array -> (int [@untagged])
+    connection -> string -> oid array -> string array -> bool array ->
+    (int [@untagged])
     = "PQsendQueryParams_stub_bc" "PQsendQueryParams_stub"
 
   external send_prepare :
-    connection -> string -> string -> (int [@untagged])
+    connection -> string -> string -> oid array -> (int [@untagged])
     = "PQsendPrepare_stub_bc" "PQsendPrepare_stub" [@@noalloc]
 
   external send_query_prepared :
@@ -928,12 +932,13 @@ object (self)
     new result (wrap_conn (fun conn -> (Stub.make_empty_res conn status)))
 
   method exec
-    ?(expect = []) ?(params = [||]) ?(binary_params = [||])
-    ?(binary_result = false) query =
+    ?(expect = []) ?(param_types = [||]) ?(params = [||])
+    ?(binary_params = [||]) ?(binary_result = false) query =
     let r =
       wrap_conn (fun conn ->
         let r =
-          Stub.exec_params conn query params binary_params binary_result
+          Stub.exec_params conn query param_types params
+                           binary_params binary_result
         in
         if Stub.result_isnull r then signal_error conn
         else r)
@@ -944,10 +949,10 @@ object (self)
       raise (Error (Unexpected_status (stat, res#error, expect)))
     else res
 
-  method prepare stm_name query =
+  method prepare ?(param_types = [||]) stm_name query =
     new result (
       wrap_conn (fun conn ->
-        let r = Stub.prepare conn stm_name query in
+        let r = Stub.prepare conn stm_name query param_types in
         if Stub.result_isnull r then signal_error conn
         else r))
 
@@ -972,14 +977,17 @@ object (self)
         if Stub.result_isnull r then signal_error conn
         else r))
 
-  method send_query ?(params = [||]) ?(binary_params = [||]) query =
+  method send_query
+    ?(param_types = [||]) ?(params = [||]) ?(binary_params = [||]) query =
     wrap_conn (fun conn ->
-      if Stub.send_query_params conn query params binary_params <> 1 then
+      if Stub.send_query_params conn query param_types params binary_params
+            <> 1 then
         signal_error conn)
 
-  method send_prepare stm_name query =
+  method send_prepare ?(param_types = [||]) stm_name query =
     wrap_conn (fun conn ->
-      if Stub.send_prepare conn stm_name query <> 1 then signal_error conn)
+      if Stub.send_prepare conn stm_name query param_types <> 1 then
+        signal_error conn)
 
   method send_query_prepared ?(params = [||]) ?(binary_params = [||]) stm_name =
     wrap_conn (fun conn ->

src/postgresql.mli

@@ -704,10 +704,11 @@ object
   *)
 
   method exec :
-    ?expect : result_status list -> ?params : string array ->
+    ?expect : result_status list ->
+    ?param_types : oid array -> ?params : string array ->
     ?binary_params : bool array -> ?binary_result : bool ->
     string -> result
-  (** [exec ?expect ?params ?binary_params ?binary_result query]
+  (** [exec ?expect ?params ?param_types ?binary_params ?binary_result query]
       synchronous execution of query or command [query].  The result
       status will be checked against all elements in [expect].  If
       [expect] is not empty and if there is no match, the exception
@@ -728,6 +729,7 @@ object
       @return result of query.
 
       @param expect default = []
+      @param param_types default = [||]
       @param params default = [||]
       @param binary_params default = [||]
       @param binary_result default = false
@@ -736,9 +738,9 @@ object
       @raise Error if there is an unexpected result status.
   *)
 
-  method prepare : string -> string -> result
-  (** [prepare stm_name query] creates a prepared query named [stm_name]
-      which will execute the query or command [query] when passed to
+  method prepare : ?param_types : oid array -> string -> string -> result
+  (** [prepare ?param_types stm_name query] creates a prepared query named
+      [stm_name] which will execute the query or command [query] when passed to
       [#exec_prepared]. *)
 
   method exec_prepared :
@@ -772,10 +774,10 @@ object
   *)
 
   method send_query :
-    ?params : string array -> ?binary_params : bool array
-    -> string -> unit
-  (** [send_query ?params ?binary_params query] asynchronous execution
-      of query or command [query].
+    ?param_types : oid array -> ?params : string array ->
+    ?binary_params : bool array -> string -> unit
+  (** [send_query ?param_types ?params ?binary_params query] asynchronous
+      execution of query or command [query].
 
       Additional query parameters can be passed in the [params] array.
       They must not be escaped and they can be referred to in [query]
@@ -787,16 +789,17 @@ object
       If no (or an empty) query parameter is passed, it is possible to
       emit several commands with a single call.
 
+      @param param_types default = [||]
       @param params default = [||]
       @param binary_params default = [||]
 
       @raise Error if there is a connection error.
   *)
 
-  method send_prepare : string -> string -> unit
-  (** [#send_prepare stm_name query] sends a query preparation without waiting
-      for the result.  This does the same as {!prepare} except that the status
-      is reported by {!get_result} when available.
+  method send_prepare : ?param_types : oid array -> string -> string -> unit
+  (** [#send_prepare ?param_types stm_name query] sends a query preparation
+      without waiting for the result.  This does the same as {!prepare} except
+      that the status is reported by {!get_result} when available.
 
       @raise Error if there is a connection error. *)
 

src/postgresql_stubs.c

@@ -583,6 +583,23 @@ static inline void free_binary_params(int *formats, int *lengths)
   if (lengths != NULL) caml_stat_free(lengths);
 }
 
+static inline Oid * copy_param_types(
+    value v_param_types, size_t nparams, size_t nparam_types)
+{
+   Oid *param_types;
+   size_t i;
+   if (nparam_types == 0) return NULL;
+   if (nparam_types > nparams) nparam_types = nparams;
+   param_types = caml_stat_alloc(nparams * sizeof(Oid));
+   for (i = 0; i < nparam_types; i++) {
+     value v_param_type = Field(v_param_types, i);
+     param_types[i] = Int_val(v_param_type);
+   }
+   memset(param_types + nparam_types, 0,
+          (nparams - nparam_types) * sizeof(Oid));
+   return param_types;
+}
+
 static inline const char * const * copy_params(value v_params, size_t nparams)
 {
   char **params;
@@ -631,8 +648,8 @@ static inline void free_params_shallow(
 }
 
 CAMLprim value PQexecParams_stub(
-  value v_conn, value v_query, value v_params, value v_binary_params,
-  value v_binary_result)
+  value v_conn, value v_query, value v_param_types, value v_params,
+  value v_binary_params, value v_binary_result)
 {
   CAMLparam1(v_conn);
   PGconn *conn = get_conn(v_conn);
@@ -642,6 +659,8 @@ CAMLprim value PQexecParams_stub(
   char *query = caml_stat_alloc(len);
   size_t nparams = Wosize_val(v_params);
   const char * const *params = copy_params(v_params, nparams);
+  size_t nparam_types = Wosize_val(v_param_types);
+  Oid *param_types = copy_param_types(v_param_types, nparams, nparam_types);
   int *formats, *lengths;
   copy_binary_params(v_params, v_binary_params, nparams, &formats, &lengths);
   memcpy(query, String_val(v_query), len);
@@ -651,17 +670,26 @@ CAMLprim value PQexecParams_stub(
       (nparams == 0 && !binary_result)
         ? PQexec(conn, query)
         : PQexecParams(
-            conn, query, nparams, NULL,
+            conn, query, nparams, param_types,
             params, lengths, formats, binary_result);
+    if (param_types != NULL) caml_stat_free(param_types);
     free_binary_params(formats, lengths);
     free_params(params, nparams);
     caml_stat_free(query);
   caml_leave_blocking_section();
   CAMLreturn(alloc_result(res, np_cb));
 }
 
+CAMLprim value PQexecParams_stub_bc(value *argv, int argn)
+{
+  (void)argn; /* unused */
+  return
+    PQexecParams_stub(argv[0], argv[1], argv[2], argv[3], argv[4], argv[5]);
+}
+
 #ifdef PG_OCAML_8_2
-CAMLprim value PQprepare_stub(value v_conn, value v_stm_name, value v_query)
+CAMLprim value PQprepare_stub(
+  value v_conn, value v_stm_name, value v_query, value v_param_types)
 {
   CAMLparam1(v_conn);
   PGconn *conn = get_conn(v_conn);
@@ -671,10 +699,13 @@ CAMLprim value PQprepare_stub(value v_conn, value v_stm_name, value v_query)
   size_t query_len = caml_string_length(v_query) + 1;
   char *stm_name = caml_stat_alloc(stm_name_len);
   char *query = caml_stat_alloc(query_len);
+  size_t nparams = Wosize_val(v_param_types);
+  Oid *param_types = copy_param_types(v_param_types, nparams, nparams);
   memcpy(stm_name, String_val(v_stm_name), stm_name_len);
   memcpy(query, String_val(v_query), query_len);
   caml_enter_blocking_section();
-    res = PQprepare(conn, stm_name, query, 0, NULL);
+    res = PQprepare(conn, stm_name, query, nparams, param_types);
+    if (param_types != NULL) caml_stat_free(param_types);
     caml_stat_free(stm_name);
     caml_stat_free(query);
   caml_leave_blocking_section();
@@ -963,45 +994,56 @@ CAMLprim value PQsetnonblocking_stub_bc(value v_conn, value v_arg)
 noalloc_conn_info(PQisnonblocking, Val_bool)
 
 CAMLprim intnat PQsendQueryParams_stub(
-  value v_conn, value v_query, value v_params, value v_binary_params)
+  value v_conn, value v_query, value v_param_types, value v_params,
+  value v_binary_params)
 {
   PGconn *conn = get_conn(v_conn);
   const char *query = String_val(v_query);
   size_t nparams = Wosize_val(v_params);
   const char * const *params = copy_params_shallow(v_params, nparams);
+  size_t nparam_types = Wosize_val(v_param_types);
+  Oid *param_types = copy_param_types(v_param_types, nparams, nparam_types);
   int *lengths, *formats;
   intnat res;
   copy_binary_params(v_params, v_binary_params, nparams, &formats, &lengths);
   res =
     (nparams == 0)
       ? PQsendQuery(conn, query)
       : PQsendQueryParams(
-          conn, query, nparams, NULL, params, lengths, formats, 0);
+          conn, query, nparams, param_types, params, lengths, formats, 0);
+  if (param_types != NULL) caml_stat_free(param_types);
   free_binary_params(formats, lengths);
   free_params_shallow(params, nparams);
   return res;
 }
 
 CAMLprim value PQsendQueryParams_stub_bc(
-  value v_conn, value v_query, value v_params, value v_binary_params)
+  value v_conn, value v_query, value v_param_types, value v_params,
+  value v_binary_params)
 {
-  return
-    Val_int(PQsendQueryParams_stub(v_conn, v_query, v_params, v_binary_params));
+  return Val_int(PQsendQueryParams_stub(
+                    v_conn, v_query, v_param_types, v_params, v_binary_params));
 }
 
 CAMLprim intnat PQsendPrepare_stub(
-    value v_conn, value v_stm_name, value v_query)
+    value v_conn, value v_stm_name, value v_query, value v_param_types)
 {
   PGconn *conn = get_conn(v_conn);
   const char *stm_name = String_val(v_stm_name);
   const char *query = String_val(v_query);
-  return PQsendPrepare(conn, stm_name, query, 0, NULL);
+  size_t nparams = Wosize_val(v_param_types);
+  Oid *param_types = copy_param_types(v_param_types, nparams, nparams);
+  intnat res;
+  res = PQsendPrepare(conn, stm_name, query, nparams, param_types);
+  if (param_types != NULL) caml_stat_free(param_types);
+  return res;
 }
 
 CAMLprim value PQsendPrepare_stub_bc(
-    value v_conn, value v_stm_name, value v_query)
+    value v_conn, value v_stm_name, value v_query, value v_param_types)
 {
-  return Val_int(PQsendPrepare_stub(v_conn, v_stm_name, v_query));
+  return
+    Val_int(PQsendPrepare_stub(v_conn, v_stm_name, v_query, v_param_types));
 }
 
 CAMLprim intnat PQsendQueryPrepared_stub(