OCaml client interface for Oracle Code

/*---------------------------------------------------------------------------
 * File:     $Id: occi.c 452 2006-11-02 21:49:32Z serge $
 * Purpose:  Provides Oracle client interface for OCaml language
 *
 * Author:   Serge Aleynikov <serge@hq.idt.net>
 * Created:  1-Oct-2006
 * Revision: $Rev: 452 $
 *           $Date: 2006-11-02 16:49:32 -0500 (Thu, 02 Nov 2006) $
 * Copyright (c) Serge Aleynikov.  All rights reserved.
 *-------------------------------------------------------------------------*/

#include <stdio.h>
#include <occi.h>
#include <time.h>
#include <errno.h>
#include <vector>
#include <iostream>
#include <fstream>
#include <exception>
#include <ocidfn.h>   // defines Oracle data type values
   // (http://download-east.oracle.com/docs/cd/B19306_01/appdev.102/b14294/types.htm#sthref434)

using namespace oracle::occi;

#ifdef  __cplusplus
extern "C" {
#endif

#include <assert.h>
#include <caml/mlvalues.h>
#include <caml/fail.h>
#include <caml/alloc.h>
#include <caml/memory.h>
#include <caml/callback.h>

#define RAISE_EXCEPTION(ec, em) fail_exception((ec), (em));

#define  ORA_CATCH \
    catch (SQLException &e) { \
        RAISE_EXCEPTION(e.getErrorCode(), e.getMessage().c_str()); \
    } catch (std::exception &e) { \
        RAISE_EXCEPTION(-__LINE__, e.what()); \
    }

#define CHECK_CONNECTED(con) if (!con) caml_invalid_argument("Session not connected!\n");

#define CON_PTR(data_val)    * ((Connection**) (Data_custom_val(data_val)))
#define STMT_PTR(data_val)   * ((Statement**)  (Data_custom_val(data_val)))
#define RESSET_PTR(data_val) * ((ResSet**)     (Data_custom_val(data_val)))


extern void unix_error (int errcode, char * cmdname, value arg);

/* Global OCCI environment variable used to manage memory and other resources for OCCI objects. */
static Environment* env = Environment::createEnvironment();

struct VarType {
    int pos;          /* Position of the OUT parameter in the PL/SQL block */
    int type;         /* OUT parameter type */
    VarType(int P=0, int T=0) : pos(P), type(T) {}
};

/* Note: unfortunately in the structure below we can't use a vector STL because
 * we need to know the precise size of the ResSet structure for caml_alloc_final() call,
 * so we manage the array of VarType structures manually. */
struct VarTypeArray {
    int count;
    VarType* types;
    
    VarTypeArray(int N) : count(N) { types = new VarType[N]; }
    ~VarTypeArray() {}
    VarType* operator[](const int i) { return (types + i); }
    void realloc(int N) { delete [] types; count = N; types = new VarType[N]; }
    int size() { return sizeof(VarTypeArray) + sizeof(VarType) * count; }
};
    
struct ResSet {
    ResultSet*   rs;   /* Query result set  */
    VarTypeArray vta;  /* Array of column types */
    ResSet(ResultSet* RS, int N) : rs(RS), vta(N) {}
    ~ResSet() { vta.realloc(0); }
    int size() { return sizeof(ResSet) + vta.size(); }
};

struct StrFieldPos { 
    int pos;
    int len;
    StrFieldPos(int p=0, int l=0) : pos(p), len(l) {}
};

/* Forward declarations */
void set_query_params(bool is_list, Statement* s, value params, VarTypeArray* var_info);
void finalize_connection(value connection);
void finalize_statement(value statement);
void finalize_resultset(value resultset);

#ifdef DEBUG
/*---------------------------------------------------------------------------
 * inspect_block: OCaml memory inspection troubleshooting function.
 *-------------------------------------------------------------------------*/
void margin (int n)
  { while (n-- > 0) printf(".");  return; }

void print_block (value v,int m) 
{
  int size, i;
  margin(m);
  if (Is_long(v)) { 
      printf("immediate long value (%d)\n", (int)Long_val(v));
      return;
  };
  
  printf ("pointer to a memory block: size=%d  -  ", size=Wosize_val(v));
  switch (Tag_val(v))
   {
    case Closure_tag : 
        printf("closure with %d free variables\n", size-1);
        margin(m+4); printf("code pointer: %p\n",Code_val(v)) ;
        for (i=1;i<size;i++)  print_block(Field(v,i), m+4);
        break;
    case String_tag :
        printf("string: %s (%s)\n", String_val(v),(char *) v);  
        break;
    case Double_tag:  
        printf("float: %g\n", Double_val(v));
        break;
    case Double_array_tag : 
        printf ("float array: "); 
        for (i=0; i < (int) (size/Double_wosize); i++)  printf("  %g", Double_field(v,i));
        printf("\n");
        break;
    case Abstract_tag : printf("abstract type\n"); break;
//    case Final_tag : printf("abstract finalized type\n"); break;
    default:  
        if (Tag_val(v)>=No_scan_tag) { printf("unknown tag (%d)\n", Tag_val(v)); break; }; 
        printf("structured block (tag=%d):\n",Tag_val(v));
        for (i=0;i<size;i++)  print_block(Field(v,i),m+4);
   }
  return ;
}

value inspect_block (value v)  
{
  print_block(v,4); 
  fflush(stdout); 
  return v; 
}

#endif

/*---------------------------------------------------------------------------
 * Return control to OCaml by throwing an exception.
 *-------------------------------------------------------------------------*/
void fail_exception(const int err_code, const char *reason)
{
    CAMLparam0();
    CAMLlocal1(v);

    v = caml_alloc_tuple(2);
    Store_field(v, 0, Val_long(err_code));
    Store_field(v, 1, caml_copy_string(reason));
    caml_raise_with_arg(*caml_named_value("ORA_EXCEPTION"), v);
    CAMLnoreturn;
}

void fail_batch_exception(const BatchSQLException& e)
{
    CAMLparam0();
    CAMLlocal3(v, ve, vt);
    int n;
    
    // Result is a tuple of two elements 
    v = caml_alloc_tuple(2);
    // The first element is the batch error reason
    Store_field(v, 0, caml_copy_string(e.what()));
    
    n = e.getFailedRowCount();
    // Second element is array of (row * err_code * reason)
    ve = caml_alloc(n, 0);       // Allocate array of n elements
    for (int i=0; i < n; ++i) {
        vt = caml_alloc(3, 0);   // Allocate a tuple
        SQLException err = e.getException(i);
        
        Store_field(vt, 0, Val_long(e.getRowNum(i)));                   // row number
        Store_field(vt, 1, Val_long(err.getErrorCode()));               // err_code
        Store_field(vt, 2, caml_copy_string(err.getMessage().c_str())); // reason
        
        Store_field(ve, i, vt);
    }
    Store_field(v, 1, ve);
    caml_raise_with_arg(*caml_named_value("ORA_BULK_EXCEPTION"), v);
    CAMLnoreturn;
}

/*---------------------------------------------------------------------------
 * Convert a set of integer parameters to double date (or a double date to 
 * an Oracle Date type).  The internal representation of an OCaml date is
 * The number of days since epoch, and the fractional part is the number of
 * seconds since midnight.
 *-------------------------------------------------------------------------*/
value date_time_to_value(unsigned int year, int mon, int day, int hour, int min, int sec)
{
    struct tm tm;
    time_t clock;
    double d;

    tm.tm_sec   = sec;
    tm.tm_min   = min;
    tm.tm_hour  = hour;
    tm.tm_mday  = day;
    tm.tm_mon   = mon - 1;
    tm.tm_year  = year - 1900;
    tm.tm_wday  = 0;
    tm.tm_yday  = 0;
    tm.tm_isdst = -1; /* tm.tm_isdst = Bool_val(Field(t, 8)); */
    clock = mktime(&tm);
    if (clock == (time_t) -1) unix_error(ERANGE, "mktime", (value) 0);
    d = ((double) clock) / 86400;
    return caml_copy_double(d);
}

void double_to_date(double d, Date& date)
{
    time_t dt     = (time_t)((d * 86400) + 0.5);  /* 0.5 is added to do proper rounding */
    struct tm *tm = localtime( &dt );
    date.setDate(tm->tm_year+1900, tm->tm_mon+1, tm->tm_mday, tm->tm_hour, tm->tm_min, tm->tm_sec);
}

/*---------------------------------------------------------------------------
 * Internal function to convert an internal float date representation 
 * to Oracle's 7-byte representation.
 *-------------------------------------------------------------------------*/
void double_to_ora_date(double d, char* ora_date)
{
    time_t dt     = (time_t)((d * 86400) + 0.5);  /* 0.5 is added to do proper rounding */
    struct tm *tm = localtime( &dt );
    int    year   = 1900 + tm->tm_year;
    
    *(ora_date + 0) = 100 + year / 100;  // the century byte: BC dates are < 100. AD dates are > 100
    *(ora_date + 1) = year % 100;        // the year of century byte
    *(ora_date + 2) = tm->tm_mon + 1;    // the month byte ranges from 1 to 12
    *(ora_date + 3) = tm->tm_mday;       // the date byte ranges from 1 to 31
    *(ora_date + 4) = tm->tm_hour + 1;   // the hour byte ranges from 1 to 24
    *(ora_date + 5) = tm->tm_min  + 1;   // the minute byte ranges from 1 to 60
    *(ora_date + 6) = tm->tm_sec  + 1;   // the second byte ranges from 1 to 60
}

/*---------------------------------------------------------------------------
 * Create/free an Oracle connection
 *-------------------------------------------------------------------------*/
value occi_connect(value user, value pwd, value db)
{
    CAMLparam3(user, pwd, db);
    CAMLlocal1(v);
    Connection  *con;
    
    try {
        con = env->createConnection(
                         String_val(user), String_val(pwd), String_val(db));
    } ORA_CATCH;

    /* Allocate a block with a finalization function */
    v = caml_alloc_final(1, finalize_connection, 1, 20);
    CON_PTR(v) = con;
    CAMLreturn(v);
}

value occi_disconnect(value connection)
{
    Connection* con = CON_PTR(connection);

#ifdef DEBUG
    printf("Finalize Connection called (%p)\n", con);
#endif

    if (con) {
        try { env->terminateConnection(con); } catch (...) {};
        CON_PTR(connection) = NULL;
    }
    return Val_unit;
}

/* This function will get called automatically by the garbage collector */
void finalize_connection(value connection)
{
    occi_disconnect(connection);
    return;
}

/*---------------------------------------------------------------------------
 * Commit / Rollback transaction
 *-------------------------------------------------------------------------*/
value occi_commit(value connection)
{
    Connection* con = CON_PTR(connection);
    CHECK_CONNECTED(con);
    try { con->commit(); } ORA_CATCH;
    return Val_unit;
}

value occi_rollback(value connection)
{
    Connection* con = CON_PTR(connection);
    CHECK_CONNECTED(con);
    try { con->rollback(); } ORA_CATCH;
    return Val_unit;
}

/*---------------------------------------------------------------------------
 * Create/free an Oracle statement.  
 * If a tag is used along with a statement name it is a string that can be 
 * used to cache the sql statement on a client.  Successive calls to execute
 * statement could reference the tag without calling create_statement again.
 * The maximum cache size is controlled by a parameter at the connection
 * level.  The tags are only aplicable if con->createStatement(sql, tag)
 * API is used.
 *-------------------------------------------------------------------------*/
value occi_create_statement(value connection, value sql)
{
    CAMLparam2(connection, sql);
    CAMLlocal1(v);
    Connection* con = CON_PTR(connection);
    CHECK_CONNECTED(con);
    Statement* stmt;
    
    try {
        stmt = con->createStatement(String_val(sql));
        stmt->setPrefetchRowCount(1000);
    } ORA_CATCH;
    
    /* Allocate a block with a finalization function */
    v = caml_alloc_final(1, finalize_statement, 1, 20);
    STMT_PTR(v) = stmt;
    CAMLreturn(v);
}

value occi_free_statement(value statement)
{
    CAMLparam1(statement);
    Statement* stmt = STMT_PTR(statement);

#ifdef DEBUG
    printf("Finalize Statement called (%p)\n", stmt);
#endif

    if (stmt) {
        Connection* con  = stmt->getConnection();
        try {
            stmt->disableCaching();
            con->terminateStatement(stmt);
            STMT_PTR(statement) = NULL;
        } ORA_CATCH;
    }
    CAMLreturn(Val_unit);
}

/* This function will get called automatically by the garbage collector */
void finalize_statement(value statement)
{
    occi_free_statement(statement);
    return;
}

/*---------------------------------------------------------------------------
 * Change SQL string in a statement.
 *-------------------------------------------------------------------------*/
value occi_prepare_sql(value statement, value sql)
{
    Statement* stmt = STMT_PTR(statement);
    try { stmt->setSQL(String_val(sql)); } ORA_CATCH;
    return Val_unit;
}

/*---------------------------------------------------------------------------
 * Change SQL string in a statement and set parameter types
 *-------------------------------------------------------------------------*/
value occi_prepare_plsql(value statement, value sql, value params)
{
    Statement* s = STMT_PTR(statement);
    try {
        s->setSQL(String_val(sql));
        set_query_params(true, s, params, (VarTypeArray*)NULL);
    } ORA_CATCH;
    return Val_unit;
}

/*---------------------------------------------------------------------------
 * Format a PL/SQL parameter or a field content into a CAML value of 
 * ora_value type:
 *   [Null | Int of int | Date of float | Float of float | Str of string]
 *-------------------------------------------------------------------------*/
value occi_field_to_value(Statement* stmt, ResultSet* rs, int type, int pos)
{
    CAMLparam0();
    CAMLlocal1(ve);
    bool query = stmt == NULL;

    /* Get the field/param value */ 
    if ( (query) ? rs->isNull(pos) : stmt->isNull(pos) )
        ve = Val_long(0);
    else {
        switch (type) {
            case OCCIINT:  
                /* Manual section 18.3.4. Non-constant constructors are one item blocks */
                ve = caml_alloc(1, 0);   /* Result is of type: Int of int */ 
                Store_field(ve, 0, Val_long((query) ? rs->getInt(pos) : stmt->getInt(pos)));
                break;
                
            case OCCIDATE:  
                int year;
                unsigned int mon, day, hour, min, sec;
                if (query)
                    rs->getDate(pos).getDate(year, mon, day, hour, min, sec);
                else
                    stmt->getDate(pos).getDate(year, mon, day, hour, min, sec);
                ve = caml_alloc(1, 1);   /* Result is of type: Date of float */  
                Store_field(ve, 0, date_time_to_value(year, mon, day, hour, min, sec));
                break;
                
            case OCCIDOUBLE:  
                ve = caml_alloc(1, 2);   /* Result is of type: Float of float */ 
                Store_field(ve, 0, caml_copy_double(
                    (query) ? rs->getDouble(pos) : stmt->getDouble(pos))); 
                break;

            case OCCISTRING:
                ve = caml_alloc(1, 3);   /* Result is of type: Str of string */ 
                Store_field(ve, 0, caml_copy_string(
                    (query) ? rs->getString(pos).c_str() : stmt->getString(pos).c_str())); 
                break;

            default:
                RAISE_EXCEPTION(type, "Invalid type");
        }
    }

    CAMLreturn(ve);
}

/*---------------------------------------------------------------------------
 * Format a result of a PL/SQL block call into a CAML list of values.
 * The result is a list of tuples, where the first element of a tuple is a 
 * position of an OUT variable in a PL/SQL block, 
 * and the second element is the value:
 *   [ (position:int * value:ora_value) ]
 *   ora_value = 
 *       [Null | Int of int | Date of float | Float of float | Str of string]
 *
 * Note: unfortunately OCCI doesn't provide an easy way of figuring out
 *       parameter types of individual fields or bind variables, so 
 *       we carry around the VarTypeArray structure initializaed at 
 *       prepare or execute time.
 *-------------------------------------------------------------------------*/
value statement_fields_to_list_value(Statement* stmt, VarTypeArray* vi)
{
    CAMLparam0();
    CAMLlocal4(v, vt, ve, tail);
    int n = (*vi).count - 1;
    int pos;
    
    try {
        for(int i = n; i >= 0; --i) {
            pos  = (*vi)[i]->pos;
            v  = caml_alloc(2, 0);       /* Allocate a cons cell */
            /* Each list element is a tuple */
            vt = caml_alloc(2, 0);       /* Allocate a tuple */
            /* First tuple element is the position of the field or PL/SQL OUT param */ 
            Store_field(vt, 0, Val_long(pos));

            ve = occi_field_to_value(stmt, NULL, (*vi)[i]->type, (*vi)[i]->pos);
            
            Store_field(vt, 1, ve);   /* Second element is the value */
            Store_field(v,  0, vt);   /* Store tuple in the cons cell */
            
            /* Store pointer to the tail in the second cons cell of a list. 
             * Val_long(0) means the end of the list */ 
            Store_field(v,  1, (i == (*vi).count-1) ? Val_long(0) : tail);   
            tail = v;
        }
    } ORA_CATCH;

    CAMLreturn(v);
}

/*---------------------------------------------------------------------------
 * Internal function: formats result of a fetch into a CAML array of values.
 * The result is an OCAML array of values: 
 *   [| value:ora_value |]
 *   ora_value = 
 *     [Null | Int of int | Date of float | Float of float | Str of string]
 *-------------------------------------------------------------------------*/
value fetch_to_array(ResultSet* rs, VarTypeArray* vi)
{
    CAMLparam0();
    CAMLlocal2(v, ve);
    int n = (*vi).count;
    
    try {
        /* Create an array of n values */
        v = caml_alloc(n, 0);
        for(int i = 0; i < n; ++i) {
            ve = occi_field_to_value(NULL, rs, (*vi)[i]->type, (*vi)[i]->pos);
            Store_field(v, i, ve);   /* Store i-th element of resulting array */
        }
    } ORA_CATCH;

    CAMLreturn(v);
}


/*---------------------------------------------------------------------------
 * Internal function that sets the value of the N-th parameter to the 
 * content of CAML's Value.
 *-------------------------------------------------------------------------*/
void set_param_from_value(Statement* s, int pos, value v, std::vector<VarType>* outpv)
{
    if (Is_long(v)) {
        /* This is a Constant constructor.  Immediate integer value is its type identifier index */
        switch (Long_val(v)) {
            case 0: /* Null_int */      s->setNull(pos, OCCIINT);    break;
            case 1: /* Null_date */     s->setNull(pos, OCCIDATE);   break;
            case 2: /* Null_float */    s->setNull(pos, OCCIDOUBLE); break;
            case 3: /* Null_str */      s->setNull(pos, OCCISTRING); break;
            case 4: /* Var_out_int */   s->registerOutParam(pos, OCCIINT);
                                        assert(outpv);
                                        outpv->push_back(VarType(pos, OCCIINT)); break;
            case 5: /* Var_out_date */  s->registerOutParam(pos, OCCIDATE);
                                        assert(outpv);
                                        outpv->push_back(VarType(pos, OCCIDATE)); break;
            case 6: /* Var_out_float */ s->registerOutParam(pos, OCCIDOUBLE);
                                        assert(outpv);
                                        outpv->push_back(VarType(pos, OCCIDOUBLE)); break;
            default:
                RAISE_EXCEPTION(Tag_val(v), "Unknown field type");
        }
    } else {
        /* This is a Non-Constant constructor.  The argument's tag is the type identifier index.
         * See OCaml's comumn_type variant type definition in ora.ml */
        assert(Wosize_val(v) == 1);
        
        switch (Tag_val(v)) {
            case 0:  /* Var_int of int */
                s->setInt(pos, Long_val(Field(v,0))); break;
            case 1:  /* Var_date of float */
                {
                  Date d(env);
                  double_to_date(Double_val(Field(v,0)), d);
                  s->setDate(pos, d);
                }
                break;
            case 2:  /* Var_float of float */
                s->setDouble(pos, Double_val(Field(v,0))); break;
            case 3:  /* Var_str of string */
                s->setString(pos, String_val(Field(v,0))); break;
            case 4:  /* Var_out_str of int */
                s->registerOutParam(pos, OCCISTRING, Long_val(Field(v,0))); 
                assert(outpv);
                outpv->push_back(VarType(pos, OCCISTRING)); break;
            default:
                RAISE_EXCEPTION(Tag_val(v), "Unknown field type");
        }
    }
}


/*---------------------------------------------------------------------------
 * Internal function used to set Oracle bind parameters from a list of OCaml
 * values. The SQL statement must be set by the time this function is called.
 * The var_info vector (if not NULL) will be assigned a vector of OUT
 * parameters indexes and types.
 *-------------------------------------------------------------------------*/
void set_query_params(bool is_list, Statement* s, value params, VarTypeArray* var_info)
{
    if (params == Val_long(0))  /* No parameters found */
        return;
    
    std::vector<VarType> outpv;
    
    try {
        if (is_list) {
            value ve, vp;
            /* vp will traverse the list of parameters.
             * The first element of the list is (pos:int * param_type), the second is either 
             * 0 (the end of list) or a value containing the tail of the list */
            for(vp = params; vp != Val_long(0); vp = Field(vp, 1)) {
                /* Each element is a cons */
                assert(Wosize_val(vp) == 2);
                ve = Field(vp, 0);
                /* Each element of a cons is a tuple of size 2 */
                assert(Wosize_val(ve) == 2);

                /* Field(ve, 0) - parameter position
                 * Field(ve, 1) - parameter value */
                set_param_from_value(s, Long_val(Field(ve, 0)), Field(ve, 1), &outpv);
            }

        } else {
            int n = Wosize_val(params);
            
            for (int i=0; i < n; ++i) {
                set_param_from_value(s, i+1, Field(params, i), &outpv);
            }
        }
    } ORA_CATCH;

    if (var_info != NULL) {
        int n = outpv.size();
        var_info->realloc(n);
        
        for(int i=0; i < n; ++i) {
            *(*var_info)[i] = outpv[i];
        }
    }
}

/*---------------------------------------------------------------------------
 * Describe table structure
 *-------------------------------------------------------------------------*/
value occi_describe(value connection, value table)
{
    CAMLparam2(connection, table);
    CAMLlocal2(result, v);
    Connection* con = (Connection*) Field(connection, 1);
    int n, type, size;
    MetaData meta = con->getMetaData(String_val(table), MetaData::PTYPE_TABLE);
    std::vector<MetaData> columns = meta.getVector(MetaData::ATTR_LIST_COLUMNS);
    size = (int) columns.size();

    result = caml_alloc(size, 0);
    for (int i=0; i < size; ++i) {
        MetaData *col = &columns[i];
        type = col->getInt(MetaData::ATTR_DATA_TYPE);
        switch (type) {
            case OCCI_SQLT_STR:
            case OCCI_SQLT_CHR:
            case OCCI_SQLT_VCS:
            case OCCI_SQLT_LNG:
                n = col->getInt(MetaData::ATTR_CHAR_SIZE); break;
            default:
                n = col->getInt(MetaData::ATTR_PRECISION);
        }
        v = caml_alloc_tuple(5);
        Store_field(v, 0, caml_copy_string((col->getString(MetaData::ATTR_NAME)).c_str()));
        Store_field(v, 1, Val_long(type));
        Store_field(v, 2, Val_long(n));
        Store_field(v, 3, Val_long(col->getInt(MetaData::ATTR_SCALE)));
        Store_field(v, 4, Val_bool(! col->getBoolean(MetaData::ATTR_IS_NULL)));
        Store_field(result, i, v);
    }
    CAMLreturn(result);
}

/*---------------------------------------------------------------------------
 * Fetch the next record from an executed query. Result is a list of 
 * field values of type:
 * [ Null | Int of int | Date of float | Float of float | Str of string ]
 *-------------------------------------------------------------------------*/
value occi_fetch(value resultset)
{
    ResSet* rset = RESSET_PTR(resultset);
    Statement* s;
    
    if (!rset || !rset->rs) {
        caml_raise_end_of_file();
    } else {
        ResultSet* rs = rset->rs;

#ifdef DEBUG
        printf("Fetch ResultSet address (%p, %p).\n", rset, rs);
#endif
        if (! rs->next()) {
            // FIXME: Is it OK to free the resultset when the statement is freed?
            s = rs->getStatement();
            if (s)
                try { s->closeResultSet(rs); } catch (...) {}
            delete rset;
            RESSET_PTR(resultset) = NULL;
            caml_raise_end_of_file();
        }
        return fetch_to_array(rs, &rset->vta);
    }
}

/*---------------------------------------------------------------------------
 * Execute a parameterless SQL statement and optionally commit.
 * Output is a list of (int * ora_value) tuples.
 *-------------------------------------------------------------------------*/
value occi_execute(value statement, value sql, value params)
{
    Statement* stmt  = STMT_PTR(statement);
    VarTypeArray vt(0);

    try {
        stmt->setSQL(String_val(sql));
        set_query_params(true, stmt, params, &vt);
#ifdef DEBUG
        printf("%s: SQL: %s\n", __FUNCTION__, stmt->getSQL().c_str());
#endif
        stmt->executeUpdate();

    } ORA_CATCH;

    if (stmt->status() == Statement::RESULT_SET_AVAILABLE)
        return fetch_to_array(stmt->getResultSet(), &vt);
    else
        return statement_fields_to_list_value(stmt, &vt);
}

/*---------------------------------------------------------------------------
 * Execute an SQL query.  If a statement stmt was created earlier by passing
 * non-empty SQL string, the 'sql' parameter in this call may be empty.
 *-------------------------------------------------------------------------*/
value occi_execute_query(value statement, value sql)
{
    int n;
    int size, sz;
    ResSet* rset;
    CAMLparam2(statement, sql);
    CAMLlocal3(names, v, result);
    Statement* stmt = STMT_PTR(statement);
    ResultSet *rs;
    
    try {
        rs = stmt->executeQuery(String_val(sql));
    } ORA_CATCH;

    // If there are no resulting rows, throw an exception
    // or else get a vector of field columns and allocate 
    // the custom memory structure holding a reference
    // to the recordset and field types.
    if (rs->status() == ResultSet::END_OF_FETCH) {
        caml_raise_end_of_file();
    } 
    
    std::vector<MetaData> columns = rs->getColumnListMetaData();
    size = (int) columns.size();

    if (! (rset = new ResSet(rs, size))) {
        caml_raise_out_of_memory();
    }
    
    // Build an array of FieldNames
    names = caml_alloc(size, 0);
    for(int i = 0; i < size; ++i) {
        Store_field(names, i, copy_string(
            (columns[i].getString(MetaData::ATTR_NAME)).c_str()));
    }

    // Build an array of FieldTypes
    for(int i = 0; i < size; ++i) {
        n = columns[i].getInt(MetaData::ATTR_DATA_TYPE);
        switch (n) {
            case OCCI_SQLT_NUM:
                n = (columns[i].getInt(MetaData::ATTR_SCALE)) ? OCCIDOUBLE : OCCIINT; break;
            case OCCI_SQLT_DAT:
                n = OCCIDATE; break;
            case OCCI_SQLT_STR:
            case OCCI_SQLT_CHR:
            case OCCI_SQLT_VCS:
            case OCCI_SQLT_AFC:
            case OCCI_SQLT_LNG:
                n = OCCISTRING; break;
            default: 
                RAISE_EXCEPTION(n, "Unsupported field type");
        }
        *rset->vta[i] = VarType(i+1, n);
    }

    sz = rset->size();
#ifdef DEBUG
    printf("ResultSet address (%p, %p). Size=%d\n", rset, rs, sz);
#endif

    v = caml_alloc_final(sz / sizeof (int), finalize_resultset, 1, 50);
    RESSET_PTR(v) = rset;
    // Result is of type:  ( ResultSet * [| FieldNames |] )
    result = caml_alloc_tuple(2);
    Store_field(result, 0, v);
    Store_field(result, 1, names);
    CAMLreturn(result);
}

/* This function will get called automatically by the garbage collector */
void finalize_resultset(value resultset)
{
    ResSet* rset = RESSET_PTR(resultset);

#ifdef DEBUG
    printf("Finalize ResultSet called (%p, %p)\n", rset, (rset) ? rset->rs : NULL);
#endif

    if (rset) {
        delete rset;
    }
    return;
}

/*---------------------------------------------------------------------------
 * Execute a parameterized query
 *-------------------------------------------------------------------------*/
value occi_execute_param_query(value statement, value sql, value params)
{
    Statement* stmt = STMT_PTR(statement);

    if (string_length(sql) > 0)
        try { stmt->setSQL(String_val(sql)); } ORA_CATCH;
 
    set_query_params(false, stmt, params, (VarTypeArray*)NULL);
  
    return occi_execute_query(statement, (value)"");
}


/*---------------------------------------------------------------------------
 * Execute an Insert/Update
 *-------------------------------------------------------------------------*/
value occi_execute_update(value statement, value params)
{
    Statement* stmt = STMT_PTR(statement);
    
    try { 
        set_query_params(false, stmt, params, (VarTypeArray*)NULL);
        stmt->executeUpdate(); 
    } ORA_CATCH;

    return Val_long((stmt->status() == Statement::UPDATE_COUNT_AVAILABLE) ? 
                    stmt->getUpdateCount() : 0);
}
 
/*---------------------------------------------------------------------------
 * Internal function that gets the type of each field and max size of the 
 * string columns.
 *-------------------------------------------------------------------------*/
int get_array_param_type(value v)
{
    int type;
    
    if (Is_long(v)) {
        /* This is a Constant constructor.  Immediate integer value is its type identifier index */
        switch (Long_val(v)) {
            case 0: /* Null_int */   type = OCCIINT;    break;
            case 1: /* Null_date */  type = OCCIDATE;   break;
            case 2: /* Null_float */ type = OCCIDOUBLE; break;
            case 3: /* Null_str */   type = OCCISTRING; break;
            default:
                RAISE_EXCEPTION(Tag_val(v), "Unknown field type");
        }
    } else {
        /* This is a Non-Constant constructor.  The argument's tag is the type identifier index.
         * See OCaml's comumn_type variant type definition in ora.ml */
        switch (Tag_val(v)) {
            case 0:  /* Var_int of int */     type = OCCIINT;    break;
            case 1:  /* Var_date of float */  type = OCCIDATE;   break;
            case 2:  /* Var_float of float */ type = OCCIDOUBLE; break;
            case 3:  /* Var_str of string */  type = OCCISTRING; break;
            default:
                RAISE_EXCEPTION(Tag_val(v), "Unknown field type");
        }
    }
    return type;
}

/*---------------------------------------------------------------------------
 * Get max lengths of string fields of the current statement
 *-------------------------------------------------------------------------*/
void get_max_string_lengths(value array, std::vector<StrFieldPos>& idxs)
{
    value v;
    
    for(std::vector<StrFieldPos>::iterator it=idxs.begin(); it != idxs.end(); ++it)
    {
        assert(Wosize_val(array) > (unsigned int)it->pos);
        v = Field(array, it->pos);
        if (v != Val_long(3)) { // Null_str
            v = Field(v, 0);  // This type is "Var_str of string", so we need one more indirection.
            assert(Tag_val(v) == String_tag);
            int n = it->len;
            int m = string_length(v);
            it->len = (m > n) ? m : n;
        }
    }
}

/*---------------------------------------------------------------------------
 * Perform an INSERT/UPDATE/DELETE operation given a list of parameters
 * where each parameter is an array of homogeneous values:
 *   [ [| param_val |] ]
 *   param_val = 
 *       [Null_int | Null_date | Null_float | Null_str | Var_int of int | 
 *        Var_date of float | Var_float of float | Var_str of string]
 *-------------------------------------------------------------------------*/
value occi_execute_array(value statement, value sql, value params)
{
  
    Statement* stmt = STMT_PTR(statement);
    value ve, vp;
    int length = 0, size;
    
    if (string_length(sql) > 0)
        try { stmt->setSQL(String_val(sql)); } ORA_CATCH;
 
    if (params == Val_long(0))
        return Val_long(0);
    else {
        // Store the number of fields in each record in the 'size' variable.  Note that 
        // we expect all arrays to be of the same shape.
        assert(Wosize_val(params) == 2);
        vp   = Field(params, 0);
        size = Wosize_val(vp);
    }

    // str_field_pos vector will hold positions of string fields
    std::vector<StrFieldPos> str_field_idxs;
    
    // This loop does type checking, and stores all string fields
    for(int i=0; i < size; ++i) {
        int type = get_array_param_type(Field(vp, i));
        if (type == OCCISTRING)
            str_field_idxs.push_back(StrFieldPos(i, 0));
    }

    // Traverse the list of parameter arrays to determine the total
    // number of records being inserted and store it in the 'length' variable.
    // Also get max lengths of all string fields
    for(vp = params; vp != Val_long(0); vp = Field(vp, 1), ++length) {
        /* Each element is a cons containing an array of fields */
        ve = Field(vp, 0);
        get_max_string_lengths(ve, str_field_idxs);
    }

    // Set max sizes of string columns
    for(std::vector<StrFieldPos>::iterator it=str_field_idxs.begin(); it != str_field_idxs.end(); ++it)
        try { stmt->setMaxParamSize(it->pos+1, it->len+1); } ORA_CATCH;
    
    try { stmt->setMaxIterations(length); } ORA_CATCH;
    
    // By now we know how many records are being inserted (length) and 
    // the max sizes of all fields.  Allocate sufficient memory buffers, and 
    // copy data.
    for(vp = params; vp != Val_long(0); vp = Field(vp, 1)) {
        ve = Field(vp, 0);  // Get the tuple
        
        for(int i=0; i < size; ++i) 
            try { 
                set_param_from_value(stmt, i+1, Field(ve, i), (std::vector<VarType>*)NULL); 
            } ORA_CATCH;
        
        if (Field(vp, 1) != Val_long(0)) 
           try { stmt->addIteration(); } ORA_CATCH;
    }

    stmt->setBatchErrorMode(true);
    try {
        stmt->executeUpdate(); 
        stmt->setBatchErrorMode(false);
    } catch (BatchSQLException &e) {
        stmt->setBatchErrorMode(false);
        fail_batch_exception(e);
    }

    return Val_long((stmt->status() == Statement::UPDATE_COUNT_AVAILABLE) ?
                    stmt->getUpdateCount() : 0);
}

#ifdef  __cplusplus
}
#endif