Open-ILS/src/c-apps/oils_cstore.c

   1 /**
   2         @file oils_cstore.c
   3         @brief As a server, perform database operations at the request of clients.
   4 */
   5
   6 #include <ctype.h>
   7 #include "opensrf/osrf_application.h"
   8 #include "opensrf/osrf_settings.h"
   9 #include "opensrf/osrf_message.h"
  10 #include "opensrf/utils.h"
  11 #include "opensrf/osrf_json.h"
  12 #include "opensrf/log.h"
  13 #include "openils/oils_utils.h"
  14 #include "openils/oils_constants.h"
  15 #include <dbi/dbi.h>
  16
  17 #include <time.h>
  18 #include <stdlib.h>
  19 #include <string.h>
  20 #include <unistd.h>
  21
  22 #ifdef RSTORE
  23 #  define MODULENAME "open-ils.reporter-store"
  24 #  define ENFORCE_PCRUD 0
  25 #else
  26 #  ifdef PCRUD
  27 #    define MODULENAME "open-ils.pcrud"
  28 #    define ENFORCE_PCRUD 1
  29 #  else
  30 #    define MODULENAME "open-ils.cstore"
  31 #    define ENFORCE_PCRUD 0
  32 #  endif
  33 #endif
  34
  35 // The next four macros are OR'd together as needed to form a set
  36 // of bitflags.  SUBCOMBO enables an extra pair of parentheses when
  37 // nesting one UNION, INTERSECT or EXCEPT inside another.
  38 // SUBSELECT tells us we're in a subquery, so don't add the
  39 // terminal semicolon yet.
  40 #define SUBCOMBO    8
  41 #define SUBSELECT   4
  42 #define DISABLE_I18N    2
  43 #define SELECT_DISTINCT 1
  44
  45 #define AND_OP_JOIN     0
  46 #define OR_OP_JOIN      1
  47
  48 struct ClassInfoStruct;
  49 typedef struct ClassInfoStruct ClassInfo;
  50
  51 #define ALIAS_STORE_SIZE 16
  52 #define CLASS_NAME_STORE_SIZE 16
  53
  54 struct ClassInfoStruct {
  55         char* alias;
  56         char* class_name;
  57         char* source_def;
  58         osrfHash* class_def;      // Points into IDL
  59         osrfHash* fields;         // Points into IDL
  60         osrfHash* links;          // Points into IDL
  61
  62         // The remaining members are private and internal.  Client code should not
  63         // access them directly.
  64
  65         ClassInfo* next;          // Supports linked list of joined classes
  66         int in_use;               // boolean
  67
  68         // We usually store the alias and class name in the following arrays, and
  69         // point the corresponding pointers at them.  When the string is too big
  70         // for the array (which will probably never happen in practice), we strdup it.
  71
  72         char alias_store[ ALIAS_STORE_SIZE + 1 ];
  73         char class_name_store[ CLASS_NAME_STORE_SIZE + 1 ];
  74 };
  75
  76 struct QueryFrameStruct;
  77 typedef struct QueryFrameStruct QueryFrame;
  78
  79 struct QueryFrameStruct {
  80         ClassInfo core;
  81         ClassInfo* join_list;  // linked list of classes joined to the core class
  82         QueryFrame* next;      // implements stack as linked list
  83         int in_use;            // boolean
  84 };
  85
  86 static int timeout_needs_resetting;
  87 static time_t time_next_reset;
  88
  89 int osrfAppChildInit();
  90 int osrfAppInitialize();
  91 void osrfAppChildExit();
  92
  93 static int verifyObjectClass ( osrfMethodContext*, const jsonObject* );
  94
  95 static void setXactId( osrfMethodContext* ctx );
  96 static inline const char* getXactId( osrfMethodContext* ctx );
  97 static inline void clearXactId( osrfMethodContext* ctx );
  98
  99 int beginTransaction ( osrfMethodContext* );
 100 int commitTransaction ( osrfMethodContext* );
 101 int rollbackTransaction ( osrfMethodContext* );
 102
 103 int setSavepoint ( osrfMethodContext* );
 104 int releaseSavepoint ( osrfMethodContext* );
 105 int rollbackSavepoint ( osrfMethodContext* );
 106
 107 int doJSONSearch ( osrfMethodContext* );
 108
 109 int dispatchCRUDMethod( osrfMethodContext* ctx );
 110 static int doSearch( osrfMethodContext* ctx );
 111 static int doIdList( osrfMethodContext* ctx );
 112 static int doCreate( osrfMethodContext* ctx );
 113 static int doRetrieve( osrfMethodContext* ctx );
 114 static int doUpdate( osrfMethodContext* ctx );
 115 static int doDelete ( osrfMethodContext* ctx );
 116 static jsonObject* doFieldmapperSearch ( osrfMethodContext* ctx, osrfHash* class_meta,
 117                 jsonObject* where_hash, jsonObject* query_hash, int* err );
 118 static jsonObject* oilsMakeFieldmapperFromResult( dbi_result, osrfHash* );
 119 static jsonObject* oilsMakeJSONFromResult( dbi_result );
 120
 121 static char* searchSimplePredicate ( const char* op, const char* class_alias,
 122                                 osrfHash* field, const jsonObject* node );
 123 static char* searchFunctionPredicate ( const char*, osrfHash*, const jsonObject*, const char* );
 124 static char* searchFieldTransform ( const char*, osrfHash*, const jsonObject*);
 125 static char* searchFieldTransformPredicate ( const ClassInfo*, osrfHash*, const jsonObject*, const char* );
 126 static char* searchBETWEENPredicate ( const char*, osrfHash*, const jsonObject* );
 127 static char* searchINPredicate ( const char*, osrfHash*,
 128                                                                  jsonObject*, const char*, osrfMethodContext* );
 129 static char* searchPredicate ( const ClassInfo*, osrfHash*, jsonObject*, osrfMethodContext* );
 130 static char* searchJOIN ( const jsonObject*, const ClassInfo* left_info );
 131 static char* searchWHERE ( const jsonObject*, const ClassInfo*, int, osrfMethodContext* );
 132 static char* buildSELECT ( jsonObject*, jsonObject*, osrfHash*, osrfMethodContext* );
 133 char* buildQuery( osrfMethodContext* ctx, jsonObject* query, int flags );
 134
 135 char* SELECT ( osrfMethodContext*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, jsonObject*, int );
 136
 137 void userDataFree( void* );
 138 static void sessionDataFree( char*, void* );
 139 static char* getRelation( osrfHash* );
 140 static int str_is_true( const char* str );
 141 static int obj_is_true( const jsonObject* obj );
 142 static const char* json_type( int code );
 143 static const char* get_primitive( osrfHash* field );
 144 static const char* get_datatype( osrfHash* field );
 145 static int is_identifier( const char* s);
 146 static int is_good_operator( const char* op );
 147 static void pop_query_frame( void );
 148 static void push_query_frame( void );
 149 static int add_query_core( const char* alias, const char* class_name );
 150 static inline ClassInfo* search_alias( const char* target );
 151 static ClassInfo* search_all_alias( const char* target );
 152 static ClassInfo* add_joined_class( const char* alias, const char* classname );
 153 static void clear_query_stack( void );
 154
 155 static const jsonObject* verifyUserPCRUD( osrfMethodContext* );
 156 static int verifyObjectPCRUD( osrfMethodContext*, const jsonObject* );
 157 static char* org_tree_root( osrfMethodContext* ctx );
 158 static jsonObject* single_hash( const char* key, const char* value );
 159
 160 static const int enforce_pcrud = ENFORCE_PCRUD;     // Boolean
 161 static const char modulename[] = MODULENAME;
 162
 163 static int child_initialized = 0;   /* boolean */
 164
 165 static dbi_conn writehandle; /* our MASTER db connection */
 166 static dbi_conn dbhandle; /* our CURRENT db connection */
 167 //static osrfHash * readHandles;
 168 static jsonObject* const jsonNULL = NULL; //
 169 static int max_flesh_depth = 100;
 170
 171 // The following points to the top of a stack of QueryFrames.  It's a little
 172 // confusing because the top level of the query is at the bottom of the stack.
 173 static QueryFrame* curr_query = NULL;
 174
 175 /**
 176         @brief Disconnect from the database.
 177
 178         This function is called when the server drone is about to terminate.
 179 */
 180 void osrfAppChildExit() {
 181         osrfLogDebug(OSRF_LOG_MARK, "Child is exiting, disconnecting from database...");
 182
 183         int same = 0;
 184         if (writehandle == dbhandle)
 185                 same = 1;
 186
 187         if (writehandle) {
 188                 dbi_conn_query(writehandle, "ROLLBACK;");
 189                 dbi_conn_close(writehandle);
 190                 writehandle = NULL;
 191         }
 192         if (dbhandle && !same)
 193                 dbi_conn_close(dbhandle);
 194
 195         // XXX add cleanup of readHandles whenever that gets used
 196
 197         return;
 198 }
 199
 200 /**
 201         @brief Initialize the application.
 202         @return Zero if successful, or non-zero if not.
 203
 204         Load the IDL file into an internal data structure for future reference.  Each non-virtual
 205         class in the IDL corresponds to a table or view in the database, or to a subquery defined
 206         in the IDL.  Ignore all virtual tables and virtual fields.
 207
 208         Register a number of methods, some of them general-purpose and others specific for
 209         particular classes.
 210
 211         The name of the application is given by the MODULENAME macro, whose value depends on
 212         conditional compilation.  The method names also incorporate MODULENAME, followed by a
 213         dot, as a prefix.  Some methods are registered or not registered depending on whether
 214         the PCRUD macro is defined, and on whether the IDL includes permacrud entries for the
 215         class and method.
 216
 217         The general-purpose methods are as follows (minus their MODULENAME prefixes):
 218
 219         - json_query (not registered for PCRUD)
 220         - transaction.begin
 221         - transaction.commit
 222         - transaction.rollback
 223         - savepoint.set
 224         - savepoint.release
 225         - savepoint.rollback
 226
 227         For each non-virtual class, create up to eight class-specific methods:
 228
 229         - create    (not for readonly classes)
 230         - retrieve
 231         - update    (not for readonly classes)
 232         - delete    (not for readonly classes
 233         - search    (atomic and non-atomic versions)
 234         - id_list   (atomic and non-atomic versions)
 235
 236         For PCRUD, the full method names follow the pattern "MODULENAME.method_type.classname".
 237         Otherwise they follow the pattern "MODULENAME.direct.XXX.method_type", where XXX is the
 238         fieldmapper name from the IDL, with every run of one or more consecutive colons replaced
 239         by a period.  In addition, the names of atomic methods have a suffix of ".atomic".
 240
 241         This function is called when the registering the application, and is executed by the
 242         listener before spawning the drones.
 243 */
 244 int osrfAppInitialize() {
 245
 246         osrfLogInfo(OSRF_LOG_MARK, "Initializing the CStore Server...");
 247         osrfLogInfo(OSRF_LOG_MARK, "Finding XML file...");
 248
 249         if (!oilsIDLInit( osrf_settings_host_value("/IDL") ))
 250                 return 1; /* return non-zero to indicate error */
 251
 252         growing_buffer* method_name = buffer_init(64);
 253
 254         if( ! enforce_pcrud ) {
 255                 // Generic search thingy (not for PCRUD)
 256                 buffer_add( method_name, modulename );
 257                 buffer_add( method_name, ".json_query" );
 258                 osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR( method_name ),
 259                         "doJSONSearch", "", 1, OSRF_METHOD_STREAMING );
 260         }
 261
 262         // first we register all the transaction and savepoint methods
 263         buffer_reset(method_name);
 264         OSRF_BUFFER_ADD(method_name, modulename );
 265         OSRF_BUFFER_ADD(method_name, ".transaction.begin");
 266         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR( method_name ),
 267                         "beginTransaction", "", 0, 0 );
 268
 269         buffer_reset(method_name);
 270         OSRF_BUFFER_ADD(method_name, modulename );
 271         OSRF_BUFFER_ADD(method_name, ".transaction.commit");
 272         osrfAppRegisterMethod( MODULENAME, OSRF_BUFFER_C_STR(method_name),
 273                         "commitTransaction", "", 0, 0 );
 274
 275         buffer_reset(method_name);
 276         OSRF_BUFFER_ADD(method_name, modulename );
 277         OSRF_BUFFER_ADD(method_name, ".transaction.rollback");
 278         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 279                         "rollbackTransaction", "", 0, 0 );
 280
 281         buffer_reset(method_name);
 282         OSRF_BUFFER_ADD(method_name, modulename );
 283         OSRF_BUFFER_ADD(method_name, ".savepoint.set");
 284         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 285                         "setSavepoint", "", 1, 0 );
 286
 287         buffer_reset(method_name);
 288         OSRF_BUFFER_ADD(method_name, modulename );
 289         OSRF_BUFFER_ADD(method_name, ".savepoint.release");
 290         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 291                         "releaseSavepoint", "", 1, 0 );
 292
 293         buffer_reset(method_name);
 294         OSRF_BUFFER_ADD(method_name, modulename );
 295         OSRF_BUFFER_ADD(method_name, ".savepoint.rollback");
 296         osrfAppRegisterMethod( modulename, OSRF_BUFFER_C_STR(method_name),
 297                         "rollbackSavepoint", "", 1, 0 );
 298
 299         static const char* global_method[] = {
 300                 "create",
 301                 "retrieve",
 302                 "update",
 303                 "delete",
 304                 "search",
 305                 "id_list"
 306         };
 307         const int global_method_count
 308                 = sizeof( global_method ) / sizeof ( global_method[0] );
 309
 310         unsigned long class_count = osrfHashGetCount( oilsIDL() );
 311         osrfLogDebug(OSRF_LOG_MARK, "%lu classes loaded", class_count );
 312         osrfLogDebug(OSRF_LOG_MARK,
 313                 "At most %lu methods will be generated",
 314                 (unsigned long) (class_count * global_method_count) );
 315
 316         osrfHashIterator* class_itr = osrfNewHashIterator( oilsIDL() );
 317         osrfHash* idlClass = NULL;
 318
 319         // For each class in the IDL...
 320         while( (idlClass = osrfHashIteratorNext( class_itr ) ) ) {
 321
 322                 const char* classname = osrfHashIteratorKey( class_itr );
 323                 osrfLogInfo(OSRF_LOG_MARK, "Generating class methods for %s", classname);
 324
 325                 if (!osrfStringArrayContains( osrfHashGet(idlClass, "controller"), modulename )) {
 326                         osrfLogInfo(OSRF_LOG_MARK, "%s is not listed as a controller for %s, moving on",
 327                                         modulename, classname);
 328                         continue;
 329                 }
 330
 331                 if ( str_is_true( osrfHashGet(idlClass, "virtual") ) ) {
 332                         osrfLogDebug(OSRF_LOG_MARK, "Class %s is virtual, skipping", classname );
 333                         continue;
 334                 }
 335
 336                 // Look up some other attributes of the current class
 337                 const char* idlClass_fieldmapper = osrfHashGet(idlClass, "fieldmapper");
 338                 if( !idlClass_fieldmapper ) {
 339                         osrfLogDebug( OSRF_LOG_MARK, "Skipping class \"%s\"; no fieldmapper in IDL",
 340                                         classname );
 341                         continue;
 342                 }
 343
 344                 osrfHash* idlClass_permacrud = NULL;
 345                 if( enforce_pcrud ) {
 346                         // For PCRUD, ignore classes with no permacrud section
 347                         idlClass_permacrud = osrfHashGet(idlClass, "permacrud");
 348                         if (!idlClass_permacrud) {
 349                                 osrfLogDebug( OSRF_LOG_MARK,
 350                                         "Skipping class \"%s\"; no permacrud in IDL", classname );
 351                                 continue;
 352                         }
 353                 }
 354
 355                 const char* readonly = osrfHashGet(idlClass, "readonly");
 356
 357                 int i;
 358                 for( i = 0; i < global_method_count; ++i ) {  // for each global method
 359                         const char* method_type = global_method[ i ];
 360                         osrfLogDebug(OSRF_LOG_MARK,
 361                                 "Using files to build %s class methods for %s", method_type, classname);
 362
 363                         if( enforce_pcrud ) {
 364                                 // Treat "id_list" or "search" as forms of "retrieve"
 365                                 const char* tmp_method = method_type;
 366                                 if ( *tmp_method == 'i' || *tmp_method == 's') {  // "id_list" or "search"
 367                                         tmp_method = "retrieve";
 368                                 }
 369                                 // Skip this method if there is no permacrud entry for it
 370                                 if (!osrfHashGet( idlClass_permacrud, tmp_method ))
 371                                         continue;
 372                         }
 373
 374                         // No create, update, or delete methods for a readonly class
 375                         if ( str_is_true( readonly )
 376                                 && ( *method_type == 'c' || *method_type == 'u' || *method_type == 'd') )
 377                                 continue;
 378
 379                         buffer_reset( method_name );
 380
 381                         // Build the method name
 382                         if( enforce_pcrud ) {
 383                                 // For PCRUD: MODULENAME.method_type.classname
 384                                 buffer_fadd(method_name, "%s.%s.%s", modulename, method_type, classname);
 385                         } else {
 386                                 // For non-PCRUD: MODULENAME.MODULENAME.direct.XXX.method_type
 387                                 // where XXX is the fieldmapper name from the IDL, with every run of
 388                                 // one or more consecutive colons replaced by a period.
 389                                 char* st_tmp = NULL;
 390                                 char* part = NULL;
 391                                 char* _fm = strdup( idlClass_fieldmapper );
 392                                 part = strtok_r(_fm, ":", &st_tmp);
 393
 394                                 buffer_fadd(method_name, "%s.direct.%s", modulename, part);
 395
 396                                 while ((part = strtok_r(NULL, ":", &st_tmp))) {
 397                                         OSRF_BUFFER_ADD_CHAR(method_name, '.');
 398                                         OSRF_BUFFER_ADD(method_name, part);
 399                                 }
 400                                 OSRF_BUFFER_ADD_CHAR(method_name, '.');
 401                                 OSRF_BUFFER_ADD(method_name, method_type);
 402                                 free(_fm);
 403                         }
 404
 405                         // For an id_list or search method we specify the OSRF_METHOD_STREAMING option.
 406                         // The consequence is that we implicitly create an atomic method in addition to
 407                         // the usual non-atomic method.
 408                         int flags = 0;
 409                         if (*method_type == 'i' || *method_type == 's') {  // id_list or search
 410                                 flags = flags | OSRF_METHOD_STREAMING;
 411                         }
 412
 413                         osrfHash* method_meta = osrfNewHash();
 414                         osrfHashSet( method_meta, idlClass, "class");
 415                         osrfHashSet( method_meta, buffer_data( method_name ), "methodname" );
 416                         osrfHashSet( method_meta, strdup(method_type), "methodtype" );
 417
 418                         // Register the method, with a pointer to an osrfHash to tell the method
 419                         // its name, type, and class.
 420                         osrfAppRegisterExtendedMethod(
 421                                 modulename,
 422                                 OSRF_BUFFER_C_STR( method_name ),
 423                                 "dispatchCRUDMethod",
 424                                 "",
 425                                 1,
 426                                 flags,
 427                                 (void*)method_meta
 428                         );
 429
 430                 } // end for each global method
 431         } // end for each class in IDL
 432
 433         buffer_free( method_name );
 434         osrfHashIteratorFree( class_itr );
 435
 436         return 0;
 437 }
 438
 439 /**
 440         @brief Get a table name, view name, or subquery for use in a FROM clause.
 441         @param class Pointer to the IDL class entry.
 442         @return A table name, a view name, or a subquery in parentheses.
 443
 444         In some cases the IDL defines a class, not with a table name or a view name, but with
 445         a SELECT statement, which may be used as a subquery.
 446 */
 447 static char* getRelation( osrfHash* class ) {
 448
 449         char* source_def = NULL;
 450         const char* tabledef = osrfHashGet(class, "tablename");
 451
 452         if ( tabledef ) {
 453                 source_def = strdup( tabledef );   // Return the name of a table or view
 454         } else {
 455                 tabledef = osrfHashGet( class, "source_definition" );
 456                 if( tabledef ) {
 457                         // Return a subquery, enclosed in parentheses
 458                         source_def = safe_malloc( strlen( tabledef ) + 3 );
 459                         source_def[ 0 ] = '(';
 460                         strcpy( source_def + 1, tabledef );
 461                         strcat( source_def, ")" );
 462                 } else {
 463                         // Not found: return an error
 464                         const char* classname = osrfHashGet( class, "classname" );
 465                         if( !classname )
 466                                 classname = "???";
 467                         osrfLogError(
 468                                 OSRF_LOG_MARK,
 469                                 "%s ERROR No tablename or source_definition for class \"%s\"",
 470                                 modulename,
 471                                 classname
 472                         );
 473                 }
 474         }
 475
 476         return source_def;
 477 }
 478
 479 /**
 480         @brief Initialize a server drone.
 481         @return Zero if successful, -1 if not.
 482
 483         Connect to the database.  For each non-virtual class in the IDL, execute a dummy "SELECT * "
 484         query to get the datatype of each column.  Record the datatypes in the loaded IDL.
 485
 486         This function is called by a server drone shortly after it is spawned by the listener.
 487 */
 488 int osrfAppChildInit() {
 489
 490         osrfLogDebug(OSRF_LOG_MARK, "Attempting to initialize libdbi...");
 491         dbi_initialize(NULL);
 492         osrfLogDebug(OSRF_LOG_MARK, "... libdbi initialized.");
 493
 494         char* driver = osrf_settings_host_value("/apps/%s/app_settings/driver", modulename );
 495         char* user   = osrf_settings_host_value("/apps/%s/app_settings/database/user", modulename );
 496         char* host   = osrf_settings_host_value("/apps/%s/app_settings/database/host", modulename );
 497         char* port   = osrf_settings_host_value("/apps/%s/app_settings/database/port", modulename );
 498         char* db     = osrf_settings_host_value("/apps/%s/app_settings/database/db", modulename );
 499         char* pw     = osrf_settings_host_value("/apps/%s/app_settings/database/pw", modulename );
 500         char* md     = osrf_settings_host_value("/apps/%s/app_settings/max_query_recursion",
 501                         modulename );
 502
 503         osrfLogDebug(OSRF_LOG_MARK, "Attempting to load the database driver [%s]...", driver);
 504         writehandle = dbi_conn_new(driver);
 505
 506         if(!writehandle) {
 507                 osrfLogError(OSRF_LOG_MARK, "Error loading database driver [%s]", driver);
 508                 return -1;
 509         }
 510         osrfLogDebug(OSRF_LOG_MARK, "Database driver [%s] seems OK", driver);
 511
 512         osrfLogInfo(OSRF_LOG_MARK, "%s connecting to database.  host=%s, "
 513                         "port=%s, user=%s, db=%s", modulename, host, port, user, db );
 514
 515         if(host) dbi_conn_set_option(writehandle, "host", host );
 516         if(port) dbi_conn_set_option_numeric( writehandle, "port", atoi(port) );
 517         if(user) dbi_conn_set_option(writehandle, "username", user);
 518         if(pw)   dbi_conn_set_option(writehandle, "password", pw );
 519         if(db)   dbi_conn_set_option(writehandle, "dbname", db );
 520
 521         if(md)                     max_flesh_depth = atoi(md);
 522         if(max_flesh_depth < 0)    max_flesh_depth = 1;
 523         if(max_flesh_depth > 1000) max_flesh_depth = 1000;
 524
 525         free(user);
 526         free(host);
 527         free(port);
 528         free(db);
 529         free(pw);
 530
 531         const char* err;
 532         if (dbi_conn_connect(writehandle) < 0) {
 533                 sleep(1);
 534                 if (dbi_conn_connect(writehandle) < 0) {
 535                         dbi_conn_error(writehandle, &err);
 536                         osrfLogError( OSRF_LOG_MARK, "Error connecting to database: %s", err);
 537                         return -1;
 538                 }
 539         }
 540
 541         osrfLogInfo(OSRF_LOG_MARK, "%s successfully connected to the database", modulename );
 542
 543         osrfHashIterator* class_itr = osrfNewHashIterator( oilsIDL() );
 544         osrfHash* class = NULL;
 545         growing_buffer* query_buf = buffer_init( 64 );
 546
 547         // For each class in the IDL...
 548         while( (class = osrfHashIteratorNext( class_itr ) ) ) {
 549                 const char* classname = osrfHashIteratorKey( class_itr );
 550                 osrfHash* fields = osrfHashGet( class, "fields" );
 551
 552                 // If the class is virtual, ignore it
 553                 if( str_is_true( osrfHashGet(class, "virtual") ) ) {
 554                         osrfLogDebug(OSRF_LOG_MARK, "Class %s is virtual, skipping", classname );
 555                         continue;
 556                 }
 557
 558                 char* tabledef = getRelation(class);
 559                 if( !tabledef )
 560                         continue;   // No such relation -- a query of it would be doomed to failure
 561
 562                 buffer_reset( query_buf );
 563                 buffer_fadd( query_buf, "SELECT * FROM %s AS x WHERE 1=0;", tabledef );
 564
 565                 free(tabledef);
 566
 567                 osrfLogDebug( OSRF_LOG_MARK, "%s Investigatory SQL = %s",
 568                                 modulename, OSRF_BUFFER_C_STR( query_buf ) );
 569
 570                 dbi_result result = dbi_conn_query( writehandle, OSRF_BUFFER_C_STR( query_buf ) );
 571                 if (result) {
 572
 573                         int columnIndex = 1;
 574                         const char* columnName;
 575                         osrfHash* _f;
 576                         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
 577
 578                                 osrfLogInternal( OSRF_LOG_MARK, "Looking for column named [%s]...",
 579                                                 columnName );
 580
 581                                 /* fetch the fieldmapper index */
 582                                 if( (_f = osrfHashGet(fields, columnName)) ) {
 583
 584                                         osrfLogDebug(OSRF_LOG_MARK, "Found [%s] in IDL hash...", columnName);
 585
 586                                         /* determine the field type and storage attributes */
 587
 588                                         switch( dbi_result_get_field_type_idx(result, columnIndex) ) {
 589
 590                                                 case DBI_TYPE_INTEGER : {
 591
 592                                                         if ( !osrfHashGet(_f, "primitive") )
 593                                                                 osrfHashSet(_f, "number", "primitive");
 594
 595                                                         int attr = dbi_result_get_field_attribs_idx(result, columnIndex);
 596                                                         if( attr & DBI_INTEGER_SIZE8 )
 597                                                                 osrfHashSet(_f, "INT8", "datatype");
 598                                                         else
 599                                                                 osrfHashSet(_f, "INT", "datatype");
 600                                                         break;
 601                                                 }
 602                                                 case DBI_TYPE_DECIMAL :
 603                                                         if ( !osrfHashGet(_f, "primitive") )
 604                                                                 osrfHashSet(_f, "number", "primitive");
 605
 606                                                         osrfHashSet(_f,"NUMERIC", "datatype");
 607                                                         break;
 608
 609                                                 case DBI_TYPE_STRING :
 610                                                         if ( !osrfHashGet(_f, "primitive") )
 611                                                                 osrfHashSet(_f,"string", "primitive");
 612
 613                                                         osrfHashSet(_f,"TEXT", "datatype");
 614                                                         break;
 615
 616                                                 case DBI_TYPE_DATETIME :
 617                                                         if ( !osrfHashGet(_f, "primitive") )
 618                                                                 osrfHashSet(_f,"string", "primitive");
 619
 620                                                         osrfHashSet(_f,"TIMESTAMP", "datatype");
 621                                                         break;
 622
 623                                                 case DBI_TYPE_BINARY :
 624                                                         if ( !osrfHashGet(_f, "primitive") )
 625                                                                 osrfHashSet(_f,"string", "primitive");
 626
 627                                                         osrfHashSet(_f,"BYTEA", "datatype");
 628                                         }
 629
 630                                         osrfLogDebug(
 631                                                 OSRF_LOG_MARK,
 632                                                 "Setting [%s] to primitive [%s] and datatype [%s]...",
 633                                                 columnName,
 634                                                 osrfHashGet(_f, "primitive"),
 635                                                 osrfHashGet(_f, "datatype")
 636                                         );
 637                                 }
 638                                 ++columnIndex;
 639                         } // end while loop for traversing columns of result
 640                         dbi_result_free(result);
 641                 } else {
 642                         osrfLogDebug(OSRF_LOG_MARK, "No data found for class [%s]...", classname);
 643                 }
 644         } // end for each class in IDL
 645
 646         buffer_free( query_buf );
 647         osrfHashIteratorFree( class_itr );
 648         child_initialized = 1;
 649         return 0;
 650 }
 651
 652 /**
 653         @brief Install a database driver.
 654         @param conn Pointer to a database driver.
 655
 656         The driver is used to process quoted strings correctly.
 657
 658         This function is a sleazy hack, intended @em only for testing and debugging without
 659         actually connecting to a database. Any real server process should initialize the
 660         database connection by calling osrfAppChildInit().
 661 */
 662 void set_cstore_dbi_conn( dbi_conn conn ) {
 663         dbhandle = writehandle = conn;
 664 }
 665
 666 /**
 667         @brief Free an osrfHash that stores a transaction ID.
 668         @param blob A pointer to the osrfHash to be freed, cast to a void pointer.
 669
 670         This function is a callback, to be called by the application session when it ends.
 671         The application session stores the osrfHash via an opaque pointer.
 672
 673         If the osrfHash contains an entry for the key "xact_id", it means that an
 674         uncommitted transaction is pending.  Roll it back.
 675 */
 676 void userDataFree( void* blob ) {
 677         osrfHash* hash = (osrfHash*) blob;
 678         if( osrfHashGet( hash, "xact_id" ) && writehandle )
 679                 dbi_conn_query( writehandle, "ROLLBACK;" );
 680
 681         osrfHashFree( hash );
 682 }
 683
 684 /**
 685         @name Managing session data
 686         @brief Maintain data stored via the userData pointer of the application session.
 687
 688         Currently, session-level data is stored in an osrfHash.  Other arrangements are
 689         possible, and some would be more efficient.  The application session calls a
 690         callback function to free userData before terminating.
 691
 692         Currently, the only data we store at the session level is the transaction id.  By this
 693         means we can ensure that any pending transactions are rolled back before the application
 694         session terminates.
 695 */
 696 /*@{*/
 697
 698 /**
 699         @brief Free an item in the application session's userData.
 700         @param key The name of a key for an osrfHash.
 701         @param item An opaque pointer to the item associated with the key.
 702
 703         We store an osrfHash as userData with the application session, and arrange (by
 704         installing userDataFree() as a different callback) for the session to free that
 705         osrfHash before terminating.
 706
 707         This function is a callback for freeing items in the osrfHash.  Currently we store
 708         two things:
 709         - Transaction id of a pending transaction; a character string.  Key: "xact_id".
 710         - Authkey; a character string.  Key: "authkey".
 711         - User object from the authentication server; a jsonObject.  Key: "user_login".
 712
 713         If we ever store anything else in userData, we will need to revisit this function so
 714         that it will free whatever else needs freeing.
 715 */
 716 static void sessionDataFree( char* key, void* item ) {
 717         if ( !strcmp( key, "xact_id" )
 718              || !strcmp( key, "authkey" ) ) {
 719                 free( item );
 720         } else if( !strcmp( key, "user_login" ) )
 721                 jsonObjectFree( (jsonObject*) item );
 722 }
 723
 724 /**
 725         @brief Save a transaction id.
 726         @param ctx Pointer to the method context.
 727
 728         Save the session_id of the current application session as a transaction id.
 729 */
 730 static void setXactId( osrfMethodContext* ctx ) {
 731         if( ctx && ctx->session ) {
 732                 osrfAppSession* session = ctx->session;
 733
 734                 osrfHash* cache = session->userData;
 735
 736                 // If the session doesn't already have a hash, create one.  Make sure
 737                 // that the application session frees the hash when it terminates.
 738                 if( NULL == cache ) {
 739                         session->userData = cache = osrfNewHash();
 740                         osrfHashSetCallback( cache, &sessionDataFree );
 741                         ctx->session->userDataFree = &userDataFree;
 742                 }
 743
 744                 // Save the transaction id in the hash, with the key "xact_id"
 745                 osrfHashSet( cache, strdup( session->session_id ), "xact_id" );
 746         }
 747 }
 748
 749 /**
 750         @brief Get the transaction ID for the current transaction, if any.
 751         @param ctx Pointer to the method context.
 752         @return Pointer to the transaction ID.
 753
 754         The return value points to an internal buffer, and will become invalid upon issuing
 755         a commit or rollback.
 756 */
 757 static inline const char* getXactId( osrfMethodContext* ctx ) {
 758         if( ctx && ctx->session && ctx->session->userData )
 759                 return osrfHashGet( (osrfHash*) ctx->session->userData, "xact_id" );
 760         else
 761                 return NULL;
 762 }
 763
 764 /**
 765         @brief Clear the current transaction id.
 766         @param ctx Pointer to the method context.
 767 */
 768 static inline void clearXactId( osrfMethodContext* ctx ) {
 769         if( ctx && ctx->session && ctx->session->userData )
 770                 osrfHashRemove( ctx->session->userData, "xact_id" );
 771 }
 772 /*@}*/
 773
 774 /**
 775         @brief Save the user's login in the userData for the current application session.
 776         @param ctx Pointer to the method context.
 777         @param user_login Pointer to the user login object to be cached (we cache the original,
 778         not a copy of it).
 779
 780         If @a user_login is NULL, remove the user login if one is already cached.
 781 */
 782 static void setUserLogin( osrfMethodContext* ctx, jsonObject* user_login ) {
 783         if( ctx && ctx->session ) {
 784                 osrfAppSession* session = ctx->session;
 785
 786                 osrfHash* cache = session->userData;
 787
 788                 // If the session doesn't already have a hash, create one.  Make sure
 789                 // that the application session frees the hash when it terminates.
 790                 if( NULL == cache ) {
 791                         session->userData = cache = osrfNewHash();
 792                         osrfHashSetCallback( cache, &sessionDataFree );
 793                         ctx->session->userDataFree = &userDataFree;
 794                 }
 795
 796                 if( user_login )
 797                         osrfHashSet( cache, user_login, "user_login" );
 798                 else
 799                         osrfHashRemove( cache, "user_login" );
 800         }
 801 }
 802
 803 /**
 804         @brief Get the user login object for the current application session, if any.
 805         @param ctx Pointer to the method context.
 806         @return Pointer to the user login object if found; otherwise NULL.
 807
 808         The user login object was returned from the authentication server, and then cached so
 809         we don't have to call the authentication server again for the same user.
 810 */
 811 static const jsonObject* getUserLogin( osrfMethodContext* ctx ) {
 812         if( ctx && ctx->session && ctx->session->userData )
 813                 return osrfHashGet( (osrfHash*) ctx->session->userData, "user_login" );
 814         else
 815                 return NULL;
 816 }
 817
 818 /**
 819         @brief Save a copy of an authkey in the userData of the current application session.
 820         @param ctx Pointer to the method context.
 821         @param authkey The authkey to be saved.
 822
 823         If @a authkey is NULL, remove the authkey if one is already cached.
 824 */
 825 static void setAuthkey( osrfMethodContext* ctx, const char* authkey ) {
 826         if( ctx && ctx->session && authkey ) {
 827                 osrfAppSession* session = ctx->session;
 828                 osrfHash* cache = session->userData;
 829
 830                 // If the session doesn't already have a hash, create one.  Make sure
 831                 // that the application session frees the hash when it terminates.
 832                 if( NULL == cache ) {
 833                         session->userData = cache = osrfNewHash();
 834                         osrfHashSetCallback( cache, &sessionDataFree );
 835                         ctx->session->userDataFree = &userDataFree;
 836                 }
 837
 838                 // Save the transaction id in the hash, with the key "xact_id"
 839                 if( authkey && *authkey )
 840                         osrfHashSet( cache, strdup( authkey ), "authkey" );
 841                 else
 842                         osrfHashRemove( cache, "authkey" );
 843         }
 844 }
 845
 846 /**
 847         @brief Reset the login timeout.
 848         @param authkey The authentication key for the current login session.
 849         @param now The current time.
 850         @return Zero if successful, or 1 if not.
 851
 852         Tell the authentication server to reset the timeout so that the login session won't
 853         expire for a while longer.
 854
 855         We could dispense with the @a now parameter by calling time().  But we just called
 856         time() in order to decide whether to reset the timeout, so we might as well reuse
 857         the result instead of calling time() again.
 858 */
 859 static int reset_timeout( const char* authkey, time_t now ) {
 860         jsonObject* auth_object = jsonNewObject( authkey );
 861
 862         // Ask the authentication server to reset the timeout.  It returns an event
 863         // indicating success or failure.
 864         jsonObject* result = oilsUtilsQuickReq( "open-ils.auth",
 865                 "open-ils.auth.session.reset_timeout", auth_object );
 866         jsonObjectFree(auth_object);
 867
 868         if( !result || result->type != JSON_HASH ) {
 869                 osrfLogError( OSRF_LOG_MARK,
 870                          "Unexpected object type receieved from open-ils.auth.session.reset_timeout" );
 871                 jsonObjectFree( result );
 872                 return 1;       // Not the right sort of object returned
 873         }
 874
 875         const jsonObject* ilsevent = jsonObjectGetKey( result, "ilsevent" );
 876         if( !ilsevent || ilsevent->type != JSON_NUMBER ) {
 877                 osrfLogError( OSRF_LOG_MARK, "ilsevent is absent or malformed" );
 878                 jsonObjectFree( result );
 879                 return 1;    // Return code from method not available
 880         }
 881
 882         if( jsonObjectGetNumber( ilsevent ) != 0.0 ){
 883                 const char* desc = jsonObjectGetString( jsonObjectGetKey( result, "desc" ) );
 884                 if( !desc )
 885                         desc = "(No reason available)";    // failsafe; shouldn't happen
 886                 osrfLogInfo( OSRF_LOG_MARK, "Failure to reset timeout: %s", desc );
 887                 jsonObjectFree( result );
 888                 return 1;
 889         }
 890
 891         // Revise our local proxy for the timeout deadline
 892         // by a smallish fraction of the timeout interval
 893         const char* timeout = jsonObjectGetString( jsonObjectGetKey( result, "payload" ) );
 894         if( !timeout )
 895                 timeout = "1";   // failsafe; shouldn't happen
 896         time_next_reset = now + atoi( timeout ) / 15;
 897
 898         jsonObjectFree( result );
 899         return 0;     // Successfully reset timeout
 900 }
 901
 902 /**
 903         @brief Get the authkey string for the current application session, if any.
 904         @param ctx Pointer to the method context.
 905         @return Pointer to the cached authkey if found; otherwise NULL.
 906
 907         If present, the authkey string was cached from a previous method call.
 908 */
 909 static const char* getAuthkey( osrfMethodContext* ctx ) {
 910         if( ctx && ctx->session && ctx->session->userData ) {
 911                 const char* authkey = osrfHashGet( (osrfHash*) ctx->session->userData, "authkey" );
 912
 913                 // Possibly reset the authentication timeout to keep the login alive.  We do so
 914                 // no more than once per method call, and not at all if it has been only a short
 915                 // time since the last reset.
 916
 917                 // Here we reset explicitly, if at all.  We also implicitly reset the timeout
 918                 // whenever we call the "open-ils.auth.session.retrieve" method.
 919                 if( timeout_needs_resetting ) {
 920                         time_t now = time( NULL );
 921                         if( now >= time_next_reset && reset_timeout( authkey, now ) )
 922                                 authkey = NULL;    // timeout has apparently expired already
 923                 }
 924
 925                 timeout_needs_resetting = 0;
 926                 return authkey;
 927         }
 928         else
 929                 return NULL;
 930 }
 931
 932 /**
 933         @brief Implement the transaction.begin method.
 934         @param ctx Pointer to the method context.
 935         @return Zero if successful, or -1 upon error.
 936
 937         Start a transaction.  Save a transaction ID for future reference.
 938
 939         Method parameters:
 940         - authkey (PCRUD only)
 941
 942         Return to client: Transaction ID
 943 */
 944 int beginTransaction ( osrfMethodContext* ctx ) {
 945         if(osrfMethodVerifyContext( ctx )) {
 946                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
 947                 return -1;
 948         }
 949
 950         if( enforce_pcrud ) {
 951                 timeout_needs_resetting = 1;
 952                 const jsonObject* user = verifyUserPCRUD( ctx );
 953                 if (!user)
 954                         return -1;
 955         }
 956
 957         dbi_result result = dbi_conn_query(writehandle, "START TRANSACTION;");
 958         if (!result) {
 959                 osrfLogError(OSRF_LOG_MARK, "%s: Error starting transaction", modulename );
 960                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
 961                                 "osrfMethodException", ctx->request, "Error starting transaction" );
 962                 return -1;
 963         } else {
 964                 setXactId( ctx );
 965                 jsonObject* ret = jsonNewObject( getXactId( ctx ) );
 966                 osrfAppRespondComplete( ctx, ret );
 967                 jsonObjectFree(ret);
 968         }
 969         return 0;
 970 }
 971
 972 /**
 973         @brief Implement the savepoint.set method.
 974         @param ctx Pointer to the method context.
 975         @return Zero if successful, or -1 if not.
 976
 977         Issue a SAVEPOINT to the database server.
 978
 979         Method parameters:
 980         - authkey (PCRUD only)
 981         - savepoint name
 982
 983         Return to client: Savepoint name
 984 */
 985 int setSavepoint ( osrfMethodContext* ctx ) {
 986         if(osrfMethodVerifyContext( ctx )) {
 987                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
 988                 return -1;
 989         }
 990
 991         int spNamePos = 0;
 992         if( enforce_pcrud ) {
 993                 spNamePos = 1;
 994                 timeout_needs_resetting = 1;
 995                 const jsonObject* user = verifyUserPCRUD( ctx );
 996                 if (!user)
 997                         return -1;
 998         }
 999
1000         // Verify that a transaction is pending
1001         const char* trans_id = getXactId( ctx );
1002         if( NULL == trans_id ) {
1003                 osrfAppSessionStatus(
1004                         ctx->session,
1005                         OSRF_STATUS_INTERNALSERVERERROR,
1006                         "osrfMethodException",
1007                         ctx->request,
1008                         "No active transaction -- required for savepoints"
1009                 );
1010                 return -1;
1011         }
1012
1013         // Get the savepoint name from the method params
1014         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1015
1016         dbi_result result = dbi_conn_queryf(writehandle, "SAVEPOINT \"%s\";", spName);
1017         if (!result) {
1018                 osrfLogError(
1019                         OSRF_LOG_MARK,
1020                         "%s: Error creating savepoint %s in transaction %s",
1021                         modulename,
1022                         spName,
1023                         trans_id
1024                 );
1025                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1026                                 "osrfMethodException", ctx->request, "Error creating savepoint" );
1027                 return -1;
1028         } else {
1029                 jsonObject* ret = jsonNewObject(spName);
1030                 osrfAppRespondComplete( ctx, ret );
1031                 jsonObjectFree(ret);
1032         }
1033         return 0;
1034 }
1035
1036 /**
1037         @brief Implement the savepoint.release method.
1038         @param ctx Pointer to the method context.
1039         @return Zero if successful, or -1 if not.
1040
1041         Issue a RELEASE SAVEPOINT to the database server.
1042
1043         Method parameters:
1044         - authkey (PCRUD only)
1045         - savepoint name
1046
1047         Return to client: Savepoint name
1048 */
1049 int releaseSavepoint ( osrfMethodContext* ctx ) {
1050         if(osrfMethodVerifyContext( ctx )) {
1051                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1052                 return -1;
1053         }
1054
1055         int spNamePos = 0;
1056         if( enforce_pcrud ) {
1057                 spNamePos = 1;
1058                 timeout_needs_resetting = 1;
1059                 const jsonObject* user = verifyUserPCRUD( ctx );
1060                 if (!user)
1061                         return -1;
1062         }
1063
1064         // Verify that a transaction is pending
1065         const char* trans_id = getXactId( ctx );
1066         if( NULL == trans_id ) {
1067                 osrfAppSessionStatus(
1068                         ctx->session,
1069                         OSRF_STATUS_INTERNALSERVERERROR,
1070                         "osrfMethodException",
1071                         ctx->request,
1072                         "No active transaction -- required for savepoints"
1073                 );
1074                 return -1;
1075         }
1076
1077         // Get the savepoint name from the method params
1078         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1079
1080         dbi_result result = dbi_conn_queryf(writehandle, "RELEASE SAVEPOINT \"%s\";", spName);
1081         if (!result) {
1082                 osrfLogError(
1083                         OSRF_LOG_MARK,
1084                         "%s: Error releasing savepoint %s in transaction %s",
1085                         modulename,
1086                         spName,
1087                         trans_id
1088                 );
1089                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1090                                 "osrfMethodException", ctx->request, "Error releasing savepoint" );
1091                 return -1;
1092         } else {
1093                 jsonObject* ret = jsonNewObject(spName);
1094                 osrfAppRespondComplete( ctx, ret );
1095                 jsonObjectFree(ret);
1096         }
1097         return 0;
1098 }
1099
1100 /**
1101         @brief Implement the savepoint.rollback method.
1102         @param ctx Pointer to the method context.
1103         @return Zero if successful, or -1 if not.
1104
1105         Issue a ROLLBACK TO SAVEPOINT to the database server.
1106
1107         Method parameters:
1108         - authkey (PCRUD only)
1109         - savepoint name
1110
1111         Return to client: Savepoint name
1112 */
1113 int rollbackSavepoint ( osrfMethodContext* ctx ) {
1114         if(osrfMethodVerifyContext( ctx )) {
1115                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1116                 return -1;
1117         }
1118
1119         int spNamePos = 0;
1120         if( enforce_pcrud ) {
1121                 spNamePos = 1;
1122                 timeout_needs_resetting = 1;
1123                 const jsonObject* user = verifyUserPCRUD( ctx );
1124                 if (!user)
1125                         return -1;
1126         }
1127
1128         // Verify that a transaction is pending
1129         const char* trans_id = getXactId( ctx );
1130         if( NULL == trans_id ) {
1131                 osrfAppSessionStatus(
1132                         ctx->session,
1133                         OSRF_STATUS_INTERNALSERVERERROR,
1134                         "osrfMethodException",
1135                         ctx->request,
1136                         "No active transaction -- required for savepoints"
1137                 );
1138                 return -1;
1139         }
1140
1141         // Get the savepoint name from the method params
1142         const char* spName = jsonObjectGetString( jsonObjectGetIndex(ctx->params, spNamePos) );
1143
1144         dbi_result result = dbi_conn_queryf(writehandle, "ROLLBACK TO SAVEPOINT \"%s\";", spName);
1145         if (!result) {
1146                 osrfLogError(
1147                         OSRF_LOG_MARK,
1148                         "%s: Error rolling back savepoint %s in transaction %s",
1149                         modulename,
1150                         spName,
1151                         trans_id
1152                 );
1153                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1154                                 "osrfMethodException", ctx->request, "Error rolling back savepoint" );
1155                 return -1;
1156         } else {
1157                 jsonObject* ret = jsonNewObject(spName);
1158                 osrfAppRespondComplete( ctx, ret );
1159                 jsonObjectFree(ret);
1160         }
1161         return 0;
1162 }
1163
1164 /**
1165         @brief Implement the transaction.commit method.
1166         @param ctx Pointer to the method context.
1167         @return Zero if successful, or -1 if not.
1168
1169         Issue a COMMIT to the database server.
1170
1171         Method parameters:
1172         - authkey (PCRUD only)
1173
1174         Return to client: Transaction ID.
1175 */
1176 int commitTransaction ( osrfMethodContext* ctx ) {
1177         if(osrfMethodVerifyContext( ctx )) {
1178                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1179                 return -1;
1180         }
1181
1182         if( enforce_pcrud ) {
1183                 timeout_needs_resetting = 1;
1184                 const jsonObject* user = verifyUserPCRUD( ctx );
1185                 if (!user)
1186                         return -1;
1187         }
1188
1189         // Verify that a transaction is pending
1190         const char* trans_id = getXactId( ctx );
1191         if( NULL == trans_id ) {
1192                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1193                                 "osrfMethodException", ctx->request, "No active transaction to commit" );
1194                 return -1;
1195         }
1196
1197         dbi_result result = dbi_conn_query(writehandle, "COMMIT;");
1198         if (!result) {
1199                 osrfLogError(OSRF_LOG_MARK, "%s: Error committing transaction", modulename );
1200                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1201                                 "osrfMethodException", ctx->request, "Error committing transaction" );
1202                 return -1;
1203         } else {
1204                 jsonObject* ret = jsonNewObject( trans_id );
1205                 osrfAppRespondComplete( ctx, ret );
1206                 jsonObjectFree(ret);
1207                 clearXactId( ctx );
1208         }
1209         return 0;
1210 }
1211
1212 /**
1213         @brief Implement the transaction.rollback method.
1214         @param ctx Pointer to the method context.
1215         @return Zero if successful, or -1 if not.
1216
1217         Issue a ROLLBACK to the database server.
1218
1219         Method parameters:
1220         - authkey (PCRUD only)
1221
1222         Return to client: Transaction ID
1223 */
1224 int rollbackTransaction ( osrfMethodContext* ctx ) {
1225         if(osrfMethodVerifyContext( ctx )) {
1226                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
1227                 return -1;
1228         }
1229
1230         if( enforce_pcrud ) {
1231                 timeout_needs_resetting = 1;
1232                 const jsonObject* user = verifyUserPCRUD( ctx );
1233                 if (!user)
1234                         return -1;
1235         }
1236
1237         // Verify that a transaction is pending
1238         const char* trans_id = getXactId( ctx );
1239         if( NULL == trans_id ) {
1240                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1241                                 "osrfMethodException", ctx->request, "No active transaction to roll back" );
1242                 return -1;
1243         }
1244
1245         dbi_result result = dbi_conn_query(writehandle, "ROLLBACK;");
1246         if (!result) {
1247                 osrfLogError(OSRF_LOG_MARK, "%s: Error rolling back transaction", modulename );
1248                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_INTERNALSERVERERROR,
1249                                 "osrfMethodException", ctx->request, "Error rolling back transaction" );
1250                 return -1;
1251         } else {
1252                 jsonObject* ret = jsonNewObject( trans_id );
1253                 osrfAppRespondComplete( ctx, ret );
1254                 jsonObjectFree(ret);
1255                 clearXactId( ctx );
1256         }
1257         return 0;
1258 }
1259
1260 /**
1261         @brief Implement the class-specific methods.
1262         @param ctx Pointer to the method context.
1263         @return Zero if successful, or -1 if not.
1264
1265         Branch on the method type: create, retrieve, update, delete, search, or id_list.
1266
1267         The method parameters and the type of value returned to the client depend on the method
1268         type.  However, for PCRUD methods, the first method parameter should always be an
1269         authkey.
1270 */
1271 int dispatchCRUDMethod ( osrfMethodContext* ctx ) {
1272
1273         // Get the method type, then can branch on it
1274         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1275         const char* methodtype = osrfHashGet( method_meta, "methodtype" );
1276
1277         if( !strcmp( methodtype, "create" ))
1278                 return doCreate( ctx );
1279         else if( !strcmp(methodtype, "retrieve" ))
1280                 return doRetrieve( ctx );
1281         else if( !strcmp(methodtype, "update" ))
1282                 return doUpdate( ctx );
1283         else if( !strcmp(methodtype, "delete" ))
1284                 return doDelete( ctx );
1285         else if( !strcmp(methodtype, "search" ))
1286                 return doSearch( ctx );
1287         else if( !strcmp(methodtype, "id_list" ))
1288                 return doIdList( ctx );
1289         else {
1290                 osrfAppRespondComplete( ctx, NULL );      // should be unreachable...
1291                 return 0;
1292         }
1293 }
1294
1295 /**
1296         @brief Implement the "search" method.
1297         @param ctx Pointer to the method context.
1298         @return Zero if successful, or -1 if not.
1299
1300         Method parameters:
1301         - authkey (PCRUD only)
1302         - WHERE clause, as jsonObject
1303         - Other SQL clause(s), as a JSON_HASH: joins, SELECT list, LIMIT, etc.
1304
1305         Return to client: rows of the specified class that satisfy a specified WHERE clause.
1306         Optionally flesh linked fields.
1307 */
1308 static int doSearch( osrfMethodContext* ctx ) {
1309         if(osrfMethodVerifyContext( ctx )) {
1310                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1311                 return -1;
1312         }
1313
1314         if( enforce_pcrud )
1315                 timeout_needs_resetting = 1;
1316
1317         jsonObject* where_clause;
1318         jsonObject* rest_of_query;
1319
1320         if( enforce_pcrud ) {
1321                 where_clause  = jsonObjectGetIndex( ctx->params, 1 );
1322                 rest_of_query = jsonObjectGetIndex( ctx->params, 2 );
1323         } else {
1324                 where_clause  = jsonObjectGetIndex( ctx->params, 0 );
1325                 rest_of_query = jsonObjectGetIndex( ctx->params, 1 );
1326         }
1327
1328         // Get the class metadata
1329         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1330         osrfHash* class_meta = osrfHashGet( method_meta, "class" );
1331
1332         // Do the query
1333         int err = 0;
1334         jsonObject* obj = doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );
1335         if( err ) {
1336                 osrfAppRespondComplete( ctx, NULL );
1337                 return -1;
1338         }
1339
1340         // Return each row to the client (except that some may be suppressed by PCRUD)
1341         jsonObject* cur = 0;
1342         unsigned long res_idx = 0;
1343         while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
1344                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
1345                         continue;
1346                 osrfAppRespond( ctx, cur );
1347         }
1348         jsonObjectFree( obj );
1349
1350         osrfAppRespondComplete( ctx, NULL );
1351         return 0;
1352 }
1353
1354 /**
1355         @brief Implement the "id_list" method.
1356         @param ctx Pointer to the method context.
1357         @param err Pointer through which to return an error code.
1358         @return Zero if successful, or -1 if not.
1359
1360         Method parameters:
1361         - authkey (PCRUD only)
1362         - WHERE clause, as jsonObject
1363         - Other SQL clause(s), as a JSON_HASH: joins, LIMIT, etc.
1364
1365         Return to client: The primary key values for all rows of the relevant class that
1366         satisfy a specified WHERE clause.
1367
1368         This method relies on the assumption that every class has a primary key consisting of
1369         a single column.
1370 */
1371 static int doIdList( osrfMethodContext* ctx ) {
1372         if(osrfMethodVerifyContext( ctx )) {
1373                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
1374                 return -1;
1375         }
1376
1377         if( enforce_pcrud )
1378                 timeout_needs_resetting = 1;
1379
1380         jsonObject* where_clause;
1381         jsonObject* rest_of_query;
1382
1383         // We use the where clause without change.  But we need to massage the rest of the
1384         // query, so we work with a copy of it instead of modifying the original.
1385
1386         if( enforce_pcrud ) {
1387                 where_clause  = jsonObjectGetIndex( ctx->params, 1 );
1388                 rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 2 ) );
1389         } else {
1390                 where_clause  = jsonObjectGetIndex( ctx->params, 0 );
1391                 rest_of_query = jsonObjectClone( jsonObjectGetIndex( ctx->params, 1 ) );
1392         }
1393
1394         // Eliminate certain SQL clauses, if present.
1395         if ( rest_of_query ) {
1396                 jsonObjectRemoveKey( rest_of_query, "select" );
1397                 jsonObjectRemoveKey( rest_of_query, "no_i18n" );
1398                 jsonObjectRemoveKey( rest_of_query, "flesh" );
1399                 jsonObjectRemoveKey( rest_of_query, "flesh_columns" );
1400         } else {
1401                 rest_of_query = jsonNewObjectType( JSON_HASH );
1402         }
1403
1404         jsonObjectSetKey( rest_of_query, "no_i18n", jsonNewBoolObject( 1 ) );
1405
1406         // Get the class metadata
1407         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1408         osrfHash* class_meta = osrfHashGet( method_meta, "class" );
1409
1410         // Build a SELECT list containing just the primary key,
1411         // i.e. like { "classname":["keyname"] }
1412         jsonObject* col_list_obj = jsonNewObjectType( JSON_ARRAY );
1413
1414         // Load array with name of primary key
1415         jsonObjectPush( col_list_obj, jsonNewObject( osrfHashGet( class_meta, "primarykey" ) ) );
1416         jsonObject* select_clause = jsonNewObjectType( JSON_HASH );
1417         jsonObjectSetKey( select_clause, osrfHashGet( class_meta, "classname" ), col_list_obj );
1418
1419         jsonObjectSetKey( rest_of_query, "select", select_clause );
1420
1421         // Do the query
1422         int err = 0;
1423         jsonObject* obj =
1424                 doFieldmapperSearch( ctx, class_meta, where_clause, rest_of_query, &err );
1425
1426         jsonObjectFree( rest_of_query );
1427         if( err ) {
1428                 osrfAppRespondComplete( ctx, NULL );
1429                 return -1;
1430         }
1431
1432         // Return each primary key value to the client
1433         jsonObject* cur;
1434         unsigned long res_idx = 0;
1435         while((cur = jsonObjectGetIndex( obj, res_idx++ ) )) {
1436                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, cur ))
1437                         continue;        // Suppress due to lack of permission
1438                 else
1439                         osrfAppRespond( ctx,
1440                                 oilsFMGetObject( cur, osrfHashGet( class_meta, "primarykey" ) ) );
1441         }
1442
1443         jsonObjectFree( obj );
1444         osrfAppRespondComplete( ctx, NULL );
1445         return 0;
1446 }
1447
1448 /**
1449         @brief Verify that we have a valid class reference.
1450         @param ctx Pointer to the method context.
1451         @param param Pointer to the method parameters.
1452         @return 1 if the class reference is valid, or zero if it isn't.
1453
1454         The class of the method params must match the class to which the method id devoted.
1455         For PCRUD there are additional restrictions.
1456 */
1457 static int verifyObjectClass ( osrfMethodContext* ctx, const jsonObject* param ) {
1458
1459         osrfHash* method_meta = (osrfHash*) ctx->method->userData;
1460         osrfHash* class = osrfHashGet( method_meta, "class" );
1461
1462         // Compare the method's class to the parameters' class
1463         if (!param->classname || (strcmp( osrfHashGet(class, "classname"), param->classname ))) {
1464
1465                 // Oops -- they don't match.  Complain.
1466                 growing_buffer* msg = buffer_init(128);
1467                 buffer_fadd(
1468                         msg,
1469                         "%s: %s method for type %s was passed a %s",
1470                         modulename,
1471                         osrfHashGet(method_meta, "methodtype"),
1472                         osrfHashGet(class, "classname"),
1473                         param->classname ? param->classname : "(null)"
1474                 );
1475
1476                 char* m = buffer_release(msg);
1477                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_BADREQUEST, "osrfMethodException",
1478                                 ctx->request, m );
1479                 free(m);
1480
1481                 return 0;
1482         }
1483
1484         if( enforce_pcrud )
1485                 return verifyObjectPCRUD( ctx, param );
1486         else
1487                 return 1;
1488 }
1489
1490 /**
1491         @brief (PCRUD only) Verify that the user is properly logged in.
1492         @param ctx Pointer to the method context.
1493         @return If the user is logged in, a pointer to the user object from the authentication
1494         server; otherwise NULL.
1495 */
1496 static const jsonObject* verifyUserPCRUD( osrfMethodContext* ctx ) {
1497
1498         // Get the authkey (the first method parameter)
1499         const char* auth = jsonObjectGetString( jsonObjectGetIndex( ctx->params, 0 ) );
1500
1501         // See if we have the same authkey, and a user object,
1502         // locally cached from a previous call
1503         const char* cached_authkey = getAuthkey( ctx );
1504         if( cached_authkey && !strcmp( cached_authkey, auth ) ) {
1505                 const jsonObject* cached_user = getUserLogin( ctx );
1506                 if( cached_user )
1507                         return cached_user;
1508         }
1509
1510         // We have no matching authentication data in the cache.  Authenticate from scratch.
1511         jsonObject* auth_object = jsonNewObject(auth);
1512
1513         // Fetch the user object from the authentication server
1514         jsonObject* user = oilsUtilsQuickReq( "open-ils.auth", "open-ils.auth.session.retrieve",
1515                         auth_object );
1516         jsonObjectFree(auth_object);
1517
1518         if (!user->classname || strcmp(user->classname, "au")) {
1519
1520                 growing_buffer* msg = buffer_init(128);
1521                 buffer_fadd(
1522                         msg,
1523                         "%s: permacrud received a bad auth token: %s",
1524                         modulename,
1525                         auth
1526                 );
1527
1528                 char* m = buffer_release(msg);
1529                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_UNAUTHORIZED, "osrfMethodException",
1530                                 ctx->request, m );
1531
1532                 free(m);
1533                 jsonObjectFree(user);
1534                 user = NULL;
1535         }
1536
1537         setUserLogin( ctx, user );
1538         setAuthkey( ctx, auth );
1539
1540         // Allow ourselves up to a second before we have to reset the login timeout.
1541         // It would be nice to use some fraction of the timeout interval enforced by the
1542         // authentication server, but that value is not readily available at this point.
1543         // Instead, we use a conservative default interval.
1544         time_next_reset = time( NULL ) + 1;
1545
1546         return user;
1547 }
1548
1549 static int verifyObjectPCRUD (  osrfMethodContext* ctx, const jsonObject* obj ) {
1550
1551         dbhandle = writehandle;
1552
1553         osrfHash* method_metadata = (osrfHash*) ctx->method->userData;
1554         osrfHash* class = osrfHashGet( method_metadata, "class" );
1555         const char* method_type = osrfHashGet( method_metadata, "methodtype" );
1556         int fetch = 0;
1557
1558         if ( ( *method_type == 's' || *method_type == 'i' ) ) {
1559                 method_type = "retrieve"; // search and id_list are equivalent to retrieve for this
1560         } else if ( *method_type == 'u' || *method_type == 'd' ) {
1561                 fetch = 1; // MUST go to the db for the object for update and delete
1562         }
1563
1564         osrfHash* pcrud = osrfHashGet( osrfHashGet(class, "permacrud"), method_type );
1565
1566         if (!pcrud) {
1567                 // No permacrud for this method type on this class
1568
1569                 growing_buffer* msg = buffer_init(128);
1570                 buffer_fadd(
1571                         msg,
1572                         "%s: %s on class %s has no permacrud IDL entry",
1573                         modulename,
1574                         osrfHashGet(method_metadata, "methodtype"),
1575                         osrfHashGet(class, "classname")
1576                 );
1577
1578                 char* m = buffer_release(msg);
1579                 osrfAppSessionStatus( ctx->session, OSRF_STATUS_FORBIDDEN,
1580                                 "osrfMethodException", ctx->request, m );
1581
1582                 free(m);
1583
1584                 return 0;
1585         }
1586
1587         const jsonObject* user = verifyUserPCRUD( ctx );
1588         if (!user)
1589                 return 0;
1590
1591         int userid = atoi( oilsFMGetString( user, "id" ) );
1592
1593         osrfStringArray* permission = osrfHashGet(pcrud, "permission");
1594         osrfStringArray* local_context = osrfHashGet(pcrud, "local_context");
1595         osrfHash* foreign_context = osrfHashGet(pcrud, "foreign_context");
1596
1597         osrfStringArray* context_org_array = osrfNewStringArray(1);
1598
1599         int err = 0;
1600         char* pkey_value = NULL;
1601         if ( str_is_true( osrfHashGet(pcrud, "global_required") ) ) {
1602                 osrfLogDebug( OSRF_LOG_MARK,
1603                                 "global-level permissions required, fetching top of the org tree" );
1604
1605                 // check for perm at top of org tree
1606                 char* org_tree_root_id = org_tree_root( ctx );
1607                 if( org_tree_root_id ) {
1608                         osrfStringArrayAdd( context_org_array, org_tree_root_id );
1609                         osrfLogDebug( OSRF_LOG_MARK, "top of the org tree is %s", org_tree_root_id );
1610                 } else  {
1611                         osrfStringArrayFree( context_org_array );
1612                         return 0;
1613                 }
1614
1615         } else {
1616             osrfLogDebug( OSRF_LOG_MARK, "global-level permissions not required, "
1617                                 "fetching context org ids" );
1618             const char* pkey = osrfHashGet(class, "primarykey");
1619                 jsonObject *param = NULL;
1620
1621                 if (obj->classname) {
1622                         pkey_value = oilsFMGetString( obj, pkey );
1623                         if (!fetch)
1624                                 param = jsonObjectClone(obj);
1625                         osrfLogDebug( OSRF_LOG_MARK, "Object supplied, using primary key value of %s",
1626                                         pkey_value );
1627                 } else {
1628                         pkey_value = jsonObjectToSimpleString( obj );
1629                         fetch = 1;
1630                         osrfLogDebug( OSRF_LOG_MARK, "Object not supplied, using primary key value "
1631                                         "of %s and retrieving from the database", pkey_value );
1632                 }
1633
1634                 if (fetch) {
1635                         jsonObject* _tmp_params = single_hash( pkey, pkey_value );
1636                         jsonObject* _list = doFieldmapperSearch( ctx, class, _tmp_params, NULL, &err );
1637                         jsonObjectFree(_tmp_params);
1638
1639                         param = jsonObjectExtractIndex(_list, 0);
1640                         jsonObjectFree(_list);
1641                 }
1642
1643                 if (!param) {
1644                         osrfLogDebug( OSRF_LOG_MARK,
1645                                         "Object not found in the database with primary key %s of %s",
1646                                         pkey, pkey_value );
1647
1648                         growing_buffer* msg = buffer_init(128);
1649                         buffer_fadd(
1650                                 msg,
1651                                 "%s: no object found with primary key %s of %s",
1652                                 modulename,
1653                                 pkey,
1654                                 pkey_value
1655                         );
1656
1657                         char* m = buffer_release(msg);
1658                         osrfAppSessionStatus(
1659                                 ctx->session,
1660                                 OSRF_STATUS_INTERNALSERVERERROR,
1661                                 "osrfMethodException",
1662                                 ctx->request,
1663                                 m
1664                         );
1665
1666                         free(m);
1667                         if (pkey_value) free(pkey_value);
1668
1669                         return 0;
1670                 }
1671
1672                 if (local_context->size > 0) {
1673                         osrfLogDebug( OSRF_LOG_MARK, "%d class-local context field(s) specified",
1674                                         local_context->size);
1675                         int i = 0;
1676                         const char* lcontext = NULL;
1677                         while ( (lcontext = osrfStringArrayGetString(local_context, i++)) ) {
1678                                 osrfStringArrayAdd( context_org_array, oilsFMGetString( param, lcontext ) );
1679                                 osrfLogDebug(
1680                                         OSRF_LOG_MARK,
1681                                         "adding class-local field %s (value: %s) to the context org list",
1682                                         lcontext,
1683                                         osrfStringArrayGetString(context_org_array, context_org_array->size - 1)
1684                                 );
1685                         }
1686                 }
1687
1688                 if (foreign_context) {
1689                         unsigned long class_count = osrfHashGetCount( foreign_context );
1690                         osrfLogDebug( OSRF_LOG_MARK, "%d foreign context classes(s) specified", class_count);
1691
1692                         if (class_count > 0) {
1693
1694                                 osrfHash* fcontext = NULL;
1695                                 osrfHashIterator* class_itr = osrfNewHashIterator( foreign_context );
1696                                 while( (fcontext = osrfHashIteratorNext( class_itr ) ) ) {
1697                                         const char* class_name = osrfHashIteratorKey( class_itr );
1698                                         osrfHash* fcontext = osrfHashGet(foreign_context, class_name);
1699
1700                                         osrfLogDebug(
1701                                                 OSRF_LOG_MARK,
1702                                                 "%d foreign context fields(s) specified for class %s",
1703                                                 ((osrfStringArray*)osrfHashGet(fcontext,"context"))->size,
1704                                                 class_name
1705                                         );
1706
1707                                         char* foreign_pkey = osrfHashGet(fcontext, "field");
1708                                         char* foreign_pkey_value =
1709                                                         oilsFMGetString(param, osrfHashGet(fcontext, "fkey"));
1710
1711                                         jsonObject* _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );
1712
1713                                         jsonObject* _list = doFieldmapperSearch(
1714                                                 ctx, osrfHashGet( oilsIDL(), class_name ), _tmp_params, NULL, &err );
1715
1716                                         jsonObject* _fparam = jsonObjectClone(jsonObjectGetIndex(_list, 0));
1717                                         jsonObjectFree(_tmp_params);
1718                                         jsonObjectFree(_list);
1719
1720                                         osrfStringArray* jump_list = osrfHashGet(fcontext, "jump");
1721
1722                                         if (_fparam && jump_list) {
1723                                                 const char* flink = NULL;
1724                                                 int k = 0;
1725                                                 while ( (flink = osrfStringArrayGetString(jump_list, k++)) && _fparam ) {
1726                                                         free(foreign_pkey_value);
1727
1728                                                         osrfHash* foreign_link_hash =
1729                                                                         oilsIDLFindPath( "/%s/links/%s", _fparam->classname, flink );
1730
1731                                                         foreign_pkey_value = oilsFMGetString(_fparam, flink);
1732                                                         foreign_pkey = osrfHashGet( foreign_link_hash, "key" );
1733
1734                                                         _tmp_params = single_hash( foreign_pkey, foreign_pkey_value );
1735
1736                                                         _list = doFieldmapperSearch(
1737                                                                 ctx,
1738                                                                 osrfHashGet( oilsIDL(),
1739                                                                                 osrfHashGet( foreign_link_hash, "class" ) ),
1740                                                                 _tmp_params,
1741                                                                 NULL,
1742                                                                 &err
1743                                                         );
1744
1745                                                         _fparam = jsonObjectClone(jsonObjectGetIndex(_list, 0));
1746                                                         jsonObjectFree(_tmp_params);
1747                                                         jsonObjectFree(_list);
1748                                                 }
1749                                         }
1750
1751                                         if (!_fparam) {
1752
1753                                                 growing_buffer* msg = buffer_init(128);
1754                                                 buffer_fadd(
1755                                                         msg,
1756                                                         "%s: no object found with primary key %s of %s",
1757                                                         modulename,
1758                                                         foreign_pkey,
1759                                                         foreign_pkey_value
1760                                                 );
1761
1762                                                 char* m = buffer_release(msg);
1763                                                 osrfAppSessionStatus(
1764                                                         ctx->session,
1765                                                         OSRF_STATUS_INTERNALSERVERERROR,
1766                                                         "osrfMethodException",
1767                                                         ctx->request,
1768                                                         m
1769                                                 );
1770
1771                                                 free(m);
1772                                                 osrfHashIteratorFree(class_itr);
1773                                                 free(foreign_pkey_value);
1774                                                 jsonObjectFree(param);
1775
1776                                                 return 0;
1777                                         }
1778
1779                                         free(foreign_pkey_value);
1780
1781                                         int j = 0;
1782                                         const char* foreign_field = NULL;
1783                                         while ( (foreign_field = osrfStringArrayGetString(
1784                                                          osrfHashGet(fcontext,"context" ), j++)) ) {
1785                                                 osrfStringArrayAdd( context_org_array,
1786                                                                 oilsFMGetString( _fparam, foreign_field ) );
1787                                                 osrfLogDebug(
1788                                                         OSRF_LOG_MARK,
1789                                                         "adding foreign class %s field %s (value: %s) to the context org list",
1790                                                         class_name,
1791                                                         foreign_field,
1792                                                         osrfStringArrayGetString(
1793                                                                         context_org_array, context_org_array->size - 1)
1794                                                 );
1795                                         }
1796
1797                                         jsonObjectFree(_fparam);
1798                                 }
1799
1800                                 osrfHashIteratorFree( class_itr );
1801                         }
1802                 }
1803
1804                 jsonObjectFree(param);
1805         }
1806
1807         const char* context_org = NULL;
1808         const char* perm = NULL;
1809         int OK = 0;
1810
1811         if (permission->size == 0) {
1812             osrfLogDebug( OSRF_LOG_MARK, "No permission specified for this action, passing through" );
1813                 OK = 1;
1814         }
1815
1816         int i = 0;
1817         while ( (perm = osrfStringArrayGetString(permission, i++)) ) {
1818                 int j = 0;
1819                 while ( (context_org = osrfStringArrayGetString(context_org_array, j++)) ) {
1820                         dbi_result result;
1821
1822                         if (pkey_value) {
1823                                 osrfLogDebug(
1824                                         OSRF_LOG_MARK,
1825                                         "Checking object permission [%s] for user %d "
1826                                                         "on object %s (class %s) at org %d",
1827                                         perm,
1828                                         userid,
1829                                         pkey_value,
1830                                         osrfHashGet(class, "classname"),
1831                                         atoi(context_org)
1832                                 );
1833
1834                                 result = dbi_conn_queryf(
1835                                         writehandle,
1836                                         "SELECT permission.usr_has_object_perm(%d, '%s', '%s', '%s', %d) AS has_perm;",
1837                                         userid,
1838                                         perm,
1839                                         osrfHashGet(class, "classname"),
1840                                         pkey_value,
1841                                         atoi(context_org)
1842                                 );
1843
1844                                 if (result) {
1845                                         osrfLogDebug(
1846                                                 OSRF_LOG_MARK,
1847                                                 "Received a result for object permission [%s] "
1848                                                                 "for user %d on object %s (class %s) at org %d",
1849                                                 perm,
1850                                                 userid,
1851                                                 pkey_value,
1852                                                 osrfHashGet(class, "classname"),
1853                                                 atoi(context_org)
1854                                         );
1855
1856                                         if (dbi_result_first_row(result)) {
1857                                                 jsonObject* return_val = oilsMakeJSONFromResult( result );
1858                                                 const char* has_perm = jsonObjectGetString(
1859                                                                 jsonObjectGetKeyConst(return_val, "has_perm") );
1860
1861                                                 osrfLogDebug(
1862                                                         OSRF_LOG_MARK,
1863                                                         "Status of object permission [%s] for user %d "
1864                                                                         "on object %s (class %s) at org %d is %s",
1865                                                         perm,
1866                                                         userid,
1867                                                         pkey_value,
1868                                                         osrfHashGet(class, "classname"),
1869                                                         atoi(context_org),
1870                                                         has_perm
1871                                                 );
1872
1873                                                 if ( *has_perm == 't' ) OK = 1;
1874                                                 jsonObjectFree(return_val);
1875                                         }
1876
1877                                         dbi_result_free(result);
1878                                         if (OK)
1879                                                 break;
1880                                 }
1881                         }
1882
1883                         osrfLogDebug( OSRF_LOG_MARK,
1884                                         "Checking non-object permission [%s] for user %d at org %d",
1885                                         perm, userid, atoi(context_org) );
1886                         result = dbi_conn_queryf(
1887                                 writehandle,
1888                                 "SELECT permission.usr_has_perm(%d, '%s', %d) AS has_perm;",
1889                                 userid,
1890                                 perm,
1891                                 atoi(context_org)
1892                         );
1893
1894                         if (result) {
1895                                 osrfLogDebug( OSRF_LOG_MARK,
1896                                                 "Received a result for permission [%s] for user %d at org %d",
1897                                                 perm, userid, atoi(context_org) );
1898                                 if ( dbi_result_first_row(result) ) {
1899                                         jsonObject* return_val = oilsMakeJSONFromResult( result );
1900                                         const char* has_perm = jsonObjectGetString(
1901                                                         jsonObjectGetKeyConst(return_val, "has_perm") );
1902                                         osrfLogDebug( OSRF_LOG_MARK,
1903                                                         "Status of permission [%s] for user %d at org %d is [%s]",
1904                                                         perm, userid, atoi(context_org), has_perm );
1905                                         if ( *has_perm == 't' )
1906                                                 OK = 1;
1907                                         jsonObjectFree(return_val);
1908                                 }
1909
1910                                 dbi_result_free(result);
1911                                 if (OK) break;
1912                         }
1913
1914                 }
1915                 if (OK)
1916                         break;
1917         }
1918
1919         if (pkey_value) free(pkey_value);
1920         osrfStringArrayFree(context_org_array);
1921
1922         return OK;
1923 }
1924
1925 /**
1926         @brief Look up the root of the org_unit tree.
1927         @param ctx Pointer to the method context.
1928         @return The id of the root org unit, as a character string.
1929
1930         Query actor.org_unit where parent_ou is null, and return the id as a string.
1931
1932         This function assumes that there is only one root org unit, i.e. that we
1933         have a single tree, not a forest.
1934
1935         The calling code is responsible for freeing the returned string.
1936 */
1937 static char* org_tree_root( osrfMethodContext* ctx ) {
1938
1939         static char cached_root_id[ 32 ] = "";  // extravagantly large buffer
1940         static time_t last_lookup_time = 0;
1941         time_t current_time = time( NULL );
1942
1943         if( cached_root_id[ 0 ] && ( current_time - last_lookup_time < 3600 ) ) {
1944                 // We successfully looked this up less than an hour ago.
1945                 // It's not likely to have changed since then.
1946                 return strdup( cached_root_id );
1947         }
1948         last_lookup_time = current_time;
1949
1950         int err = 0;
1951         jsonObject* where_clause = single_hash( "parent_ou", NULL );
1952         jsonObject* result = doFieldmapperSearch(
1953                 ctx, osrfHashGet( oilsIDL(), "aou" ), where_clause, NULL, &err );
1954         jsonObjectFree( where_clause );
1955
1956         jsonObject* tree_top = jsonObjectGetIndex( result, 0 );
1957
1958         if (! tree_top) {
1959                 jsonObjectFree( result );
1960
1961                 growing_buffer* msg = buffer_init(128);
1962                 OSRF_BUFFER_ADD( msg, modulename );
1963                 OSRF_BUFFER_ADD( msg,
1964                                 ": Internal error, could not find the top of the org tree (parent_ou = NULL)" );
1965
1966                 char* m = buffer_release(msg);
1967                 osrfAppSessionStatus( ctx->session,
1968                                 OSRF_STATUS_INTERNALSERVERERROR, "osrfMethodException", ctx->request, m );
1969                 free(m);
1970
1971                 cached_root_id[ 0 ] = '\0';
1972                 return NULL;
1973         }
1974
1975         char* root_org_unit_id = oilsFMGetString( tree_top, "id" );
1976         osrfLogDebug( OSRF_LOG_MARK, "Top of the org tree is %s", root_org_unit_id );
1977
1978         jsonObjectFree( result );
1979
1980         strcpy( cached_root_id, root_org_unit_id );
1981         return root_org_unit_id;
1982 }
1983
1984 /**
1985         @brief Create a JSON_HASH with a single key/value pair.
1986         @param key The key of the key/value pair.
1987         @param value the value of the key/value pair.
1988         @return Pointer to a newly created jsonObject of type JSON_HASH.
1989
1990         The value of the key/value is either a string or (if @a value is NULL) a null.
1991 */
1992 static jsonObject* single_hash( const char* key, const char* value ) {
1993         // Sanity check
1994         if( ! key ) key = "";
1995
1996         jsonObject* hash = jsonNewObjectType( JSON_HASH );
1997         jsonObjectSetKey( hash, key, jsonNewObject( value ) );
1998         return hash;
1999 }
2000
2001
2002 static int doCreate( osrfMethodContext* ctx ) {
2003         if(osrfMethodVerifyContext( ctx )) {
2004                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
2005                 return -1;
2006         }
2007
2008         if( enforce_pcrud )
2009                 timeout_needs_resetting = 1;
2010
2011         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
2012         jsonObject* target = NULL;
2013         jsonObject* options = NULL;
2014
2015         if( enforce_pcrud ) {
2016                 target = jsonObjectGetIndex( ctx->params, 1 );
2017                 options = jsonObjectGetIndex( ctx->params, 2 );
2018         } else {
2019                 target = jsonObjectGetIndex( ctx->params, 0 );
2020                 options = jsonObjectGetIndex( ctx->params, 1 );
2021         }
2022
2023         if ( !verifyObjectClass( ctx, target )) {
2024                 osrfAppRespondComplete( ctx, NULL );
2025                 return -1;
2026         }
2027
2028         osrfLogDebug( OSRF_LOG_MARK, "Object seems to be of the correct type" );
2029
2030         const char* trans_id = getXactId( ctx );
2031         if ( !trans_id ) {
2032                 osrfLogError( OSRF_LOG_MARK, "No active transaction -- required for CREATE" );
2033
2034                 osrfAppSessionStatus(
2035                         ctx->session,
2036                         OSRF_STATUS_BADREQUEST,
2037                         "osrfMethodException",
2038                         ctx->request,
2039                         "No active transaction -- required for CREATE"
2040                 );
2041                 osrfAppRespondComplete( ctx, NULL );
2042                 return -1;
2043         }
2044
2045         // The following test is harmless but redundant.  If a class is
2046         // readonly, we don't register a create method for it.
2047         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
2048                 osrfAppSessionStatus(
2049                         ctx->session,
2050                         OSRF_STATUS_BADREQUEST,
2051                         "osrfMethodException",
2052                         ctx->request,
2053                         "Cannot INSERT readonly class"
2054                 );
2055                 osrfAppRespondComplete( ctx, NULL );
2056                 return -1;
2057         }
2058
2059         // Set the last_xact_id
2060         int index = oilsIDL_ntop( target->classname, "last_xact_id" );
2061         if (index > -1) {
2062                 osrfLogDebug(OSRF_LOG_MARK, "Setting last_xact_id to %s on %s at position %d",
2063                         trans_id, target->classname, index);
2064                 jsonObjectSetIndex(target, index, jsonNewObject(trans_id));
2065         }
2066
2067         osrfLogDebug( OSRF_LOG_MARK, "There is a transaction running..." );
2068
2069         dbhandle = writehandle;
2070
2071         osrfHash* fields = osrfHashGet(meta, "fields");
2072         char* pkey = osrfHashGet(meta, "primarykey");
2073         char* seq = osrfHashGet(meta, "sequence");
2074
2075         growing_buffer* table_buf = buffer_init(128);
2076         growing_buffer* col_buf = buffer_init(128);
2077         growing_buffer* val_buf = buffer_init(128);
2078
2079         OSRF_BUFFER_ADD(table_buf, "INSERT INTO ");
2080         OSRF_BUFFER_ADD(table_buf, osrfHashGet(meta, "tablename"));
2081         OSRF_BUFFER_ADD_CHAR( col_buf, '(' );
2082         buffer_add(val_buf,"VALUES (");
2083
2084
2085         int first = 1;
2086         osrfHash* field = NULL;
2087         osrfHashIterator* field_itr = osrfNewHashIterator( fields );
2088         while( (field = osrfHashIteratorNext( field_itr ) ) ) {
2089
2090                 const char* field_name = osrfHashIteratorKey( field_itr );
2091
2092                 if( str_is_true( osrfHashGet( field, "virtual" ) ) )
2093                         continue;
2094
2095                 const jsonObject* field_object = oilsFMGetObject( target, field_name );
2096
2097                 char* value;
2098                 if (field_object && field_object->classname) {
2099                         value = oilsFMGetString(
2100                                 field_object,
2101                                 (char*)oilsIDLFindPath("/%s/primarykey", field_object->classname)
2102                         );
2103                 } else if( field_object && JSON_BOOL == field_object->type ) {
2104                         if( jsonBoolIsTrue( field_object ) )
2105                                 value = strdup( "t" );
2106                         else
2107                                 value = strdup( "f" );
2108                 } else {
2109                         value = jsonObjectToSimpleString( field_object );
2110                 }
2111
2112                 if (first) {
2113                         first = 0;
2114                 } else {
2115                         OSRF_BUFFER_ADD_CHAR( col_buf, ',' );
2116                         OSRF_BUFFER_ADD_CHAR( val_buf, ',' );
2117                 }
2118
2119                 buffer_add(col_buf, field_name);
2120
2121                 if (!field_object || field_object->type == JSON_NULL) {
2122                         buffer_add( val_buf, "DEFAULT" );
2123
2124                 } else if ( !strcmp(get_primitive( field ), "number") ) {
2125                         const char* numtype = get_datatype( field );
2126                         if ( !strcmp( numtype, "INT8") ) {
2127                                 buffer_fadd( val_buf, "%lld", atoll(value) );
2128
2129                         } else if ( !strcmp( numtype, "INT") ) {
2130                                 buffer_fadd( val_buf, "%d", atoi(value) );
2131
2132                         } else if ( !strcmp( numtype, "NUMERIC") ) {
2133                                 buffer_fadd( val_buf, "%f", atof(value) );
2134                         }
2135                 } else {
2136                         if ( dbi_conn_quote_string(writehandle, &value) ) {
2137                                 OSRF_BUFFER_ADD( val_buf, value );
2138
2139                         } else {
2140                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]", modulename, value);
2141                                 osrfAppSessionStatus(
2142                                         ctx->session,
2143                                         OSRF_STATUS_INTERNALSERVERERROR,
2144                                         "osrfMethodException",
2145                                         ctx->request,
2146                                         "Error quoting string -- please see the error log for more details"
2147                                 );
2148                                 free(value);
2149                                 buffer_free(table_buf);
2150                                 buffer_free(col_buf);
2151                                 buffer_free(val_buf);
2152                                 osrfAppRespondComplete( ctx, NULL );
2153                                 return -1;
2154                         }
2155                 }
2156
2157                 free(value);
2158
2159         }
2160
2161         osrfHashIteratorFree( field_itr );
2162
2163         OSRF_BUFFER_ADD_CHAR( col_buf, ')' );
2164         OSRF_BUFFER_ADD_CHAR( val_buf, ')' );
2165
2166         char* table_str = buffer_release(table_buf);
2167         char* col_str   = buffer_release(col_buf);
2168         char* val_str   = buffer_release(val_buf);
2169         growing_buffer* sql = buffer_init(128);
2170         buffer_fadd( sql, "%s %s %s;", table_str, col_str, val_str );
2171         free(table_str);
2172         free(col_str);
2173         free(val_str);
2174
2175         char* query = buffer_release(sql);
2176
2177         osrfLogDebug(OSRF_LOG_MARK, "%s: Insert SQL [%s]", modulename, query);
2178
2179         jsonObject* obj = NULL;
2180         int rc = 0;
2181
2182         dbi_result result = dbi_conn_query( writehandle, query );
2183         if (!result) {
2184                 obj = jsonNewObject(NULL);
2185                 osrfLogError(
2186                         OSRF_LOG_MARK,
2187                         "%s ERROR inserting %s object using query [%s]",
2188                         modulename,
2189                         osrfHashGet(meta, "fieldmapper"),
2190                         query
2191                 );
2192                 osrfAppSessionStatus(
2193                         ctx->session,
2194                         OSRF_STATUS_INTERNALSERVERERROR,
2195                         "osrfMethodException",
2196                         ctx->request,
2197                         "INSERT error -- please see the error log for more details"
2198                 );
2199                 rc = -1;
2200         } else {
2201
2202                 char* id = oilsFMGetString(target, pkey);
2203                 if (!id) {
2204                         unsigned long long new_id = dbi_conn_sequence_last(writehandle, seq);
2205                         growing_buffer* _id = buffer_init(10);
2206                         buffer_fadd(_id, "%lld", new_id);
2207                         id = buffer_release(_id);
2208                 }
2209
2210                 // Find quietness specification, if present
2211                 const char* quiet_str = NULL;
2212                 if ( options ) {
2213                         const jsonObject* quiet_obj = jsonObjectGetKeyConst( options, "quiet" );
2214                         if( quiet_obj )
2215                                 quiet_str = jsonObjectGetString( quiet_obj );
2216                 }
2217
2218                 if( str_is_true( quiet_str ) ) {  // if quietness is specified
2219                         obj = jsonNewObject(id);
2220                 }
2221                 else {
2222
2223                         // Fetch the row that we just inserted, so that we can return it to the client
2224                         jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
2225                         jsonObjectSetKey( where_clause, pkey, jsonNewObject(id) );
2226
2227                         int err = 0;
2228                         jsonObject* list = doFieldmapperSearch( ctx, meta, where_clause, NULL, &err );
2229                         if( err )
2230                                 rc = -1;
2231                         else
2232                                 obj = jsonObjectClone( jsonObjectGetIndex( list, 0 ));
2233
2234                         jsonObjectFree( list );
2235                         jsonObjectFree( where_clause );
2236                 }
2237
2238                 free(id);
2239         }
2240
2241         free(query);
2242         osrfAppRespondComplete( ctx, obj );
2243         jsonObjectFree( obj );
2244         return rc;
2245 }
2246
2247 /**
2248         @brief Implement the retrieve method.
2249         @param ctx Pointer to the method context.
2250         @param err Pointer through which to return an error code.
2251         @return If successful, a pointer to the result to be returned to the client;
2252         otherwise NULL.
2253
2254         From the method's class, fetch a row with a specified value in the primary key.  This
2255         method relies on the database design convention that a primary key consists of a single
2256         column.
2257
2258         Method parameters:
2259         - authkey (PCRUD only)
2260         - value of the primary key for the desired row, for building the WHERE clause
2261         - a JSON_HASH containing any other SQL clauses: select, join, etc.
2262
2263         Return to client: One row from the query.
2264 */
2265 static int doRetrieve(osrfMethodContext* ctx ) {
2266         if(osrfMethodVerifyContext( ctx )) {
2267                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
2268                 return -1;
2269         }
2270
2271         if( enforce_pcrud )
2272                 timeout_needs_resetting = 1;
2273
2274         int id_pos = 0;
2275         int order_pos = 1;
2276
2277         if( enforce_pcrud ) {
2278                 id_pos = 1;
2279                 order_pos = 2;
2280         }
2281
2282         // Get the class metadata
2283         osrfHash* class_def = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
2284
2285         // Get the value of the primary key, from a method parameter
2286         const jsonObject* id_obj = jsonObjectGetIndex(ctx->params, id_pos);
2287
2288         osrfLogDebug(
2289                 OSRF_LOG_MARK,
2290                 "%s retrieving %s object with primary key value of %s",
2291                 modulename,
2292                 osrfHashGet( class_def, "fieldmapper" ),
2293                 jsonObjectGetString( id_obj )
2294         );
2295
2296         // Build a WHERE clause based on the key value
2297         jsonObject* where_clause = jsonNewObjectType( JSON_HASH );
2298         jsonObjectSetKey(
2299                 where_clause,
2300                 osrfHashGet( class_def, "primarykey" ),  // name of key column
2301                 jsonObjectClone( id_obj )                // value of key column
2302         );
2303
2304         jsonObject* rest_of_query = jsonObjectGetIndex(ctx->params, order_pos);
2305
2306         // Do the query
2307         int err = 0;
2308         jsonObject* list = doFieldmapperSearch( ctx, class_def, where_clause, rest_of_query, &err );
2309
2310         jsonObjectFree( where_clause );
2311         if( err ) {
2312                 osrfAppRespondComplete( ctx, NULL );
2313                 return -1;
2314         }
2315
2316         jsonObject* obj = jsonObjectExtractIndex( list, 0 );
2317         jsonObjectFree( list );
2318
2319         if( enforce_pcrud ) {
2320                 if(!verifyObjectPCRUD(ctx, obj)) {
2321                         jsonObjectFree(obj);
2322
2323                         growing_buffer* msg = buffer_init(128);
2324                         OSRF_BUFFER_ADD( msg, modulename );
2325                         OSRF_BUFFER_ADD( msg, ": Insufficient permissions to retrieve object" );
2326
2327                         char* m = buffer_release(msg);
2328                         osrfAppSessionStatus( ctx->session, OSRF_STATUS_NOTALLOWED, "osrfMethodException",
2329                                         ctx->request, m );
2330                         free(m);
2331
2332                         osrfAppRespondComplete( ctx, NULL );
2333                         return -1;
2334                 }
2335         }
2336
2337         osrfAppRespondComplete( ctx, obj );
2338         jsonObjectFree( obj );
2339         return 0;
2340 }
2341
2342 static char* jsonNumberToDBString ( osrfHash* field, const jsonObject* value ) {
2343         growing_buffer* val_buf = buffer_init(32);
2344         const char* numtype = get_datatype( field );
2345
2346         if ( !strncmp( numtype, "INT", 3 ) ) {
2347                 if (value->type == JSON_NUMBER)
2348                         //buffer_fadd( val_buf, "%ld", (long)jsonObjectGetNumber(value) );
2349                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2350                 else {
2351                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2352                 }
2353
2354         } else if ( !strcmp( numtype, "NUMERIC" ) ) {
2355                 if (value->type == JSON_NUMBER)
2356                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2357                 else {
2358                         buffer_fadd( val_buf, jsonObjectGetString( value ) );
2359                 }
2360
2361         } else {
2362                 // Presumably this was really intended ot be a string, so quote it
2363                 char* str = jsonObjectToSimpleString( value );
2364                 if ( dbi_conn_quote_string(dbhandle, &str) ) {
2365                         OSRF_BUFFER_ADD( val_buf, str );
2366                         free(str);
2367                 } else {
2368                         osrfLogError(OSRF_LOG_MARK, "%s: Error quoting key string [%s]", modulename, str);
2369                         free(str);
2370                         buffer_free(val_buf);
2371                         return NULL;
2372                 }
2373         }
2374
2375         return buffer_release(val_buf);
2376 }
2377
2378 static char* searchINPredicate (const char* class_alias, osrfHash* field,
2379                 jsonObject* node, const char* op, osrfMethodContext* ctx ) {
2380         growing_buffer* sql_buf = buffer_init(32);
2381
2382         buffer_fadd(
2383                 sql_buf,
2384                 "\"%s\".%s ",
2385                 class_alias,
2386                 osrfHashGet(field, "name")
2387         );
2388
2389         if (!op) {
2390                 buffer_add(sql_buf, "IN (");
2391         } else if (!(strcasecmp(op,"not in"))) {
2392                 buffer_add(sql_buf, "NOT IN (");
2393         } else {
2394                 buffer_add(sql_buf, "IN (");
2395         }
2396
2397         if (node->type == JSON_HASH) {
2398                 // subquery predicate
2399                 char* subpred = buildQuery( ctx, node, SUBSELECT );
2400                 if( ! subpred ) {
2401                         buffer_free( sql_buf );
2402                         return NULL;
2403                 }
2404
2405                 buffer_add(sql_buf, subpred);
2406                 free(subpred);
2407
2408         } else if (node->type == JSON_ARRAY) {
2409                 // literal value list
2410                 int in_item_index = 0;
2411                 int in_item_first = 1;
2412                 const jsonObject* in_item;
2413                 while ( (in_item = jsonObjectGetIndex(node, in_item_index++)) ) {
2414
2415                         if (in_item_first)
2416                                 in_item_first = 0;
2417                         else
2418                                 buffer_add(sql_buf, ", ");
2419
2420                         // Sanity check
2421                         if ( in_item->type != JSON_STRING && in_item->type != JSON_NUMBER ) {
2422                                 osrfLogError( OSRF_LOG_MARK,
2423                                                 "%s: Expected string or number within IN list; found %s",
2424                                                 modulename, json_type( in_item->type ) );
2425                                 buffer_free(sql_buf);
2426                                 return NULL;
2427                         }
2428
2429                         // Append the literal value -- quoted if not a number
2430                         if ( JSON_NUMBER == in_item->type ) {
2431                                 char* val = jsonNumberToDBString( field, in_item );
2432                                 OSRF_BUFFER_ADD( sql_buf, val );
2433                                 free(val);
2434
2435                         } else if ( !strcmp( get_primitive( field ), "number") ) {
2436                                 char* val = jsonNumberToDBString( field, in_item );
2437                                 OSRF_BUFFER_ADD( sql_buf, val );
2438                                 free(val);
2439
2440                         } else {
2441                                 char* key_string = jsonObjectToSimpleString(in_item);
2442                                 if ( dbi_conn_quote_string(dbhandle, &key_string) ) {
2443                                         OSRF_BUFFER_ADD( sql_buf, key_string );
2444                                         free(key_string);
2445                                 } else {
2446                                         osrfLogError(OSRF_LOG_MARK,
2447                                                         "%s: Error quoting key string [%s]", modulename, key_string);
2448                                         free(key_string);
2449                                         buffer_free(sql_buf);
2450                                         return NULL;
2451                                 }
2452                         }
2453                 }
2454
2455                 if( in_item_first ) {
2456                         osrfLogError(OSRF_LOG_MARK, "%s: Empty IN list", modulename );
2457                         buffer_free( sql_buf );
2458                         return NULL;
2459                 }
2460         } else {
2461                 osrfLogError(OSRF_LOG_MARK, "%s: Expected object or array for IN clause; found %s",
2462                         modulename, json_type( node->type ) );
2463                 buffer_free(sql_buf);
2464                 return NULL;
2465         }
2466
2467         OSRF_BUFFER_ADD_CHAR( sql_buf, ')' );
2468
2469         return buffer_release(sql_buf);
2470 }
2471
2472 // Receive a JSON_ARRAY representing a function call.  The first
2473 // entry in the array is the function name.  The rest are parameters.
2474 static char* searchValueTransform( const jsonObject* array ) {
2475
2476         if( array->size < 1 ) {
2477                 osrfLogError(OSRF_LOG_MARK, "%s: Empty array for value transform", MODULENAME);
2478                 return NULL;
2479         }
2480
2481         // Get the function name
2482         jsonObject* func_item = jsonObjectGetIndex( array, 0 );
2483         if( func_item->type != JSON_STRING ) {
2484                 osrfLogError(OSRF_LOG_MARK, "%s: Error: expected function name, found %s",
2485                         MODULENAME, json_type( func_item->type ) );
2486                 return NULL;
2487         }
2488
2489         growing_buffer* sql_buf = buffer_init(32);
2490
2491         OSRF_BUFFER_ADD( sql_buf, jsonObjectGetString( func_item ) );
2492         OSRF_BUFFER_ADD( sql_buf, "( " );
2493
2494         // Get the parameters
2495         int func_item_index = 1;   // We already grabbed the zeroth entry
2496         while ( (func_item = jsonObjectGetIndex(array, func_item_index++)) ) {
2497
2498                 // Add a separator comma, if we need one
2499                 if( func_item_index > 2 )
2500                         buffer_add( sql_buf, ", " );
2501
2502                 // Add the current parameter
2503                 if (func_item->type == JSON_NULL) {
2504                         buffer_add( sql_buf, "NULL" );
2505                 } else {
2506                         char* val = jsonObjectToSimpleString(func_item);
2507                         if ( dbi_conn_quote_string(dbhandle, &val) ) {
2508                                 OSRF_BUFFER_ADD( sql_buf, val );
2509                                 free(val);
2510                         } else {
2511                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting key string [%s]", MODULENAME, val);
2512                                 buffer_free(sql_buf);
2513                                 free(val);
2514                                 return NULL;
2515                         }
2516                 }
2517         }
2518
2519         buffer_add( sql_buf, " )" );
2520
2521         return buffer_release(sql_buf);
2522 }
2523
2524 static char* searchFunctionPredicate (const char* class_alias, osrfHash* field,
2525                 const jsonObject* node, const char* op) {
2526
2527         if( ! is_good_operator( op ) ) {
2528                 osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", MODULENAME, op );
2529                 return NULL;
2530         }
2531
2532         char* val = searchValueTransform(node);
2533         if( !val )
2534                 return NULL;
2535
2536         growing_buffer* sql_buf = buffer_init(32);
2537         buffer_fadd(
2538                 sql_buf,
2539                 "\"%s\".%s %s %s",
2540                 class_alias,
2541                 osrfHashGet(field, "name"),
2542                 op,
2543                 val
2544         );
2545
2546         free(val);
2547
2548         return buffer_release(sql_buf);
2549 }
2550
2551 // class_alias is a class name or other table alias
2552 // field is a field definition as stored in the IDL
2553 // node comes from the method parameter, and may represent an entry in the SELECT list
2554 static char* searchFieldTransform (const char* class_alias, osrfHash* field, const jsonObject* node) {
2555         growing_buffer* sql_buf = buffer_init(32);
2556
2557         const char* field_transform = jsonObjectGetString(
2558                 jsonObjectGetKeyConst( node, "transform" ) );
2559         const char* transform_subcolumn = jsonObjectGetString(
2560                 jsonObjectGetKeyConst( node, "result_field" ) );
2561
2562         if(transform_subcolumn) {
2563                 if( ! is_identifier( transform_subcolumn ) ) {
2564                         osrfLogError( OSRF_LOG_MARK, "%s: Invalid subfield name: \"%s\"\n",
2565                                         MODULENAME, transform_subcolumn );
2566                         buffer_free( sql_buf );
2567                         return NULL;
2568                 }
2569                 OSRF_BUFFER_ADD_CHAR( sql_buf, '(' );    // enclose transform in parentheses
2570         }
2571
2572         if (field_transform) {
2573
2574                 if( ! is_identifier( field_transform ) ) {
2575                         osrfLogError( OSRF_LOG_MARK, "%s: Expected function name, found \"%s\"\n",
2576                                         MODULENAME, field_transform );
2577                         buffer_free( sql_buf );
2578                         return NULL;
2579                 }
2580
2581                 if( obj_is_true( jsonObjectGetKeyConst( node, "distinct" ) ) ) {
2582                         buffer_fadd( sql_buf, "%s(DISTINCT \"%s\".%s",
2583                                 field_transform, class_alias, osrfHashGet(field, "name"));
2584                 } else {
2585                         buffer_fadd( sql_buf, "%s(\"%s\".%s",
2586                                 field_transform, class_alias, osrfHashGet(field, "name"));
2587                 }
2588
2589                 const jsonObject* array = jsonObjectGetKeyConst( node, "params" );
2590
2591                 if (array) {
2592                         if( array->type != JSON_ARRAY ) {
2593                                 osrfLogError( OSRF_LOG_MARK,
2594                                         "%s: Expected JSON_ARRAY for function params; found %s",
2595                                         MODULENAME, json_type( array->type ) );
2596                                 buffer_free( sql_buf );
2597                                 return NULL;
2598                         }
2599                         int func_item_index = 0;
2600                         jsonObject* func_item;
2601                         while ( (func_item = jsonObjectGetIndex(array, func_item_index++)) ) {
2602
2603                                 char* val = jsonObjectToSimpleString(func_item);
2604
2605                                 if ( !val ) {
2606                                         buffer_add( sql_buf, ",NULL" );
2607                                 } else if ( dbi_conn_quote_string(dbhandle, &val) ) {
2608                                         OSRF_BUFFER_ADD_CHAR( sql_buf, ',' );
2609                                         OSRF_BUFFER_ADD( sql_buf, val );
2610                                 } else {
2611                                         osrfLogError( OSRF_LOG_MARK,
2612                                                         "%s: Error quoting key string [%s]", MODULENAME, val);
2613                                         free(val);
2614                                         buffer_free(sql_buf);
2615                                         return NULL;
2616                                 }
2617                                 free(val);
2618                         }
2619                 }
2620
2621                 buffer_add( sql_buf, " )" );
2622
2623         } else {
2624                 buffer_fadd( sql_buf, "\"%s\".%s", class_alias, osrfHashGet(field, "name"));
2625         }
2626
2627         if (transform_subcolumn)
2628                 buffer_fadd( sql_buf, ").\"%s\"", transform_subcolumn );
2629
2630         return buffer_release(sql_buf);
2631 }
2632
2633 static char* searchFieldTransformPredicate( const ClassInfo* class_info, osrfHash* field,
2634                 const jsonObject* node, const char* op ) {
2635
2636         if( ! is_good_operator( op ) ) {
2637                 osrfLogError(OSRF_LOG_MARK, "%s: Error: Invalid operator %s", MODULENAME, op);
2638                 return NULL;
2639         }
2640
2641         char* field_transform = searchFieldTransform( class_info->alias, field, node );
2642         if( ! field_transform )
2643                 return NULL;
2644         char* value = NULL;
2645         int extra_parens = 0;   // boolean
2646
2647         const jsonObject* value_obj = jsonObjectGetKeyConst( node, "value" );
2648         if ( ! value_obj ) {
2649                 value = searchWHERE( node, class_info, AND_OP_JOIN, NULL );
2650                 if( !value ) {
2651                         osrfLogError( OSRF_LOG_MARK, "%s: Error building condition for field transform",
2652                                 MODULENAME );
2653                         free(field_transform);
2654                         return NULL;
2655                 }
2656                 extra_parens = 1;
2657         } else if ( value_obj->type == JSON_ARRAY ) {
2658                 value = searchValueTransform( value_obj );
2659                 if( !value ) {
2660                         osrfLogError(OSRF_LOG_MARK,
2661                                 "%s: Error building value transform for field transform", MODULENAME);
2662                         free( field_transform );
2663                         return NULL;
2664                 }
2665         } else if ( value_obj->type == JSON_HASH ) {
2666                 value = searchWHERE( value_obj, class_info, AND_OP_JOIN, NULL );
2667                 if( !value ) {
2668                         osrfLogError(OSRF_LOG_MARK, "%s: Error building predicate for field transform",
2669                                 MODULENAME);
2670                         free(field_transform);
2671                         return NULL;
2672                 }
2673                 extra_parens = 1;
2674         } else if ( value_obj->type == JSON_NUMBER ) {
2675                 value = jsonNumberToDBString( field, value_obj );
2676         } else if ( value_obj->type == JSON_NULL ) {
2677                 osrfLogError( OSRF_LOG_MARK,
2678                         "%s: Error building predicate for field transform: null value", MODULENAME );
2679                 free(field_transform);
2680                 return NULL;
2681         } else if ( value_obj->type == JSON_BOOL ) {
2682                 osrfLogError( OSRF_LOG_MARK,
2683                         "%s: Error building predicate for field transform: boolean value", MODULENAME );
2684                 free(field_transform);
2685                 return NULL;
2686         } else {
2687                 if ( !strcmp( get_primitive( field ), "number") ) {
2688                         value = jsonNumberToDBString( field, value_obj );
2689                 } else {
2690                         value = jsonObjectToSimpleString( value_obj );
2691                         if ( !dbi_conn_quote_string(dbhandle, &value) ) {
2692                                 osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
2693                                         MODULENAME, value);
2694                                 free(value);
2695                                 free(field_transform);
2696                                 return NULL;
2697                         }
2698                 }
2699         }
2700
2701         const char* left_parens  = "";
2702         const char* right_parens = "";
2703
2704         if( extra_parens ) {
2705                 left_parens  = "(";
2706                 right_parens = ")";
2707         }
2708
2709         growing_buffer* sql_buf = buffer_init(32);
2710
2711         buffer_fadd(
2712                 sql_buf,
2713                 "%s%s %s %s %s %s%s",
2714                 left_parens,
2715                 field_transform,
2716                 op,
2717                 left_parens,
2718                 value,
2719                 right_parens,
2720                 right_parens
2721         );
2722
2723         free(value);
2724         free(field_transform);
2725
2726         return buffer_release(sql_buf);
2727 }
2728
2729 static char* searchSimplePredicate (const char* op, const char* class_alias,
2730                 osrfHash* field, const jsonObject* node) {
2731
2732         if( ! is_good_operator( op ) ) {
2733                 osrfLogError( OSRF_LOG_MARK, "%s: Invalid operator [%s]", MODULENAME, op );
2734                 return NULL;
2735         }
2736
2737         char* val = NULL;
2738
2739         // Get the value to which we are comparing the specified column
2740         if (node->type != JSON_NULL) {
2741                 if ( node->type == JSON_NUMBER ) {
2742                         val = jsonNumberToDBString( field, node );
2743                 } else if ( !strcmp( get_primitive( field ), "number" ) ) {
2744                         val = jsonNumberToDBString( field, node );
2745                 } else {
2746                         val = jsonObjectToSimpleString(node);
2747                 }
2748         }
2749
2750         if( val ) {
2751                 if( JSON_NUMBER != node->type && strcmp( get_primitive( field ), "number") ) {
2752                         // Value is not numeric; enclose it in quotes
2753                         if ( !dbi_conn_quote_string( dbhandle, &val ) ) {
2754                                 osrfLogError( OSRF_LOG_MARK, "%s: Error quoting key string [%s]",
2755                                         MODULENAME, val );
2756                                 free( val );
2757                                 return NULL;
2758                         }
2759                 }
2760         } else {
2761                 // Compare to a null value
2762                 val = strdup( "NULL" );
2763                 if (strcmp( op, "=" ))
2764                         op = "IS NOT";
2765                 else
2766                         op = "IS";
2767         }
2768
2769         growing_buffer* sql_buf = buffer_init(32);
2770         buffer_fadd( sql_buf, "\"%s\".%s %s %s", class_alias, osrfHashGet(field, "name"), op, val );
2771         char* pred = buffer_release( sql_buf );
2772
2773         free(val);
2774
2775         return pred;
2776 }
2777
2778 static char* searchBETWEENPredicate (const char* class_alias,
2779                 osrfHash* field, const jsonObject* node) {
2780
2781         const jsonObject* x_node = jsonObjectGetIndex( node, 0 );
2782         const jsonObject* y_node = jsonObjectGetIndex( node, 1 );
2783
2784         if( NULL == y_node ) {
2785                 osrfLogError( OSRF_LOG_MARK, "%s: Not enough operands for BETWEEN operator", MODULENAME );
2786                 return NULL;
2787         }
2788         else if( NULL != jsonObjectGetIndex( node, 2 ) ) {
2789                 osrfLogError( OSRF_LOG_MARK, "%s: Too many operands for BETWEEN operator", MODULENAME );
2790                 return NULL;
2791         }
2792
2793         char* x_string;
2794         char* y_string;
2795
2796         if ( !strcmp( get_primitive( field ), "number") ) {
2797                 x_string = jsonNumberToDBString(field, x_node);
2798                 y_string = jsonNumberToDBString(field, y_node);
2799
2800         } else {
2801                 x_string = jsonObjectToSimpleString(x_node);
2802                 y_string = jsonObjectToSimpleString(y_node);
2803                 if ( !(dbi_conn_quote_string(dbhandle, &x_string) && dbi_conn_quote_string(dbhandle, &y_string)) ) {
2804                         osrfLogError(OSRF_LOG_MARK, "%s: Error quoting key strings [%s] and [%s]",
2805                                         MODULENAME, x_string, y_string);
2806                         free(x_string);
2807                         free(y_string);
2808                         return NULL;
2809                 }
2810         }
2811
2812         growing_buffer* sql_buf = buffer_init(32);
2813         buffer_fadd( sql_buf, "\"%s\".%s BETWEEN %s AND %s",
2814                         class_alias, osrfHashGet(field, "name"), x_string, y_string );
2815         free(x_string);
2816         free(y_string);
2817
2818         return buffer_release(sql_buf);
2819 }
2820
2821 static char* searchPredicate ( const ClassInfo* class_info, osrfHash* field,
2822                                                            jsonObject* node, osrfMethodContext* ctx ) {
2823
2824         char* pred = NULL;
2825         if (node->type == JSON_ARRAY) { // equality IN search
2826                 pred = searchINPredicate( class_info->alias, field, node, NULL, ctx );
2827         } else if (node->type == JSON_HASH) { // other search
2828                 jsonIterator* pred_itr = jsonNewIterator( node );
2829                 if( !jsonIteratorHasNext( pred_itr ) ) {
2830                         osrfLogError( OSRF_LOG_MARK, "%s: Empty predicate for field \"%s\"",
2831                                         MODULENAME, osrfHashGet(field, "name") );
2832                 } else {
2833                         jsonObject* pred_node = jsonIteratorNext( pred_itr );
2834
2835                         // Verify that there are no additional predicates
2836                         if( jsonIteratorHasNext( pred_itr ) ) {
2837                                 osrfLogError( OSRF_LOG_MARK, "%s: Multiple predicates for field \"%s\"",
2838                                                 MODULENAME, osrfHashGet(field, "name") );
2839                         } else if ( !(strcasecmp( pred_itr->key,"between" )) )
2840                                 pred = searchBETWEENPredicate( class_info->alias, field, pred_node );
2841                         else if ( !(strcasecmp( pred_itr->key,"in" ))
2842                                         || !(strcasecmp( pred_itr->key,"not in" )) )
2843                                 pred = searchINPredicate(
2844                                         class_info->alias, field, pred_node, pred_itr->key, ctx );
2845                         else if ( pred_node->type == JSON_ARRAY )
2846                                 pred = searchFunctionPredicate(
2847                                         class_info->alias, field, pred_node, pred_itr->key );
2848                         else if ( pred_node->type == JSON_HASH )
2849                                 pred = searchFieldTransformPredicate(
2850                                         class_info, field, pred_node, pred_itr->key );
2851                         else
2852                                 pred = searchSimplePredicate( pred_itr->key, class_info->alias, field, pred_node );
2853                 }
2854                 jsonIteratorFree(pred_itr);
2855
2856         } else if (node->type == JSON_NULL) { // IS NULL search
2857                 growing_buffer* _p = buffer_init(64);
2858                 buffer_fadd(
2859                         _p,
2860                         "\"%s\".%s IS NULL",
2861                         class_info->class_name,
2862                         osrfHashGet(field, "name")
2863                 );
2864                 pred = buffer_release(_p);
2865         } else { // equality search
2866                 pred = searchSimplePredicate( "=", class_info->alias, field, node );
2867         }
2868
2869         return pred;
2870
2871 }
2872
2873
2874 /*
2875
2876 join : {
2877         acn : {
2878                 field : record,
2879                 fkey : id
2880                 type : left
2881                 filter_op : or
2882                 filter : { ... },
2883                 join : {
2884                         acp : {
2885                                 field : call_number,
2886                                 fkey : id,
2887                                 filter : { ... },
2888                         },
2889                 },
2890         },
2891         mrd : {
2892                 field : record,
2893                 type : inner
2894                 fkey : id,
2895                 filter : { ... },
2896         }
2897 }
2898
2899 */
2900
2901 static char* searchJOIN ( const jsonObject* join_hash, const ClassInfo* left_info ) {
2902
2903         const jsonObject* working_hash;
2904         jsonObject* freeable_hash = NULL;
2905
2906         if (join_hash->type == JSON_HASH) {
2907                 working_hash = join_hash;
2908         } else if (join_hash->type == JSON_STRING) {
2909                 // turn it into a JSON_HASH by creating a wrapper
2910                 // around a copy of the original
2911                 const char* _tmp = jsonObjectGetString( join_hash );
2912                 freeable_hash = jsonNewObjectType(JSON_HASH);
2913                 jsonObjectSetKey(freeable_hash, _tmp, NULL);
2914                 working_hash = freeable_hash;
2915         } else {
2916                 osrfLogError(
2917                         OSRF_LOG_MARK,
2918                         "%s: JOIN failed; expected JSON object type not found",
2919                         MODULENAME
2920                 );
2921                 return NULL;
2922         }
2923
2924         growing_buffer* join_buf = buffer_init(128);
2925         const char* leftclass = left_info->class_name;
2926
2927         jsonObject* snode = NULL;
2928         jsonIterator* search_itr = jsonNewIterator( working_hash );
2929
2930         while ( (snode = jsonIteratorNext( search_itr )) ) {
2931                 const char* right_alias = search_itr->key;
2932                 const char* class =
2933                                 jsonObjectGetString( jsonObjectGetKeyConst( snode, "class" ) );
2934                 if( ! class )
2935                         class = right_alias;
2936
2937                 const ClassInfo* right_info = add_joined_class( right_alias, class );
2938                 if( !right_info ) {
2939                         osrfLogError(
2940                                 OSRF_LOG_MARK,
2941                                 "%s: JOIN failed.  Class \"%s\" not resolved in IDL",
2942                                 MODULENAME,
2943                                 search_itr->key
2944                         );
2945                         jsonIteratorFree( search_itr );
2946                         buffer_free( join_buf );
2947                         if( freeable_hash )
2948                                 jsonObjectFree( freeable_hash );
2949                         return NULL;
2950                 }
2951                 osrfHash* links    = right_info->links;
2952                 const char* table  = right_info->source_def;
2953
2954                 const char* fkey  = jsonObjectGetString( jsonObjectGetKeyConst( snode, "fkey" ) );
2955                 const char* field = jsonObjectGetString( jsonObjectGetKeyConst( snode, "field" ) );
2956
2957                 if (field && !fkey) {
2958                         // Look up the corresponding join column in the IDL.
2959                         // The link must be defined in the child table,
2960                         // and point to the right parent table.
2961                         osrfHash* idl_link = (osrfHash*) osrfHashGet( links, field );
2962                         const char* reltype = NULL;
2963                         const char* other_class = NULL;
2964                         reltype = osrfHashGet( idl_link, "reltype" );
2965                         if( reltype && strcmp( reltype, "has_many" ) )
2966                                 other_class = osrfHashGet( idl_link, "class" );
2967                         if( other_class && !strcmp( other_class, leftclass ) )
2968                                 fkey = osrfHashGet( idl_link, "key" );
2969                         if (!fkey) {
2970                                 osrfLogError(
2971                                         OSRF_LOG_MARK,
2972                                         "%s: JOIN failed.  No link defined from %s.%s to %s",
2973                                         MODULENAME,
2974                                         class,
2975                                         field,
2976                                         leftclass
2977                                 );
2978                                 buffer_free(join_buf);
2979                                 if(freeable_hash)
2980                                         jsonObjectFree(freeable_hash);
2981                                 jsonIteratorFree(search_itr);
2982                                 return NULL;
2983                         }
2984
2985                 } else if (!field && fkey) {
2986                         // Look up the corresponding join column in the IDL.
2987                         // The link must be defined in the child table,
2988                         // and point to the right parent table.
2989                         osrfHash* left_links = left_info->links;
2990                         osrfHash* idl_link = (osrfHash*) osrfHashGet( left_links, fkey );
2991                         const char* reltype = NULL;
2992                         const char* other_class = NULL;
2993                         reltype = osrfHashGet( idl_link, "reltype" );
2994                         if( reltype && strcmp( reltype, "has_many" ) )
2995                                 other_class = osrfHashGet( idl_link, "class" );
2996                         if( other_class && !strcmp( other_class, class ) )
2997                                 field = osrfHashGet( idl_link, "key" );
2998                         if (!field) {
2999                                 osrfLogError(
3000                                         OSRF_LOG_MARK,
3001                                         "%s: JOIN failed.  No link defined from %s.%s to %s",
3002                                         MODULENAME,
3003                                         leftclass,
3004                                         fkey,
3005                                         class
3006                                 );
3007                                 buffer_free(join_buf);
3008                                 if(freeable_hash)
3009                                         jsonObjectFree(freeable_hash);
3010                                 jsonIteratorFree(search_itr);
3011                                 return NULL;
3012                         }
3013
3014                 } else if (!field && !fkey) {
3015                         osrfHash* left_links = left_info->links;
3016
3017                         // For each link defined for the left class:
3018                         // see if the link references the joined class
3019                         osrfHashIterator* itr = osrfNewHashIterator( left_links );
3020                         osrfHash* curr_link = NULL;
3021                         while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
3022                                 const char* other_class = osrfHashGet( curr_link, "class" );
3023                                 if( other_class && !strcmp( other_class, class ) ) {
3024
3025                                         // In the IDL, the parent class doesn't always know then names of the child
3026                                         // columns that are pointing to it, so don't use that end of the link
3027                                         const char* reltype = osrfHashGet( curr_link, "reltype" );
3028                                         if( reltype && strcmp( reltype, "has_many" ) ) {
3029                                                 // Found a link between the classes
3030                                                 fkey = osrfHashIteratorKey( itr );
3031                                                 field = osrfHashGet( curr_link, "key" );
3032                                                 break;
3033                                         }
3034                                 }
3035                         }
3036                         osrfHashIteratorFree( itr );
3037
3038                         if (!field || !fkey) {
3039                                 // Do another such search, with the classes reversed
3040
3041                                 // For each link defined for the joined class:
3042                                 // see if the link references the left class
3043                                 osrfHashIterator* itr = osrfNewHashIterator( links );
3044                                 osrfHash* curr_link = NULL;
3045                                 while( (curr_link = osrfHashIteratorNext( itr ) ) ) {
3046                                         const char* other_class = osrfHashGet( curr_link, "class" );
3047                                         if( other_class && !strcmp( other_class, leftclass ) ) {
3048
3049                                                 // In the IDL, the parent class doesn't know then names of the child
3050                                                 // columns that are pointing to it, so don't use that end of the link
3051                                                 const char* reltype = osrfHashGet( curr_link, "reltype" );
3052                                                 if( reltype && strcmp( reltype, "has_many" ) ) {
3053                                                         // Found a link between the classes
3054                                                         field = osrfHashIteratorKey( itr );
3055                                                         fkey = osrfHashGet( curr_link, "key" );
3056                                                         break;
3057                                                 }
3058                                         }
3059                                 }
3060                                 osrfHashIteratorFree( itr );
3061                         }
3062
3063                         if (!field || !fkey) {
3064                                 osrfLogError(
3065                                         OSRF_LOG_MARK,
3066                                         "%s: JOIN failed.  No link defined between %s and %s",
3067                                         MODULENAME,
3068                                         leftclass,
3069                                         class
3070                                 );
3071                                 buffer_free(join_buf);
3072                                 if(freeable_hash)
3073                                         jsonObjectFree(freeable_hash);
3074                                 jsonIteratorFree(search_itr);
3075                                 return NULL;
3076                         }
3077                 }
3078
3079                 const char* type = jsonObjectGetString( jsonObjectGetKeyConst( snode, "type" ) );
3080                 if (type) {
3081                         if ( !strcasecmp(type,"left") ) {
3082                                 buffer_add(join_buf, " LEFT JOIN");
3083                         } else if ( !strcasecmp(type,"right") ) {
3084                                 buffer_add(join_buf, " RIGHT JOIN");
3085                         } else if ( !strcasecmp(type,"full") ) {
3086                                 buffer_add(join_buf, " FULL JOIN");
3087                         } else {
3088                                 buffer_add(join_buf, " INNER JOIN");
3089                         }
3090                 } else {
3091                         buffer_add(join_buf, " INNER JOIN");
3092                 }
3093
3094                 buffer_fadd(join_buf, " %s AS \"%s\" ON ( \"%s\".%s = \"%s\".%s",
3095                                         table, right_alias, right_alias, field, left_info->alias, fkey);
3096
3097                 // Add any other join conditions as specified by "filter"
3098                 const jsonObject* filter = jsonObjectGetKeyConst( snode, "filter" );
3099                 if (filter) {
3100                         const char* filter_op = jsonObjectGetString(
3101                                 jsonObjectGetKeyConst( snode, "filter_op" ) );
3102                         if ( filter_op && !strcasecmp("or",filter_op) ) {
3103                                 buffer_add( join_buf, " OR " );
3104                         } else {
3105                                 buffer_add( join_buf, " AND " );
3106                         }
3107
3108                         char* jpred = searchWHERE( filter, right_info, AND_OP_JOIN, NULL );
3109                         if( jpred ) {
3110                                 OSRF_BUFFER_ADD_CHAR( join_buf, ' ' );
3111                                 OSRF_BUFFER_ADD( join_buf, jpred );
3112                                 free(jpred);
3113                         } else {
3114                                 osrfLogError(
3115                                         OSRF_LOG_MARK,
3116                                         "%s: JOIN failed.  Invalid conditional expression.",
3117                                         MODULENAME
3118                                 );
3119                                 jsonIteratorFree( search_itr );
3120                                 buffer_free( join_buf );
3121                                 if( freeable_hash )
3122                                         jsonObjectFree( freeable_hash );
3123                                 return NULL;
3124                         }
3125                 }
3126
3127                 buffer_add(join_buf, " ) ");
3128
3129                 // Recursively add a nested join, if one is present
3130                 const jsonObject* join_filter = jsonObjectGetKeyConst( snode, "join" );
3131                 if (join_filter) {
3132                         char* jpred = searchJOIN( join_filter, right_info );
3133                         if( jpred ) {
3134                                 OSRF_BUFFER_ADD_CHAR( join_buf, ' ' );
3135                                 OSRF_BUFFER_ADD( join_buf, jpred );
3136                                 free(jpred);
3137                         } else {
3138                                 osrfLogError(  OSRF_LOG_MARK, "%s: Invalid nested join.", MODULENAME );
3139                                 jsonIteratorFree( search_itr );
3140                                 buffer_free( join_buf );
3141                                 if( freeable_hash )
3142                                         jsonObjectFree( freeable_hash );
3143                                 return NULL;
3144                         }
3145                 }
3146         }
3147
3148         if(freeable_hash)
3149                 jsonObjectFree(freeable_hash);
3150         jsonIteratorFree(search_itr);
3151
3152         return buffer_release(join_buf);
3153 }
3154
3155 /*
3156
3157 { +class : { -or|-and : { field : { op : value }, ... } ... }, ... }
3158 { +class : { -or|-and : [ { field : { op : value }, ... }, ...] ... }, ... }
3159 [ { +class : { -or|-and : [ { field : { op : value }, ... }, ...] ... }, ... }, ... ]
3160
3161 Generate code to express a set of conditions, as for a WHERE clause.  Parameters:
3162
3163 search_hash is the JSON expression of the conditions.
3164 meta is the class definition from the IDL, for the relevant table.
3165 opjoin_type indicates whether multiple conditions, if present, should be
3166         connected by AND or OR.
3167 osrfMethodContext is loaded with all sorts of stuff, but all we do with it here is
3168         to pass it to other functions -- and all they do with it is to use the session
3169         and request members to send error messages back to the client.
3170
3171 */
3172
3173 static char* searchWHERE ( const jsonObject* search_hash, const ClassInfo* class_info,
3174                 int opjoin_type, osrfMethodContext* ctx ) {
3175
3176         osrfLogDebug(
3177                 OSRF_LOG_MARK,
3178                 "%s: Entering searchWHERE; search_hash addr = %p, meta addr = %p, "
3179                 "opjoin_type = %d, ctx addr = %p",
3180                 MODULENAME,
3181                 search_hash,
3182                 class_info->class_def,
3183                 opjoin_type,
3184                 ctx
3185         );
3186
3187         growing_buffer* sql_buf = buffer_init(128);
3188
3189         jsonObject* node = NULL;
3190
3191         int first = 1;
3192         if ( search_hash->type == JSON_ARRAY ) {
3193                 osrfLogDebug( OSRF_LOG_MARK,
3194                         "%s: In WHERE clause, condition type is JSON_ARRAY", MODULENAME );
3195                 if( 0 == search_hash->size ) {
3196                         osrfLogError(
3197                                 OSRF_LOG_MARK,
3198                                 "%s: Invalid predicate structure: empty JSON array",
3199                                 MODULENAME
3200                         );
3201                         buffer_free( sql_buf );
3202                         return NULL;
3203                 }
3204
3205                 unsigned long i = 0;
3206                 while((node = jsonObjectGetIndex( search_hash, i++ ) )) {
3207                         if (first) {
3208                                 first = 0;
3209                         } else {
3210                                 if (opjoin_type == OR_OP_JOIN)
3211                                         buffer_add(sql_buf, " OR ");
3212                                 else
3213                                         buffer_add(sql_buf, " AND ");
3214                         }
3215
3216                         char* subpred = searchWHERE( node, class_info, opjoin_type, ctx );
3217                         if( ! subpred ) {
3218                                 buffer_free( sql_buf );
3219                                 return NULL;
3220                         }
3221
3222                         buffer_fadd(sql_buf, "( %s )", subpred);
3223                         free(subpred);
3224                 }
3225
3226         } else if ( search_hash->type == JSON_HASH ) {
3227                 osrfLogDebug( OSRF_LOG_MARK,
3228                         "%s: In WHERE clause, condition type is JSON_HASH", MODULENAME );
3229                 jsonIterator* search_itr = jsonNewIterator( search_hash );
3230                 if( !jsonIteratorHasNext( search_itr ) ) {
3231                         osrfLogError(
3232                                 OSRF_LOG_MARK,
3233                                 "%s: Invalid predicate structure: empty JSON object",
3234                                 MODULENAME
3235                         );
3236                         jsonIteratorFree( search_itr );
3237                         buffer_free( sql_buf );
3238                         return NULL;
3239                 }
3240
3241                 while ( (node = jsonIteratorNext( search_itr )) ) {
3242
3243                         if (first) {
3244                                 first = 0;
3245                         } else {
3246                                 if (opjoin_type == OR_OP_JOIN)
3247                                         buffer_add(sql_buf, " OR ");
3248                                 else
3249                                         buffer_add(sql_buf, " AND ");
3250                         }
3251
3252                         if ( '+' == search_itr->key[ 0 ] ) {
3253
3254                                 // This plus sign prefixes a class name or other table alias;
3255                                 // make sure the table alias is in scope
3256                                 ClassInfo* alias_info = search_all_alias( search_itr->key + 1 );
3257                                 if( ! alias_info ) {
3258                                         osrfLogError(
3259                                                          OSRF_LOG_MARK,
3260                                                         "%s: Invalid table alias \"%s\" in WHERE clause",
3261                                                         MODULENAME,
3262                                                         search_itr->key + 1
3263                                         );
3264                                         jsonIteratorFree( search_itr );
3265                                         buffer_free( sql_buf );
3266                                         return NULL;
3267                                 }
3268
3269                                 if ( node->type == JSON_STRING ) {
3270                                         // It's the name of a column; make sure it belongs to the class
3271                                         const char* fieldname = jsonObjectGetString( node );
3272                                         if( ! osrfHashGet( alias_info->fields, fieldname ) ) {
3273                                                 osrfLogError(
3274                                                         OSRF_LOG_MARK,
3275                                                         "%s: Invalid column name \"%s\" in WHERE clause "
3276                                                         "for table alias \"%s\"",
3277                                                         MODULENAME,
3278                                                         fieldname,
3279                                                         alias_info->alias
3280                                                 );
3281                                                 jsonIteratorFree( search_itr );
3282                                                 buffer_free( sql_buf );
3283                                                 return NULL;
3284                                         }
3285
3286                                         buffer_fadd(sql_buf, " \"%s\".%s ", alias_info->alias, fieldname );
3287                                 } else {
3288                                         // It's something more complicated
3289                                         char* subpred = searchWHERE( node, alias_info, AND_OP_JOIN, ctx );
3290                                         if( ! subpred ) {
3291                                                 jsonIteratorFree( search_itr );
3292                                                 buffer_free( sql_buf );
3293                                                 return NULL;
3294                                         }
3295
3296                                         buffer_fadd(sql_buf, "( %s )", subpred);
3297                                         free(subpred);
3298                                 }
3299                         } else if ( '-' == search_itr->key[ 0 ] ) {
3300                                 if ( !strcasecmp("-or",search_itr->key) ) {
3301                                         char* subpred = searchWHERE( node, class_info, OR_OP_JOIN, ctx );
3302                                         if( ! subpred ) {
3303                                                 jsonIteratorFree( search_itr );
3304                                                 buffer_free( sql_buf );
3305                                                 return NULL;
3306                                         }
3307
3308                                         buffer_fadd(sql_buf, "( %s )", subpred);
3309                                         free( subpred );
3310                                 } else if ( !strcasecmp("-and",search_itr->key) ) {
3311                                         char* subpred = searchWHERE( node, class_info, AND_OP_JOIN, ctx );
3312                                         if( ! subpred ) {
3313                                                 jsonIteratorFree( search_itr );
3314                                                 buffer_free( sql_buf );
3315                                                 return NULL;
3316                                         }
3317
3318                                         buffer_fadd(sql_buf, "( %s )", subpred);
3319                                         free( subpred );
3320                                 } else if ( !strcasecmp("-not",search_itr->key) ) {
3321                                         char* subpred = searchWHERE( node, class_info, AND_OP_JOIN, ctx );
3322                                         if( ! subpred ) {
3323                                                 jsonIteratorFree( search_itr );
3324                                                 buffer_free( sql_buf );
3325                                                 return NULL;
3326                                         }
3327
3328                                         buffer_fadd(sql_buf, " NOT ( %s )", subpred);
3329                                         free( subpred );
3330                                 } else if ( !strcasecmp("-exists",search_itr->key) ) {
3331                                         char* subpred = buildQuery( ctx, node, SUBSELECT );
3332                                         if( ! subpred ) {
3333                                                 jsonIteratorFree( search_itr );
3334                                                 buffer_free( sql_buf );
3335                                                 return NULL;
3336                                         }
3337
3338                                         buffer_fadd(sql_buf, "EXISTS ( %s )", subpred);
3339                                         free(subpred);
3340                                 } else if ( !strcasecmp("-not-exists",search_itr->key) ) {
3341                                         char* subpred = buildQuery( ctx, node, SUBSELECT );
3342                                         if( ! subpred ) {
3343                                                 jsonIteratorFree( search_itr );
3344                                                 buffer_free( sql_buf );
3345                                                 return NULL;
3346                                         }
3347
3348                                         buffer_fadd(sql_buf, "NOT EXISTS ( %s )", subpred);
3349                                         free(subpred);
3350                                 } else {     // Invalid "minus" operator
3351                                         osrfLogError(
3352                                                          OSRF_LOG_MARK,
3353                                                         "%s: Invalid operator \"%s\" in WHERE clause",
3354                                                         MODULENAME,
3355                                                         search_itr->key
3356                                         );
3357                                         jsonIteratorFree( search_itr );
3358                                         buffer_free( sql_buf );
3359                                         return NULL;
3360                                 }
3361
3362                         } else {
3363
3364                                 const char* class = class_info->class_name;
3365                                 osrfHash* fields = class_info->fields;
3366                                 osrfHash* field = osrfHashGet( fields, search_itr->key );
3367
3368                                 if (!field) {
3369                                         const char* table = class_info->source_def;
3370                                         osrfLogError(
3371                                                 OSRF_LOG_MARK,
3372                                                 "%s: Attempt to reference non-existent column \"%s\" on %s (%s)",
3373                                                 MODULENAME,
3374                                                 search_itr->key,
3375                                                 table ? table : "?",
3376                                                 class ? class : "?"
3377                                         );
3378                                         jsonIteratorFree(search_itr);
3379                                         buffer_free(sql_buf);
3380                                         return NULL;
3381                                 }
3382
3383                                 char* subpred = searchPredicate( class_info, field, node, ctx );
3384                                 if( ! subpred ) {
3385                                         buffer_free(sql_buf);
3386                                         jsonIteratorFree(search_itr);
3387                                         return NULL;
3388                                 }
3389
3390                                 buffer_add( sql_buf, subpred );
3391                                 free(subpred);
3392                         }
3393                 }
3394                 jsonIteratorFree(search_itr);
3395
3396         } else {
3397                 // ERROR ... only hash and array allowed at this level
3398                 char* predicate_string = jsonObjectToJSON( search_hash );
3399                 osrfLogError(
3400                         OSRF_LOG_MARK,
3401                         "%s: Invalid predicate structure: %s",
3402                         MODULENAME,
3403                         predicate_string
3404                 );
3405                 buffer_free(sql_buf);
3406                 free(predicate_string);
3407                 return NULL;
3408         }
3409
3410         return buffer_release(sql_buf);
3411 }
3412
3413 /* Build a JSON_ARRAY of field names for a given table alias
3414 */
3415 static jsonObject* defaultSelectList( const char* table_alias ) {
3416
3417         if( ! table_alias )
3418                 table_alias = "";
3419
3420         ClassInfo* class_info = search_all_alias( table_alias );
3421         if( ! class_info ) {
3422                 osrfLogError(
3423                         OSRF_LOG_MARK,
3424                         "%s: Can't build default SELECT clause for \"%s\"; no such table alias",
3425                         MODULENAME,
3426                         table_alias
3427                 );
3428                 return NULL;
3429         }
3430
3431         jsonObject* array = jsonNewObjectType( JSON_ARRAY );
3432         osrfHash* field_def = NULL;
3433         osrfHashIterator* field_itr = osrfNewHashIterator( class_info->fields );
3434         while( ( field_def = osrfHashIteratorNext( field_itr ) ) ) {
3435                 const char* field_name = osrfHashIteratorKey( field_itr );
3436                 if( ! str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
3437                         jsonObjectPush( array, jsonNewObject( field_name ) );
3438                 }
3439         }
3440         osrfHashIteratorFree( field_itr );
3441
3442         return array;
3443 }
3444
3445 // Translate a jsonObject into a UNION, INTERSECT, or EXCEPT query.
3446 // The jsonObject must be a JSON_HASH with an single entry for "union",
3447 // "intersect", or "except".  The data associated with this key must be an
3448 // array of hashes, each hash being a query.
3449 // Also allowed but currently ignored: entries for "order_by" and "alias".
3450 static char* doCombo( osrfMethodContext* ctx, jsonObject* combo, int flags ) {
3451         // Sanity check
3452         if( ! combo || combo->type != JSON_HASH )
3453                 return NULL;      // should be impossible; validated by caller
3454
3455         const jsonObject* query_array = NULL;   // array of subordinate queries
3456         const char* op = NULL;     // name of operator, e.g. UNION
3457         const char* alias = NULL;  // alias for the query (needed for ORDER BY)
3458         int op_count = 0;          // for detecting conflicting operators
3459         int excepting = 0;         // boolean
3460         int all = 0;               // boolean
3461         jsonObject* order_obj = NULL;
3462
3463         // Identify the elements in the hash
3464         jsonIterator* query_itr = jsonNewIterator( combo );
3465         jsonObject* curr_obj = NULL;
3466         while( (curr_obj = jsonIteratorNext( query_itr ) ) ) {
3467                 if( ! strcmp( "union", query_itr->key ) ) {
3468                         ++op_count;
3469                         op = " UNION ";
3470                         query_array = curr_obj;
3471                 } else if( ! strcmp( "intersect", query_itr->key ) ) {
3472                         ++op_count;
3473                         op = " INTERSECT ";
3474                         query_array = curr_obj;
3475                 } else if( ! strcmp( "except", query_itr->key ) ) {
3476                         ++op_count;
3477                         op = " EXCEPT ";
3478                         excepting = 1;
3479                         query_array = curr_obj;
3480                 } else if( ! strcmp( "order_by", query_itr->key ) ) {
3481                         osrfLogWarning(
3482                                 OSRF_LOG_MARK,
3483                                 "%s: ORDER BY not supported for UNION, INTERSECT, or EXCEPT",
3484                                 MODULENAME
3485                         );
3486                         order_obj = curr_obj;
3487                 } else if( ! strcmp( "alias", query_itr->key ) ) {
3488                         if( curr_obj->type != JSON_STRING ) {
3489                                 jsonIteratorFree( query_itr );
3490                                 return NULL;
3491                         }
3492                         alias = jsonObjectGetString( curr_obj );
3493                 } else if( ! strcmp( "all", query_itr->key ) ) {
3494                         if( obj_is_true( curr_obj ) )
3495                                 all = 1;
3496                 } else {
3497                         if( ctx )
3498                                 osrfAppSessionStatus(
3499                                         ctx->session,
3500                                         OSRF_STATUS_INTERNALSERVERERROR,
3501                                         "osrfMethodException",
3502                                         ctx->request,
3503                                         "Malformed query; unexpected entry in query object"
3504                                 );
3505                         osrfLogError(
3506                                 OSRF_LOG_MARK,
3507                                 "%s: Unexpected entry for \"%s\" in%squery",
3508                                 MODULENAME,
3509                                 query_itr->key,
3510                                 op
3511                         );
3512                         jsonIteratorFree( query_itr );
3513                         return NULL;
3514                 }
3515         }
3516         jsonIteratorFree( query_itr );
3517
3518         // More sanity checks
3519         if( ! query_array ) {
3520                 if( ctx )
3521                         osrfAppSessionStatus(
3522                                 ctx->session,
3523                                 OSRF_STATUS_INTERNALSERVERERROR,
3524                                 "osrfMethodException",
3525                                 ctx->request,
3526                                 "Expected UNION, INTERSECT, or EXCEPT operator not found"
3527                         );
3528                 osrfLogError(
3529                         OSRF_LOG_MARK,
3530                         "%s: Expected UNION, INTERSECT, or EXCEPT operator not found",
3531                         MODULENAME
3532                 );
3533                 return NULL;        // should be impossible...
3534         } else if( op_count > 1 ) {
3535                 if( ctx )
3536                                 osrfAppSessionStatus(
3537                                 ctx->session,
3538                                 OSRF_STATUS_INTERNALSERVERERROR,
3539                                 "osrfMethodException",
3540                                 ctx->request,
3541                                 "Found more than one of UNION, INTERSECT, and EXCEPT in same query"
3542                         );
3543                 osrfLogError(
3544                         OSRF_LOG_MARK,
3545                         "%s: Found more than one of UNION, INTERSECT, and EXCEPT in same query",
3546                         MODULENAME
3547                 );
3548                 return NULL;
3549         } if( query_array->type != JSON_ARRAY ) {
3550                 if( ctx )
3551                                 osrfAppSessionStatus(
3552                                 ctx->session,
3553                                 OSRF_STATUS_INTERNALSERVERERROR,
3554                                 "osrfMethodException",
3555                                 ctx->request,
3556                                 "Malformed query: expected array of queries under UNION, INTERSECT or EXCEPT"
3557                         );
3558                 osrfLogError(
3559                         OSRF_LOG_MARK,
3560                         "%s: Expected JSON_ARRAY of queries for%soperator; found %s",
3561                         MODULENAME,
3562                         op,
3563                         json_type( query_array->type )
3564                 );
3565                 return NULL;
3566         } if( query_array->size < 2 ) {
3567                 if( ctx )
3568                         osrfAppSessionStatus(
3569                                 ctx->session,
3570                                 OSRF_STATUS_INTERNALSERVERERROR,
3571                                 "osrfMethodException",
3572                                 ctx->request,
3573                                 "UNION, INTERSECT or EXCEPT requires multiple queries as operands"
3574                         );
3575                 osrfLogError(
3576                         OSRF_LOG_MARK,
3577                         "%s:%srequires multiple queries as operands",
3578                         MODULENAME,
3579                         op
3580                 );
3581                 return NULL;
3582         } else if( excepting && query_array->size > 2 ) {
3583                 if( ctx )
3584                         osrfAppSessionStatus(
3585                                 ctx->session,
3586                                 OSRF_STATUS_INTERNALSERVERERROR,
3587                                 "osrfMethodException",
3588                                 ctx->request,
3589                                 "EXCEPT operator has too many queries as operands"
3590                         );
3591                 osrfLogError(
3592                         OSRF_LOG_MARK,
3593                         "%s:EXCEPT operator has too many queries as operands",
3594                         MODULENAME
3595                 );
3596                 return NULL;
3597         } else if( order_obj && ! alias ) {
3598                 if( ctx )
3599                         osrfAppSessionStatus(
3600                                 ctx->session,
3601                                 OSRF_STATUS_INTERNALSERVERERROR,
3602                                 "osrfMethodException",
3603                                 ctx->request,
3604                                 "ORDER BY requires an alias for a UNION, INTERSECT, or EXCEPT"
3605                         );
3606                 osrfLogError(
3607                         OSRF_LOG_MARK,
3608                         "%s:ORDER BY requires an alias for a UNION, INTERSECT, or EXCEPT",
3609                         MODULENAME
3610                 );
3611                 return NULL;
3612         }
3613
3614         // So far so good.  Now build the SQL.
3615         growing_buffer* sql = buffer_init( 256 );
3616
3617         // If we nested inside another UNION, INTERSECT, or EXCEPT,
3618         // Add a layer of parentheses
3619         if( flags & SUBCOMBO )
3620                 OSRF_BUFFER_ADD( sql, "( " );
3621
3622         // Traverse the query array.  Each entry should be a hash.
3623         int first = 1;   // boolean
3624         int i = 0;
3625         jsonObject* query = NULL;
3626         while((query = jsonObjectGetIndex( query_array, i++ ) )) {
3627                 if( query->type != JSON_HASH ) {
3628                         if( ctx )
3629                                 osrfAppSessionStatus(
3630                                         ctx->session,
3631                                         OSRF_STATUS_INTERNALSERVERERROR,
3632                                         "osrfMethodException",
3633                                         ctx->request,
3634                                         "Malformed query under UNION, INTERSECT or EXCEPT"
3635                                 );
3636                         osrfLogError(
3637                                 OSRF_LOG_MARK,
3638                                 "%s: Malformed query under%s -- expected JSON_HASH, found %s",
3639                                 MODULENAME,
3640                                 op,
3641                                 json_type( query->type )
3642                         );
3643                         buffer_free( sql );
3644                         return NULL;
3645                 }
3646
3647                 if( first )
3648                         first = 0;
3649                 else {
3650                         OSRF_BUFFER_ADD( sql, op );
3651                         if( all )
3652                                 OSRF_BUFFER_ADD( sql, "ALL " );
3653                 }
3654
3655                 char* query_str = buildQuery( ctx, query, SUBSELECT | SUBCOMBO );
3656                 if( ! query_str ) {
3657                         osrfLogError(
3658                                 OSRF_LOG_MARK,
3659                                 "%s: Error building query under%s",
3660                                 MODULENAME,
3661                                 op
3662                         );
3663                         buffer_free( sql );
3664                         return NULL;
3665                 }
3666
3667                 OSRF_BUFFER_ADD( sql, query_str );
3668         }
3669
3670         if( flags & SUBCOMBO )
3671                 OSRF_BUFFER_ADD_CHAR( sql, ')' );
3672
3673         if ( !(flags & SUBSELECT) )
3674                 OSRF_BUFFER_ADD_CHAR( sql, ';' );
3675
3676         return buffer_release( sql );
3677 }
3678
3679 // Translate a jsonObject into a SELECT, UNION, INTERSECT, or EXCEPT query.
3680 // The jsonObject must be a JSON_HASH with an entry for "from", "union", "intersect",
3681 // or "except" to indicate the type of query.
3682 char* buildQuery( osrfMethodContext* ctx, jsonObject* query, int flags ) {
3683         // Sanity checks
3684         if( ! query ) {
3685                 if( ctx )
3686                         osrfAppSessionStatus(
3687                                 ctx->session,
3688                                 OSRF_STATUS_INTERNALSERVERERROR,
3689                                 "osrfMethodException",
3690                                 ctx->request,
3691                                 "Malformed query; no query object"
3692                         );
3693                 osrfLogError( OSRF_LOG_MARK, "%s: Null pointer to query object", MODULENAME );
3694                 return NULL;
3695         } else if( query->type != JSON_HASH ) {
3696                 if( ctx )
3697                         osrfAppSessionStatus(
3698                                 ctx->session,
3699                                 OSRF_STATUS_INTERNALSERVERERROR,
3700                                 "osrfMethodException",
3701                                 ctx->request,
3702                                 "Malformed query object"
3703                         );
3704                 osrfLogError(
3705                         OSRF_LOG_MARK,
3706                         "%s: Query object is %s instead of JSON_HASH",
3707                         MODULENAME,
3708                         json_type( query->type )
3709                 );
3710                 return NULL;
3711         }
3712
3713         // Determine what kind of query it purports to be, and dispatch accordingly.
3714         if( jsonObjectGetKey( query, "union" ) ||
3715                 jsonObjectGetKey( query, "intersect" ) ||
3716                 jsonObjectGetKey( query, "except" ) ) {
3717                 return doCombo( ctx, query, flags );
3718         } else {
3719                 // It is presumably a SELECT query
3720
3721                 // Push a node onto the stack for the current query.  Every level of
3722                 // subquery gets its own QueryFrame on the Stack.
3723                 push_query_frame();
3724
3725                 // Build an SQL SELECT statement
3726                 char* sql = SELECT(
3727                         ctx,
3728                         jsonObjectGetKey( query, "select" ),
3729                         jsonObjectGetKey( query, "from" ),
3730                         jsonObjectGetKey( query, "where" ),
3731                         jsonObjectGetKey( query, "having" ),
3732                         jsonObjectGetKey( query, "order_by" ),
3733                         jsonObjectGetKey( query, "limit" ),
3734                         jsonObjectGetKey( query, "offset" ),
3735                         flags
3736                 );
3737                 pop_query_frame();
3738                 return sql;
3739         }
3740 }
3741
3742 char* SELECT (
3743                 /* method context */ osrfMethodContext* ctx,
3744
3745                 /* SELECT   */ jsonObject* selhash,
3746                 /* FROM     */ jsonObject* join_hash,
3747                 /* WHERE    */ jsonObject* search_hash,
3748                 /* HAVING   */ jsonObject* having_hash,
3749                 /* ORDER BY */ jsonObject* order_hash,
3750                 /* LIMIT    */ jsonObject* limit,
3751                 /* OFFSET   */ jsonObject* offset,
3752                 /* flags    */ int flags
3753 ) {
3754         const char* locale = osrf_message_get_last_locale();
3755
3756         // general tmp objects
3757         const jsonObject* tmp_const;
3758         jsonObject* selclass = NULL;
3759         jsonObject* snode = NULL;
3760         jsonObject* onode = NULL;
3761
3762         char* string = NULL;
3763         int from_function = 0;
3764         int first = 1;
3765         int gfirst = 1;
3766         //int hfirst = 1;
3767
3768         osrfLogDebug(OSRF_LOG_MARK, "cstore SELECT locale: %s", locale ? locale : "(none)" );
3769
3770         // punt if there's no FROM clause
3771         if (!join_hash || ( join_hash->type == JSON_HASH && !join_hash->size )) {
3772                 osrfLogError(
3773                         OSRF_LOG_MARK,
3774                         "%s: FROM clause is missing or empty",
3775                         MODULENAME
3776                 );
3777                 if( ctx )
3778                         osrfAppSessionStatus(
3779                                 ctx->session,
3780                                 OSRF_STATUS_INTERNALSERVERERROR,
3781                                 "osrfMethodException",
3782                                 ctx->request,
3783                                 "FROM clause is missing or empty in JSON query"
3784                         );
3785                 return NULL;
3786         }
3787
3788         // the core search class
3789         const char* core_class = NULL;
3790
3791         // get the core class -- the only key of the top level FROM clause, or a string
3792         if (join_hash->type == JSON_HASH) {
3793                 jsonIterator* tmp_itr = jsonNewIterator( join_hash );
3794                 snode = jsonIteratorNext( tmp_itr );
3795
3796                 // Populate the current QueryFrame with information
3797                 // about the core class
3798                 if( add_query_core( NULL, tmp_itr->key ) ) {
3799                         if( ctx )
3800                                 osrfAppSessionStatus(
3801                                         ctx->session,
3802                                         OSRF_STATUS_INTERNALSERVERERROR,
3803                                         "osrfMethodException",
3804                                         ctx->request,
3805                                         "Unable to look up core class"
3806                                 );
3807                         return NULL;
3808                 }
3809                 core_class = curr_query->core.class_name;
3810                 join_hash = snode;
3811
3812                 jsonObject* extra = jsonIteratorNext( tmp_itr );
3813
3814                 jsonIteratorFree( tmp_itr );
3815                 snode = NULL;
3816
3817                 // There shouldn't be more than one entry in join_hash
3818                 if( extra ) {
3819                         osrfLogError(
3820                                 OSRF_LOG_MARK,
3821                                 "%s: Malformed FROM clause: extra entry in JSON_HASH",
3822                                 MODULENAME
3823                         );
3824                         if( ctx )
3825                                 osrfAppSessionStatus(
3826                                         ctx->session,
3827                                         OSRF_STATUS_INTERNALSERVERERROR,
3828                                         "osrfMethodException",
3829                                         ctx->request,
3830                                         "Malformed FROM clause in JSON query"
3831                                 );
3832                         return NULL;    // Malformed join_hash; extra entry
3833                 }
3834         } else if (join_hash->type == JSON_ARRAY) {
3835                 // We're selecting from a function, not from a table
3836                 from_function = 1;
3837                 core_class = jsonObjectGetString( jsonObjectGetIndex(join_hash, 0) );
3838                 selhash = NULL;
3839
3840         } else if (join_hash->type == JSON_STRING) {
3841                 // Populate the current QueryFrame with information
3842                 // about the core class
3843                 core_class = jsonObjectGetString( join_hash );
3844                 join_hash = NULL;
3845                 if( add_query_core( NULL, core_class ) ) {
3846                         if( ctx )
3847                                 osrfAppSessionStatus(
3848                                         ctx->session,
3849                                         OSRF_STATUS_INTERNALSERVERERROR,
3850                                         "osrfMethodException",
3851                                         ctx->request,
3852                                         "Unable to look up core class"
3853                                 );
3854                         return NULL;
3855                 }
3856         }
3857         else {
3858                 osrfLogError(
3859                         OSRF_LOG_MARK,
3860                         "%s: FROM clause is unexpected JSON type: %s",
3861                         MODULENAME,
3862                         json_type( join_hash->type )
3863                 );
3864                 if( ctx )
3865                         osrfAppSessionStatus(
3866                                 ctx->session,
3867                                 OSRF_STATUS_INTERNALSERVERERROR,
3868                                 "osrfMethodException",
3869                                 ctx->request,
3870                                 "Ill-formed FROM clause in JSON query"
3871                         );
3872                 return NULL;
3873         }
3874
3875         // Build the join clause, if any, while filling out the list
3876         // of joined classes in the current QueryFrame.
3877         char* join_clause = NULL;
3878         if( join_hash && ! from_function ) {
3879
3880                 join_clause = searchJOIN( join_hash, &curr_query->core );
3881                 if( ! join_clause ) {
3882                         if (ctx)
3883                                 osrfAppSessionStatus(
3884                                         ctx->session,
3885                                         OSRF_STATUS_INTERNALSERVERERROR,
3886                                         "osrfMethodException",
3887                                         ctx->request,
3888                                         "Unable to construct JOIN clause(s)"
3889                                 );
3890                         return NULL;
3891                 }
3892         }
3893
3894         // For in case we don't get a select list
3895         jsonObject* defaultselhash = NULL;
3896
3897         // if there is no select list, build a default select list ...
3898         if (!selhash && !from_function) {
3899                 jsonObject* default_list = defaultSelectList( core_class );
3900                 if( ! default_list ) {
3901                         if (ctx) {
3902                                 osrfAppSessionStatus(
3903                                         ctx->session,
3904                                         OSRF_STATUS_INTERNALSERVERERROR,
3905                                         "osrfMethodException",
3906                                         ctx->request,
3907                                         "Unable to build default SELECT clause in JSON query"
3908                                 );
3909                                 free( join_clause );
3910                                 return NULL;
3911                         }
3912                 }
3913
3914                 selhash = defaultselhash = jsonNewObjectType(JSON_HASH);
3915                 jsonObjectSetKey( selhash, core_class, default_list );
3916         }
3917
3918         // The SELECT clause can be encoded only by a hash
3919         if( !from_function && selhash->type != JSON_HASH ) {
3920                 osrfLogError(
3921                         OSRF_LOG_MARK,
3922                         "%s: Expected JSON_HASH for SELECT clause; found %s",
3923                         MODULENAME,
3924                         json_type( selhash->type )
3925                 );
3926
3927                 if (ctx)
3928                         osrfAppSessionStatus(
3929                                 ctx->session,
3930                                 OSRF_STATUS_INTERNALSERVERERROR,
3931                                 "osrfMethodException",
3932                                 ctx->request,
3933                                 "Malformed SELECT clause in JSON query"
3934                         );
3935                 free( join_clause );
3936                 return NULL;
3937         }
3938
3939         // If you see a null or wild card specifier for the core class, or an
3940         // empty array, replace it with a default SELECT list
3941         tmp_const = jsonObjectGetKeyConst( selhash, core_class );
3942         if ( tmp_const ) {
3943                 int default_needed = 0;   // boolean
3944                 if( JSON_STRING == tmp_const->type
3945                         && !strcmp( "*", jsonObjectGetString( tmp_const ) ))
3946                                 default_needed = 1;
3947                 else if( JSON_NULL == tmp_const->type )
3948                         default_needed = 1;
3949
3950                 if( default_needed ) {
3951                         // Build a default SELECT list
3952                         jsonObject* default_list = defaultSelectList( core_class );
3953                         if( ! default_list ) {
3954                                 if (ctx) {
3955                                         osrfAppSessionStatus(
3956                                                 ctx->session,
3957                                                 OSRF_STATUS_INTERNALSERVERERROR,
3958                                                 "osrfMethodException",
3959                                                 ctx->request,
3960                                                 "Can't build default SELECT clause in JSON query"
3961                                         );
3962                                         free( join_clause );
3963                                         return NULL;
3964                                 }
3965                         }
3966
3967                         jsonObjectSetKey( selhash, core_class, default_list );
3968                 }
3969         }
3970
3971         // temp buffers for the SELECT list and GROUP BY clause
3972         growing_buffer* select_buf = buffer_init(128);
3973         growing_buffer* group_buf = buffer_init(128);
3974
3975         int aggregate_found = 0;     // boolean
3976
3977         // Build a select list
3978         if(from_function)   // From a function we select everything
3979                 OSRF_BUFFER_ADD_CHAR( select_buf, '*' );
3980         else {
3981
3982                 // Build the SELECT list as SQL
3983             int sel_pos = 1;
3984             first = 1;
3985             gfirst = 1;
3986             jsonIterator* selclass_itr = jsonNewIterator( selhash );
3987             while ( (selclass = jsonIteratorNext( selclass_itr )) ) {    // For each class
3988
3989                         const char* cname = selclass_itr->key;
3990
3991                         // Make sure the target relation is in the FROM clause.
3992
3993                         // At this point join_hash is a step down from the join_hash we
3994                         // received as a parameter.  If the original was a JSON_STRING,
3995                         // then json_hash is now NULL.  If the original was a JSON_HASH,
3996                         // then json_hash is now the first (and only) entry in it,
3997                         // denoting the core class.  We've already excluded the
3998                         // possibility that the original was a JSON_ARRAY, because in
3999                         // that case from_function would be non-NULL, and we wouldn't
4000                         // be here.
4001
4002                         // If the current table alias isn't in scope, bail out
4003                         ClassInfo* class_info = search_alias( cname );
4004                         if( ! class_info ) {
4005                                 osrfLogError(
4006                                         OSRF_LOG_MARK,
4007                                         "%s: SELECT clause references class not in FROM clause: \"%s\"",
4008                                         MODULENAME,
4009                                         cname
4010                                 );
4011                                 if( ctx )
4012                                         osrfAppSessionStatus(
4013                                                 ctx->session,
4014                                                 OSRF_STATUS_INTERNALSERVERERROR,
4015                                                 "osrfMethodException",
4016                                                 ctx->request,
4017                                                 "Selected class not in FROM clause in JSON query"
4018                                         );
4019                                 jsonIteratorFree( selclass_itr );
4020                                 buffer_free( select_buf );
4021                                 buffer_free( group_buf );
4022                                 if( defaultselhash )
4023                                         jsonObjectFree( defaultselhash );
4024                                 free( join_clause );
4025                                 return NULL;
4026                         }
4027
4028                         if( selclass->type != JSON_ARRAY ) {
4029                                 osrfLogError(
4030                                         OSRF_LOG_MARK,
4031                                         "%s: Malformed SELECT list for class \"%s\"; not an array",
4032                                         MODULENAME,
4033                                         cname
4034                                 );
4035                                 if( ctx )
4036                                         osrfAppSessionStatus(
4037                                                 ctx->session,
4038                                                 OSRF_STATUS_INTERNALSERVERERROR,
4039                                                 "osrfMethodException",
4040                                                 ctx->request,
4041                                                 "Selected class not in FROM clause in JSON query"
4042                                         );
4043
4044                                 jsonIteratorFree( selclass_itr );
4045                                 buffer_free( select_buf );
4046                                 buffer_free( group_buf );
4047                                 if( defaultselhash )
4048                                         jsonObjectFree( defaultselhash );
4049                                 free( join_clause );
4050                                 return NULL;
4051                         }
4052
4053                         // Look up some attributes of the current class
4054                         osrfHash* idlClass = class_info->class_def;
4055                         osrfHash* class_field_set = class_info->fields;
4056                         const char* class_pkey = osrfHashGet( idlClass, "primarykey" );
4057                         const char* class_tname = osrfHashGet( idlClass, "tablename" );
4058
4059                         if( 0 == selclass->size ) {
4060                                 osrfLogWarning(
4061                                         OSRF_LOG_MARK,
4062                                         "%s: No columns selected from \"%s\"",
4063                                         MODULENAME,
4064                                         cname
4065                                 );
4066                         }
4067
4068                         // stitch together the column list for the current table alias...
4069                         unsigned long field_idx = 0;
4070                         jsonObject* selfield = NULL;
4071                         while((selfield = jsonObjectGetIndex( selclass, field_idx++ ) )) {
4072
4073                                 // If we need a separator comma, add one
4074                                 if (first) {
4075                                         first = 0;
4076                                 } else {
4077                                         OSRF_BUFFER_ADD_CHAR( select_buf, ',' );
4078                                 }
4079
4080                                 // if the field specification is a string, add it to the list
4081                                 if (selfield->type == JSON_STRING) {
4082
4083                                         // Look up the field in the IDL
4084                                         const char* col_name = jsonObjectGetString( selfield );
4085                                         osrfHash* field_def = osrfHashGet( class_field_set, col_name );
4086                                         if ( !field_def ) {
4087                                                 // No such field in current class
4088                                                 osrfLogError(
4089                                                         OSRF_LOG_MARK,
4090                                                         "%s: Selected column \"%s\" not defined in IDL for class \"%s\"",
4091                                                         MODULENAME,
4092                                                         col_name,
4093                                                         cname
4094                                                 );
4095                                                 if( ctx )
4096                                                         osrfAppSessionStatus(
4097                                                                 ctx->session,
4098                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4099                                                                 "osrfMethodException",
4100                                                                 ctx->request,
4101                                                                 "Selected column not defined in JSON query"
4102                                                         );
4103                                                 jsonIteratorFree( selclass_itr );
4104                                                 buffer_free( select_buf );
4105                                                 buffer_free( group_buf );
4106                                                 if( defaultselhash )
4107                                                         jsonObjectFree( defaultselhash );
4108                                                 free( join_clause );
4109                                                 return NULL;
4110                                         } else if ( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4111                                                 // Virtual field not allowed
4112                                                 osrfLogError(
4113                                                         OSRF_LOG_MARK,
4114                                                         "%s: Selected column \"%s\" for class \"%s\" is virtual",
4115                                                         MODULENAME,
4116                                                         col_name,
4117                                                         cname
4118                                                 );
4119                                                 if( ctx )
4120                                                         osrfAppSessionStatus(
4121                                                                 ctx->session,
4122                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4123                                                                 "osrfMethodException",
4124                                                                 ctx->request,
4125                                                                 "Selected column may not be virtual in JSON query"
4126                                                         );
4127                                                 jsonIteratorFree( selclass_itr );
4128                                                 buffer_free( select_buf );
4129                                                 buffer_free( group_buf );
4130                                                 if( defaultselhash )
4131                                                         jsonObjectFree( defaultselhash );
4132                                                 free( join_clause );
4133                                                 return NULL;
4134                                         }
4135
4136                                         if (locale) {
4137                                                 const char* i18n;
4138                                                 if (flags & DISABLE_I18N)
4139                                                         i18n = NULL;
4140                                                 else
4141                                                         i18n = osrfHashGet(field_def, "i18n");
4142
4143                                                 if( str_is_true( i18n ) ) {
4144                                                         buffer_fadd( select_buf, " oils_i18n_xlate('%s', '%s', '%s', "
4145                                                                 "'%s', \"%s\".%s::TEXT, '%s') AS \"%s\"",
4146                                                                 class_tname, cname, col_name, class_pkey,
4147                                                                 cname, class_pkey, locale, col_name );
4148                                                 } else {
4149                                                         buffer_fadd(select_buf, " \"%s\".%s AS \"%s\"",
4150                                                                 cname, col_name, col_name );
4151                                                 }
4152                                         } else {
4153                                                 buffer_fadd(select_buf, " \"%s\".%s AS \"%s\"",
4154                                                                 cname, col_name, col_name );
4155                                         }
4156
4157                                 // ... but it could be an object, in which case we check for a Field Transform
4158                                 } else if (selfield->type == JSON_HASH) {
4159
4160                                         const char* col_name = jsonObjectGetString(
4161                                                         jsonObjectGetKeyConst( selfield, "column" ) );
4162
4163                                         // Get the field definition from the IDL
4164                                         osrfHash* field_def = osrfHashGet( class_field_set, col_name );
4165                                         if ( !field_def ) {
4166                                                 // No such field in current class
4167                                                 osrfLogError(
4168                                                         OSRF_LOG_MARK,
4169                                                         "%s: Selected column \"%s\" is not defined in IDL for class \"%s\"",
4170                                                         MODULENAME,
4171                                                         col_name,
4172                                                         cname
4173                                                 );
4174                                                 if( ctx )
4175                                                         osrfAppSessionStatus(
4176                                                                 ctx->session,
4177                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4178                                                                 "osrfMethodException",
4179                                                                 ctx->request,
4180                                                                 "Selected column is not defined in JSON query"
4181                                                         );
4182                                                 jsonIteratorFree( selclass_itr );
4183                                                 buffer_free( select_buf );
4184                                                 buffer_free( group_buf );
4185                                                 if( defaultselhash )
4186                                                         jsonObjectFree( defaultselhash );
4187                                                 free( join_clause );
4188                                                 return NULL;
4189                                         } else if ( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4190                                                 // No such field in current class
4191                                                 osrfLogError(
4192                                                         OSRF_LOG_MARK,
4193                                                         "%s: Selected column \"%s\" is virtual for class \"%s\"",
4194                                                         MODULENAME,
4195                                                         col_name,
4196                                                         cname
4197                                                 );
4198                                                 if( ctx )
4199                                                         osrfAppSessionStatus(
4200                                                                 ctx->session,
4201                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4202                                                                 "osrfMethodException",
4203                                                                 ctx->request,
4204                                                                 "Selected column is virtual in JSON query"
4205                                                         );
4206                                                 jsonIteratorFree( selclass_itr );
4207                                                 buffer_free( select_buf );
4208                                                 buffer_free( group_buf );
4209                                                 if( defaultselhash )
4210                                                         jsonObjectFree( defaultselhash );
4211                                                 free( join_clause );
4212                                                 return NULL;
4213                                         }
4214
4215                                         // Decide what to use as a column alias
4216                                         const char* _alias;
4217                                         if ((tmp_const = jsonObjectGetKeyConst( selfield, "alias" ))) {
4218                                                 _alias = jsonObjectGetString( tmp_const );
4219                                         } else {         // Use field name as the alias
4220                                                 _alias = col_name;
4221                                         }
4222
4223                                         if (jsonObjectGetKeyConst( selfield, "transform" )) {
4224                                                 char* transform_str = searchFieldTransform(
4225                                                         class_info->alias, field_def, selfield );
4226                                                 if( transform_str ) {
4227                                                         buffer_fadd(select_buf, " %s AS \"%s\"", transform_str, _alias);
4228                                                         free(transform_str);
4229                                                 } else {
4230                                                         if( ctx )
4231                                                                 osrfAppSessionStatus(
4232                                                                         ctx->session,
4233                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4234                                                                         "osrfMethodException",
4235                                                                         ctx->request,
4236                                                                         "Unable to generate transform function in JSON query"
4237                                                                 );
4238                                                         jsonIteratorFree( selclass_itr );
4239                                                         buffer_free( select_buf );
4240                                                         buffer_free( group_buf );
4241                                                         if( defaultselhash )
4242                                                                 jsonObjectFree( defaultselhash );
4243                                                         free( join_clause );
4244                                                         return NULL;
4245                                                 }
4246                                         } else {
4247
4248                                                 if (locale) {
4249                                                         const char* i18n;
4250                                                         if (flags & DISABLE_I18N)
4251                                                                 i18n = NULL;
4252                                                         else
4253                                                                 i18n = osrfHashGet(field_def, "i18n");
4254
4255                                                         if( str_is_true( i18n ) ) {
4256                                                                 buffer_fadd( select_buf,
4257                                                                         " oils_i18n_xlate('%s', '%s', '%s', '%s', "
4258                                                                         "\"%s\".%s::TEXT, '%s') AS \"%s\"",
4259                                                                         class_tname, cname, col_name, class_pkey, cname,
4260                                                                         class_pkey, locale, _alias);
4261                                                         } else {
4262                                                                 buffer_fadd( select_buf, " \"%s\".%s AS \"%s\"",
4263                                                                         cname, col_name, _alias );
4264                                                         }
4265                                                 } else {
4266                                                         buffer_fadd( select_buf, " \"%s\".%s AS \"%s\"",
4267                                                                 cname, col_name, _alias);
4268                                                 }
4269                                         }
4270                                 }
4271                                 else {
4272                                         osrfLogError(
4273                                                 OSRF_LOG_MARK,
4274                                                 "%s: Selected item is unexpected JSON type: %s",
4275                                                 MODULENAME,
4276                                                 json_type( selfield->type )
4277                                         );
4278                                         if( ctx )
4279                                                 osrfAppSessionStatus(
4280                                                         ctx->session,
4281                                                         OSRF_STATUS_INTERNALSERVERERROR,
4282                                                         "osrfMethodException",
4283                                                         ctx->request,
4284                                                         "Ill-formed SELECT item in JSON query"
4285                                                 );
4286                                         jsonIteratorFree( selclass_itr );
4287                                         buffer_free( select_buf );
4288                                         buffer_free( group_buf );
4289                                         if( defaultselhash )
4290                                                 jsonObjectFree( defaultselhash );
4291                                         free( join_clause );
4292                                         return NULL;
4293                                 }
4294
4295                                 const jsonObject* agg_obj = jsonObjectGetKey( selfield, "aggregate" );
4296                                 if( obj_is_true( agg_obj ) )
4297                                         aggregate_found = 1;
4298                                 else {
4299                                         // Append a comma (except for the first one)
4300                                         // and add the column to a GROUP BY clause
4301                                         if (gfirst)
4302                                                 gfirst = 0;
4303                                         else
4304                                                 OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4305
4306                                         buffer_fadd(group_buf, " %d", sel_pos);
4307                                 }
4308
4309 #if 0
4310                             if (is_agg->size || (flags & SELECT_DISTINCT)) {
4311
4312                                         const jsonObject* aggregate_obj = jsonObjectGetKey( selfield, "aggregate" );
4313                                     if ( ! obj_is_true( aggregate_obj ) ) {
4314                                             if (gfirst) {
4315                                                     gfirst = 0;
4316                                             } else {
4317                                                         OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4318                                             }
4319
4320                                             buffer_fadd(group_buf, " %d", sel_pos);
4321
4322                                         /*
4323                                     } else if (is_agg = jsonObjectGetKey( selfield, "having" )) {
4324                                             if (gfirst) {
4325                                                     gfirst = 0;
4326                                             } else {
4327                                                         OSRF_BUFFER_ADD_CHAR( group_buf, ',' );
4328                                             }
4329
4330                                             _column = searchFieldTransform(class_info->alias, field, selfield);
4331                                                 OSRF_BUFFER_ADD_CHAR(group_buf, ' ');
4332                                                 OSRF_BUFFER_ADD(group_buf, _column);
4333                                             _column = searchFieldTransform(class_info->alias, field, selfield);
4334                                         */
4335                                     }
4336                             }
4337 #endif
4338
4339                                 sel_pos++;
4340                         } // end while -- iterating across SELECT columns
4341
4342                 } // end while -- iterating across classes
4343
4344                 jsonIteratorFree(selclass_itr);
4345         }
4346
4347
4348         char* col_list = buffer_release(select_buf);
4349
4350         // Make sure the SELECT list isn't empty.  This can happen, for example,
4351         // if we try to build a default SELECT clause from a non-core table.
4352
4353         if( ! *col_list ) {
4354                 osrfLogError(OSRF_LOG_MARK, "%s: SELECT clause is empty", MODULENAME );
4355                 if (ctx)
4356                         osrfAppSessionStatus(
4357                                 ctx->session,
4358                                 OSRF_STATUS_INTERNALSERVERERROR,
4359                                 "osrfMethodException",
4360                                 ctx->request,
4361                                 "SELECT list is empty"
4362                 );
4363                 free( col_list );
4364                 buffer_free( group_buf );
4365                 if( defaultselhash )
4366                         jsonObjectFree( defaultselhash );
4367                 free( join_clause );
4368                 return NULL;
4369         }
4370
4371         char* table = NULL;
4372         if (from_function) table = searchValueTransform(join_hash);
4373         else table = strdup( curr_query->core.source_def );
4374
4375         if( !table ) {
4376                 if (ctx)
4377                         osrfAppSessionStatus(
4378                                 ctx->session,
4379                                 OSRF_STATUS_INTERNALSERVERERROR,
4380                                 "osrfMethodException",
4381                                 ctx->request,
4382                                 "Unable to identify table for core class"
4383                         );
4384                 free( col_list );
4385                 buffer_free( group_buf );
4386                 if( defaultselhash )
4387                         jsonObjectFree( defaultselhash );
4388                 free( join_clause );
4389                 return NULL;
4390         }
4391
4392         // Put it all together
4393         growing_buffer* sql_buf = buffer_init(128);
4394         buffer_fadd(sql_buf, "SELECT %s FROM %s AS \"%s\" ", col_list, table, core_class );
4395         free(col_list);
4396         free(table);
4397
4398         // Append the join clause, if any
4399         if( join_clause ) {
4400                 buffer_add(sql_buf, join_clause);
4401                 free(join_clause);
4402         }
4403
4404         char* order_by_list = NULL;
4405         char* having_buf = NULL;
4406
4407         if (!from_function) {
4408
4409                 // Build a WHERE clause, if there is one
4410                 if ( search_hash ) {
4411                         buffer_add(sql_buf, " WHERE ");
4412
4413                         // and it's on the WHERE clause
4414                         char* pred = searchWHERE( search_hash, &curr_query->core, AND_OP_JOIN, ctx );
4415                         if ( ! pred ) {
4416                                 if (ctx) {
4417                                         osrfAppSessionStatus(
4418                                                 ctx->session,
4419                                                 OSRF_STATUS_INTERNALSERVERERROR,
4420                                                 "osrfMethodException",
4421                                                 ctx->request,
4422                                                 "Severe query error in WHERE predicate -- see error log for more details"
4423                                         );
4424                                 }
4425                                 buffer_free(group_buf);
4426                                 buffer_free(sql_buf);
4427                                 if (defaultselhash)
4428                                         jsonObjectFree(defaultselhash);
4429                                 return NULL;
4430                         }
4431
4432                         buffer_add(sql_buf, pred);
4433                         free(pred);
4434                 }
4435
4436                 // Build a HAVING clause, if there is one
4437                 if ( having_hash ) {
4438
4439                         // and it's on the the WHERE clause
4440                         having_buf = searchWHERE( having_hash, &curr_query->core, AND_OP_JOIN, ctx );
4441
4442                         if( ! having_buf ) {
4443                                 if (ctx) {
4444                                                 osrfAppSessionStatus(
4445                                                 ctx->session,
4446                                                 OSRF_STATUS_INTERNALSERVERERROR,
4447                                                 "osrfMethodException",
4448                                                 ctx->request,
4449                                                 "Severe query error in HAVING predicate -- see error log for more details"
4450                                         );
4451                                 }
4452                                 buffer_free(group_buf);
4453                                 buffer_free(sql_buf);
4454                                 if (defaultselhash)
4455                                         jsonObjectFree(defaultselhash);
4456                                 return NULL;
4457                         }
4458                 }
4459
4460                 growing_buffer* order_buf = NULL;  // to collect ORDER BY list
4461
4462                 // Build an ORDER BY clause, if there is one
4463                 if( NULL == order_hash )
4464                         ;  // No ORDER BY? do nothing
4465                 else if( JSON_ARRAY == order_hash->type ) {
4466                         // Array of field specifications, each specification being a
4467                         // hash to define the class, field, and other details
4468                         int order_idx = 0;
4469                         jsonObject* order_spec;
4470                         while( (order_spec = jsonObjectGetIndex( order_hash, order_idx++ ) ) ) {
4471
4472                                 if( JSON_HASH != order_spec->type ) {
4473                                         osrfLogError(OSRF_LOG_MARK,
4474                                                  "%s: Malformed field specification in ORDER BY clause; expected JSON_HASH, found %s",
4475                                                 MODULENAME, json_type( order_spec->type ) );
4476                                         if( ctx )
4477                                                 osrfAppSessionStatus(
4478                                                          ctx->session,
4479                                                         OSRF_STATUS_INTERNALSERVERERROR,
4480                                                         "osrfMethodException",
4481                                                         ctx->request,
4482                                                         "Malformed ORDER BY clause -- see error log for more details"
4483                                                 );
4484                                         buffer_free( order_buf );
4485                                         free(having_buf);
4486                                         buffer_free(group_buf);
4487                                         buffer_free(sql_buf);
4488                                         if (defaultselhash)
4489                                                 jsonObjectFree(defaultselhash);
4490                                         return NULL;
4491                                 }
4492
4493                                 const char* class_alias =
4494                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "class" ) );
4495                                 const char* field =
4496                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "field" ) );
4497
4498                                 if ( order_buf )
4499                                         OSRF_BUFFER_ADD(order_buf, ", ");
4500                                 else
4501                                         order_buf = buffer_init(128);
4502
4503                                 if( !field || !class_alias ) {
4504                                         osrfLogError( OSRF_LOG_MARK,
4505                                                 "%s: Missing class or field name in field specification "
4506                                                 "of ORDER BY clause",
4507                                                 MODULENAME );
4508                                         if( ctx )
4509                                                 osrfAppSessionStatus(
4510                                                         ctx->session,
4511                                                         OSRF_STATUS_INTERNALSERVERERROR,
4512                                                         "osrfMethodException",
4513                                                         ctx->request,
4514                                                         "Malformed ORDER BY clause -- see error log for more details"
4515                                                 );
4516                                         buffer_free( order_buf );
4517                                         free(having_buf);
4518                                         buffer_free(group_buf);
4519                                         buffer_free(sql_buf);
4520                                         if (defaultselhash)
4521                                                 jsonObjectFree(defaultselhash);
4522                                         return NULL;
4523                                 }
4524
4525                                 ClassInfo* order_class_info = search_alias( class_alias );
4526                                 if( ! order_class_info ) {
4527                                         osrfLogError(OSRF_LOG_MARK, "%s: ORDER BY clause references class \"%s\" "
4528                                                         "not in FROM clause", MODULENAME, class_alias );
4529                                         if( ctx )
4530                                                 osrfAppSessionStatus(
4531                                                         ctx->session,
4532                                                         OSRF_STATUS_INTERNALSERVERERROR,
4533                                                         "osrfMethodException",
4534                                                         ctx->request,
4535                                                         "Invalid class referenced in ORDER BY clause -- "
4536                                                         "see error log for more details"
4537                                                 );
4538                                         free(having_buf);
4539                                         buffer_free(group_buf);
4540                                         buffer_free(sql_buf);
4541                                         if (defaultselhash)
4542                                                 jsonObjectFree(defaultselhash);
4543                                         return NULL;
4544                                 }
4545
4546                                 osrfHash* field_def = osrfHashGet( order_class_info->fields, field );
4547                                 if( !field_def ) {
4548                                         osrfLogError( OSRF_LOG_MARK,
4549                                                 "%s: Invalid field \"%s\".%s referenced in ORDER BY clause",
4550                                                 MODULENAME, class_alias, field );
4551                                         if( ctx )
4552                                                 osrfAppSessionStatus(
4553                                                         ctx->session,
4554                                                         OSRF_STATUS_INTERNALSERVERERROR,
4555                                                         "osrfMethodException",
4556                                                         ctx->request,
4557                                                         "Invalid field referenced in ORDER BY clause -- "
4558                                                         "see error log for more details"
4559                                                 );
4560                                         free(having_buf);
4561                                         buffer_free(group_buf);
4562                                         buffer_free(sql_buf);
4563                                         if (defaultselhash)
4564                                                 jsonObjectFree(defaultselhash);
4565                                         return NULL;
4566                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4567                                         osrfLogError(OSRF_LOG_MARK, "%s: Virtual field \"%s\" in ORDER BY clause",
4568                                                                  MODULENAME, field );
4569                                         if( ctx )
4570                                                 osrfAppSessionStatus(
4571                                                         ctx->session,
4572                                                         OSRF_STATUS_INTERNALSERVERERROR,
4573                                                         "osrfMethodException",
4574                                                         ctx->request,
4575                                                         "Virtual field in ORDER BY clause -- see error log for more details"
4576                                                 );
4577                                         buffer_free( order_buf );
4578                                         free(having_buf);
4579                                         buffer_free(group_buf);
4580                                         buffer_free(sql_buf);
4581                                         if (defaultselhash)
4582                                                 jsonObjectFree(defaultselhash);
4583                                         return NULL;
4584                                 }
4585
4586                                 if( jsonObjectGetKeyConst( order_spec, "transform" ) ) {
4587                                         char* transform_str = searchFieldTransform(
4588                                                         class_alias, field_def, order_spec );
4589                                         if( ! transform_str ) {
4590                                                 if( ctx )
4591                                                         osrfAppSessionStatus(
4592                                                                 ctx->session,
4593                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4594                                                                 "osrfMethodException",
4595                                                                 ctx->request,
4596                                                                 "Severe query error in ORDER BY clause -- "
4597                                                                 "see error log for more details"
4598                                                         );
4599                                                 buffer_free( order_buf );
4600                                                 free(having_buf);
4601                                                 buffer_free(group_buf);
4602                                                 buffer_free(sql_buf);
4603                                                 if (defaultselhash)
4604                                                         jsonObjectFree(defaultselhash);
4605                                                 return NULL;
4606                                         }
4607
4608                                         OSRF_BUFFER_ADD( order_buf, transform_str );
4609                                         free( transform_str );
4610                                 }
4611                                 else
4612                                         buffer_fadd( order_buf, "\"%s\".%s", class_alias, field );
4613
4614                                 const char* direction =
4615                                                 jsonObjectGetString( jsonObjectGetKeyConst( order_spec, "direction" ) );
4616                                 if( direction ) {
4617                                         if( direction[ 0 ] || 'D' == direction[ 0 ] )
4618                                                 OSRF_BUFFER_ADD( order_buf, " DESC" );
4619                                         else
4620                                                 OSRF_BUFFER_ADD( order_buf, " ASC" );
4621                                 }
4622                         }
4623                 } else if( JSON_HASH == order_hash->type ) {
4624                         // This hash is keyed on class alias.  Each class has either
4625                         // an array of field names or a hash keyed on field name.
4626                         jsonIterator* class_itr = jsonNewIterator( order_hash );
4627                         while ( (snode = jsonIteratorNext( class_itr )) ) {
4628
4629                                 ClassInfo* order_class_info = search_alias( class_itr->key );
4630                                 if( ! order_class_info ) {
4631                                         osrfLogError(OSRF_LOG_MARK,
4632                                                 "%s: Invalid class \"%s\" referenced in ORDER BY clause",
4633                                                 MODULENAME, class_itr->key );
4634                                         if( ctx )
4635                                                 osrfAppSessionStatus(
4636                                                         ctx->session,
4637                                                         OSRF_STATUS_INTERNALSERVERERROR,
4638                                                         "osrfMethodException",
4639                                                         ctx->request,
4640                                                         "Invalid class referenced in ORDER BY clause -- "
4641                                                         "see error log for more details"
4642                                                 );
4643                                         jsonIteratorFree( class_itr );
4644                                         buffer_free( order_buf );
4645                                         free(having_buf);
4646                                         buffer_free(group_buf);
4647                                         buffer_free(sql_buf);
4648                                         if (defaultselhash)
4649                                                 jsonObjectFree(defaultselhash);
4650                                         return NULL;
4651                                 }
4652
4653                                 osrfHash* field_list_def = order_class_info->fields;
4654
4655                                 if ( snode->type == JSON_HASH ) {
4656
4657                                         // Hash is keyed on field names from the current class.  For each field
4658                                         // there is another layer of hash to define the sorting details, if any,
4659                                         // or a string to indicate direction of sorting.
4660                                         jsonIterator* order_itr = jsonNewIterator( snode );
4661                                         while ( (onode = jsonIteratorNext( order_itr )) ) {
4662
4663                                                 osrfHash* field_def = osrfHashGet( field_list_def, order_itr->key );
4664                                                 if( !field_def ) {
4665                                                         osrfLogError( OSRF_LOG_MARK,
4666                                                                 "%s: Invalid field \"%s\" in ORDER BY clause",
4667                                                                 MODULENAME, order_itr->key );
4668                                                         if( ctx )
4669                                                                 osrfAppSessionStatus(
4670                                                                         ctx->session,
4671                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4672                                                                         "osrfMethodException",
4673                                                                         ctx->request,
4674                                                                         "Invalid field in ORDER BY clause -- see error log for more details"
4675                                                                 );
4676                                                         jsonIteratorFree( order_itr );
4677                                                         jsonIteratorFree( class_itr );
4678                                                         buffer_free( order_buf );
4679                                                         free(having_buf);
4680                                                         buffer_free(group_buf);
4681                                                         buffer_free(sql_buf);
4682                                                         if (defaultselhash)
4683                                                                 jsonObjectFree(defaultselhash);
4684                                                         return NULL;
4685                                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4686                                                         osrfLogError( OSRF_LOG_MARK,
4687                                                                 "%s: Virtual field \"%s\" in ORDER BY clause",
4688                                                                 MODULENAME, order_itr->key );
4689                                                         if( ctx )
4690                                                                 osrfAppSessionStatus(
4691                                                                         ctx->session,
4692                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4693                                                                         "osrfMethodException",
4694                                                                         ctx->request,
4695                                                                         "Virtual field in ORDER BY clause -- "
4696                                                                         "see error log for more details"
4697                                                         );
4698                                                         jsonIteratorFree( order_itr );
4699                                                         jsonIteratorFree( class_itr );
4700                                                         buffer_free( order_buf );
4701                                                         free(having_buf);
4702                                                         buffer_free(group_buf);
4703                                                         buffer_free(sql_buf);
4704                                                         if (defaultselhash)
4705                                                                 jsonObjectFree(defaultselhash);
4706                                                         return NULL;
4707                                                 }
4708
4709                                                 const char* direction = NULL;
4710                                                 if ( onode->type == JSON_HASH ) {
4711                                                         if ( jsonObjectGetKeyConst( onode, "transform" ) ) {
4712                                                                 string = searchFieldTransform(
4713                                                                         class_itr->key,
4714                                                                         osrfHashGet( field_list_def, order_itr->key ),
4715                                                                         onode
4716                                                                 );
4717                                                                 if( ! string ) {
4718                                                                         if( ctx ) osrfAppSessionStatus(
4719                                                                                 ctx->session,
4720                                                                                 OSRF_STATUS_INTERNALSERVERERROR,
4721                                                                                 "osrfMethodException",
4722                                                                                 ctx->request,
4723                                                                                 "Severe query error in ORDER BY clause -- "
4724                                                                                 "see error log for more details"
4725                                                                         );
4726                                                                         jsonIteratorFree( order_itr );
4727                                                                         jsonIteratorFree( class_itr );
4728                                                                         free(having_buf);
4729                                                                         buffer_free(group_buf);
4730                                                                         buffer_free(order_buf);
4731                                                                         buffer_free(sql_buf);
4732                                                                         if (defaultselhash)
4733                                                                                 jsonObjectFree(defaultselhash);
4734                                                                         return NULL;
4735                                                                 }
4736                                                         } else {
4737                                                                 growing_buffer* field_buf = buffer_init(16);
4738                                                                 buffer_fadd( field_buf, "\"%s\".%s",
4739                                                                         class_itr->key, order_itr->key );
4740                                                                 string = buffer_release(field_buf);
4741                                                         }
4742
4743                                                         if ( (tmp_const = jsonObjectGetKeyConst( onode, "direction" )) ) {
4744                                                                 const char* dir = jsonObjectGetString(tmp_const);
4745                                                                 if (!strncasecmp(dir, "d", 1)) {
4746                                                                         direction = " DESC";
4747                                                                 } else {
4748                                                                         direction = " ASC";
4749                                                                 }
4750                                                         }
4751
4752                                                 } else if ( JSON_NULL == onode->type || JSON_ARRAY == onode->type ) {
4753                                                         osrfLogError( OSRF_LOG_MARK,
4754                                                                 "%s: Expected JSON_STRING in ORDER BY clause; found %s",
4755                                                                 MODULENAME, json_type( onode->type ) );
4756                                                         if( ctx )
4757                                                                 osrfAppSessionStatus(
4758                                                                         ctx->session,
4759                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4760                                                                         "osrfMethodException",
4761                                                                         ctx->request,
4762                                                                         "Malformed ORDER BY clause -- see error log for more details"
4763                                                                 );
4764                                                         jsonIteratorFree( order_itr );
4765                                                         jsonIteratorFree( class_itr );
4766                                                         free(having_buf);
4767                                                         buffer_free(group_buf);
4768                                                         buffer_free(order_buf);
4769                                                         buffer_free(sql_buf);
4770                                                         if (defaultselhash)
4771                                                                 jsonObjectFree(defaultselhash);
4772                                                         return NULL;
4773
4774                                                 } else {
4775                                                         string = strdup(order_itr->key);
4776                                                         const char* dir = jsonObjectGetString(onode);
4777                                                         if (!strncasecmp(dir, "d", 1)) {
4778                                                                 direction = " DESC";
4779                                                         } else {
4780                                                                 direction = " ASC";
4781                                                         }
4782                                                 }
4783
4784                                                 if ( order_buf )
4785                                                         OSRF_BUFFER_ADD(order_buf, ", ");
4786                                                 else
4787                                                         order_buf = buffer_init(128);
4788
4789                                                 OSRF_BUFFER_ADD(order_buf, string);
4790                                                 free(string);
4791
4792                                                 if (direction) {
4793                                                          OSRF_BUFFER_ADD(order_buf, direction);
4794                                                 }
4795
4796                                         } // end while
4797                                         jsonIteratorFree(order_itr);
4798
4799                                 } else if ( snode->type == JSON_ARRAY ) {
4800
4801                                         // Array is a list of fields from the current class
4802                                         unsigned long order_idx = 0;
4803                                         while(( onode = jsonObjectGetIndex( snode, order_idx++ ) )) {
4804
4805                                                 const char* _f = jsonObjectGetString( onode );
4806
4807                                                 osrfHash* field_def = osrfHashGet( field_list_def, _f );
4808                                                 if( !field_def ) {
4809                                                         osrfLogError( OSRF_LOG_MARK,
4810                                                                         "%s: Invalid field \"%s\" in ORDER BY clause",
4811                                                                         MODULENAME, _f );
4812                                                         if( ctx )
4813                                                                 osrfAppSessionStatus(
4814                                                                         ctx->session,
4815                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4816                                                                         "osrfMethodException",
4817                                                                         ctx->request,
4818                                                                         "Invalid field in ORDER BY clause -- "
4819                                                                         "see error log for more details"
4820                                                                 );
4821                                                         jsonIteratorFree( class_itr );
4822                                                         buffer_free( order_buf );
4823                                                         free(having_buf);
4824                                                         buffer_free(group_buf);
4825                                                         buffer_free(sql_buf);
4826                                                         if (defaultselhash)
4827                                                                 jsonObjectFree(defaultselhash);
4828                                                         return NULL;
4829                                                 } else if( str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4830                                                         osrfLogError( OSRF_LOG_MARK,
4831                                                                 "%s: Virtual field \"%s\" in ORDER BY clause",
4832                                                                 MODULENAME, _f );
4833                                                         if( ctx )
4834                                                                 osrfAppSessionStatus(
4835                                                                         ctx->session,
4836                                                                         OSRF_STATUS_INTERNALSERVERERROR,
4837                                                                         "osrfMethodException",
4838                                                                         ctx->request,
4839                                                                         "Virtual field in ORDER BY clause -- "
4840                                                                         "see error log for more details"
4841                                                                 );
4842                                                         jsonIteratorFree( class_itr );
4843                                                         buffer_free( order_buf );
4844                                                         free(having_buf);
4845                                                         buffer_free(group_buf);
4846                                                         buffer_free(sql_buf);
4847                                                         if (defaultselhash)
4848                                                                 jsonObjectFree(defaultselhash);
4849                                                         return NULL;
4850                                                 }
4851
4852                                                 if ( order_buf )
4853                                                         OSRF_BUFFER_ADD(order_buf, ", ");
4854                                                 else
4855                                                         order_buf = buffer_init(128);
4856
4857                                                 buffer_fadd( order_buf, "\"%s\".%s", class_itr->key, _f);
4858
4859                                         } // end while
4860
4861                                 // IT'S THE OOOOOOOOOOOLD STYLE!
4862                                 } else {
4863                                         osrfLogError(OSRF_LOG_MARK,
4864                                                 "%s: Possible SQL injection attempt; direct order by is not allowed",
4865                                                 MODULENAME);
4866                                         if (ctx) {
4867                                                 osrfAppSessionStatus(
4868                                                         ctx->session,
4869                                                         OSRF_STATUS_INTERNALSERVERERROR,
4870                                                         "osrfMethodException",
4871                                                         ctx->request,
4872                                                         "Severe query error -- see error log for more details"
4873                                                 );
4874                                         }
4875
4876                                         free(having_buf);
4877                                         buffer_free(group_buf);
4878                                         buffer_free(order_buf);
4879                                         buffer_free(sql_buf);
4880                                         if (defaultselhash)
4881                                                 jsonObjectFree(defaultselhash);
4882                                         jsonIteratorFree(class_itr);
4883                                         return NULL;
4884                                 }
4885                         } // end while
4886                         jsonIteratorFree( class_itr );
4887                 } else {
4888                         osrfLogError(OSRF_LOG_MARK,
4889                                 "%s: Malformed ORDER BY clause; expected JSON_HASH or JSON_ARRAY, found %s",
4890                                 MODULENAME, json_type( order_hash->type ) );
4891                         if( ctx )
4892                                 osrfAppSessionStatus(
4893                                         ctx->session,
4894                                         OSRF_STATUS_INTERNALSERVERERROR,
4895                                         "osrfMethodException",
4896                                         ctx->request,
4897                                         "Malformed ORDER BY clause -- see error log for more details"
4898                                 );
4899                         buffer_free( order_buf );
4900                         free(having_buf);
4901                         buffer_free(group_buf);
4902                         buffer_free(sql_buf);
4903                         if (defaultselhash)
4904                                 jsonObjectFree(defaultselhash);
4905                         return NULL;
4906                 }
4907
4908                 if( order_buf )
4909                         order_by_list = buffer_release( order_buf );
4910         }
4911
4912
4913         string = buffer_release(group_buf);
4914
4915         if ( *string && ( aggregate_found || (flags & SELECT_DISTINCT) ) ) {
4916                 OSRF_BUFFER_ADD( sql_buf, " GROUP BY " );
4917                 OSRF_BUFFER_ADD( sql_buf, string );
4918         }
4919
4920         free(string);
4921
4922         if( having_buf && *having_buf ) {
4923                 OSRF_BUFFER_ADD( sql_buf, " HAVING " );
4924                 OSRF_BUFFER_ADD( sql_buf, having_buf );
4925                 free( having_buf );
4926         }
4927
4928         if( order_by_list ) {
4929
4930                 if ( *order_by_list ) {
4931                         OSRF_BUFFER_ADD( sql_buf, " ORDER BY " );
4932                         OSRF_BUFFER_ADD( sql_buf, order_by_list );
4933                 }
4934
4935                 free( order_by_list );
4936         }
4937
4938         if ( limit ){
4939                 const char* str = jsonObjectGetString(limit);
4940                 buffer_fadd( sql_buf, " LIMIT %d", atoi(str) );
4941         }
4942
4943         if (offset) {
4944                 const char* str = jsonObjectGetString(offset);
4945                 buffer_fadd( sql_buf, " OFFSET %d", atoi(str) );
4946         }
4947
4948         if (!(flags & SUBSELECT)) OSRF_BUFFER_ADD_CHAR(sql_buf, ';');
4949
4950         if (defaultselhash)
4951                  jsonObjectFree(defaultselhash);
4952
4953         return buffer_release(sql_buf);
4954
4955 } // end of SELECT()
4956
4957 static char* buildSELECT ( jsonObject* search_hash, jsonObject* order_hash, osrfHash* meta, osrfMethodContext* ctx ) {
4958
4959         const char* locale = osrf_message_get_last_locale();
4960
4961         osrfHash* fields = osrfHashGet(meta, "fields");
4962         char* core_class = osrfHashGet(meta, "classname");
4963
4964         const jsonObject* join_hash = jsonObjectGetKeyConst( order_hash, "join" );
4965
4966         jsonObject* node = NULL;
4967         jsonObject* snode = NULL;
4968         jsonObject* onode = NULL;
4969         const jsonObject* _tmp = NULL;
4970         jsonObject* selhash = NULL;
4971         jsonObject* defaultselhash = NULL;
4972
4973         growing_buffer* sql_buf = buffer_init(128);
4974         growing_buffer* select_buf = buffer_init(128);
4975
4976         if ( !(selhash = jsonObjectGetKey( order_hash, "select" )) ) {
4977                 defaultselhash = jsonNewObjectType(JSON_HASH);
4978                 selhash = defaultselhash;
4979         }
4980
4981         // If there's no SELECT list for the core class, build one
4982         if ( !jsonObjectGetKeyConst(selhash,core_class) ) {
4983                 jsonObject* field_list = jsonNewObjectType( JSON_ARRAY );
4984
4985                 // Add every non-virtual field to the field list
4986                 osrfHash* field_def = NULL;
4987                 osrfHashIterator* field_itr = osrfNewHashIterator( fields );
4988                 while( ( field_def = osrfHashIteratorNext( field_itr ) ) ) {
4989                         if( ! str_is_true( osrfHashGet( field_def, "virtual" ) ) ) {
4990                                 const char* field = osrfHashIteratorKey( field_itr );
4991                                 jsonObjectPush( field_list, jsonNewObject( field ) );
4992                         }
4993                 }
4994                 osrfHashIteratorFree( field_itr );
4995                 jsonObjectSetKey( selhash, core_class, field_list );
4996         }
4997
4998         int first = 1;
4999         jsonIterator* class_itr = jsonNewIterator( selhash );
5000         while ( (snode = jsonIteratorNext( class_itr )) ) {
5001
5002                 const char* cname = class_itr->key;
5003                 osrfHash* idlClass = osrfHashGet( oilsIDL(), cname );
5004                 if (!idlClass)
5005                         continue;
5006
5007                 if (strcmp(core_class,class_itr->key)) {
5008                         if (!join_hash)
5009                                 continue;
5010
5011                         jsonObject* found =  jsonObjectFindPath(join_hash, "//%s", class_itr->key);
5012                         if (!found->size) {
5013                                 jsonObjectFree(found);
5014                                 continue;
5015                         }
5016
5017                         jsonObjectFree(found);
5018                 }
5019
5020                 jsonIterator* select_itr = jsonNewIterator( snode );
5021                 while ( (node = jsonIteratorNext( select_itr )) ) {
5022                         const char* item_str = jsonObjectGetString( node );
5023                         osrfHash* field = osrfHashGet( osrfHashGet( idlClass, "fields" ), item_str );
5024                         char* fname = osrfHashGet(field, "name");
5025
5026                         if (!field)
5027                                 continue;
5028
5029                         if (first) {
5030                                 first = 0;
5031                         } else {
5032                                 OSRF_BUFFER_ADD_CHAR(select_buf, ',');
5033                         }
5034
5035                         if (locale) {
5036                                 const char* i18n;
5037                                 const jsonObject* no_i18n_obj = jsonObjectGetKey( order_hash, "no_i18n" );
5038                                 if ( obj_is_true( no_i18n_obj ) )    // Suppress internationalization?
5039                                         i18n = NULL;
5040                                 else
5041                                         i18n = osrfHashGet(field, "i18n");
5042
5043                                 if( str_is_true( i18n ) ) {
5044                                         char* pkey = osrfHashGet(idlClass, "primarykey");
5045                                         char* tname = osrfHashGet(idlClass, "tablename");
5046
5047                                         buffer_fadd(select_buf, " oils_i18n_xlate('%s', '%s', '%s', "
5048                                                         "'%s', \"%s\".%s::TEXT, '%s') AS \"%s\"",
5049                                                         tname, cname, fname, pkey, cname, pkey, locale, fname);
5050                                 } else {
5051                                         buffer_fadd(select_buf, " \"%s\".%s", cname, fname);
5052                                 }
5053                         } else {
5054                                 buffer_fadd(select_buf, " \"%s\".%s", cname, fname);
5055                         }
5056                 }
5057
5058                 jsonIteratorFree(select_itr);
5059         }
5060
5061         jsonIteratorFree(class_itr);
5062
5063         char* col_list = buffer_release(select_buf);
5064         char* table = getRelation(meta);
5065         if( !table )
5066                 table = strdup( "(null)" );
5067
5068         buffer_fadd(sql_buf, "SELECT %s FROM %s AS \"%s\"", col_list, table, core_class );
5069         free(col_list);
5070         free(table);
5071
5072         // Clear the query stack (as a fail-safe precaution against possible
5073         // leftover garbage); then push the first query frame onto the stack.
5074         clear_query_stack();
5075         push_query_frame();
5076         if( add_query_core( NULL, core_class ) ) {
5077                 if( ctx )
5078                         osrfAppSessionStatus(
5079                                 ctx->session,
5080                                 OSRF_STATUS_INTERNALSERVERERROR,
5081                                 "osrfMethodException",
5082                                 ctx->request,
5083                                 "Unable to build query frame for core class"
5084                         );
5085                 return NULL;
5086         }
5087
5088         if ( join_hash ) {
5089                 char* join_clause = searchJOIN( join_hash, &curr_query->core );
5090                 OSRF_BUFFER_ADD_CHAR(sql_buf, ' ');
5091                 OSRF_BUFFER_ADD(sql_buf, join_clause);
5092                 free(join_clause);
5093         }
5094
5095         osrfLogDebug(OSRF_LOG_MARK, "%s pre-predicate SQL =  %s",
5096                                  MODULENAME, OSRF_BUFFER_C_STR(sql_buf));
5097
5098         OSRF_BUFFER_ADD(sql_buf, " WHERE ");
5099
5100         char* pred = searchWHERE( search_hash, &curr_query->core, AND_OP_JOIN, ctx );
5101         if (!pred) {
5102                 osrfAppSessionStatus(
5103                         ctx->session,
5104                         OSRF_STATUS_INTERNALSERVERERROR,
5105                                 "osrfMethodException",
5106                                 ctx->request,
5107                                 "Severe query error -- see error log for more details"
5108                         );
5109                 buffer_free(sql_buf);
5110                 if(defaultselhash)
5111                         jsonObjectFree(defaultselhash);
5112                 clear_query_stack();
5113                 return NULL;
5114         } else {
5115                 buffer_add(sql_buf, pred);
5116                 free(pred);
5117         }
5118
5119         if (order_hash) {
5120                 char* string = NULL;
5121                 if ( (_tmp = jsonObjectGetKeyConst( order_hash, "order_by" )) ){
5122
5123                         growing_buffer* order_buf = buffer_init(128);
5124
5125                         first = 1;
5126                         jsonIterator* class_itr = jsonNewIterator( _tmp );
5127                         while ( (snode = jsonIteratorNext( class_itr )) ) {
5128
5129                                 if (!jsonObjectGetKeyConst(selhash,class_itr->key))
5130                                         continue;
5131
5132                                 if ( snode->type == JSON_HASH ) {
5133
5134                                         jsonIterator* order_itr = jsonNewIterator( snode );
5135                                         while ( (onode = jsonIteratorNext( order_itr )) ) {
5136
5137                                                 osrfHash* field_def = oilsIDLFindPath( "/%s/fields/%s",
5138                                                                 class_itr->key, order_itr->key );
5139                                                 if ( !field_def )
5140                                                         continue;
5141
5142                                                 char* direction = NULL;
5143                                                 if ( onode->type == JSON_HASH ) {
5144                                                         if ( jsonObjectGetKeyConst( onode, "transform" ) ) {
5145                                                                 string = searchFieldTransform( class_itr->key, field_def, onode );
5146                                                                 if( ! string ) {
5147                                                                         osrfAppSessionStatus(
5148                                                                                 ctx->session,
5149                                                                                 OSRF_STATUS_INTERNALSERVERERROR,
5150                                                                                 "osrfMethodException",
5151                                                                                 ctx->request,
5152                                                                                 "Severe query error in ORDER BY clause -- "
5153                                                                                 "see error log for more details"
5154                                                                         );
5155                                                                         jsonIteratorFree( order_itr );
5156                                                                         jsonIteratorFree( class_itr );
5157                                                                         buffer_free( order_buf );
5158                                                                         buffer_free( sql_buf );
5159                                                                         if( defaultselhash )
5160                                                                                 jsonObjectFree( defaultselhash );
5161                                                                         clear_query_stack();
5162                                                                         return NULL;
5163                                                                 }
5164                                                         } else {
5165                                                                 growing_buffer* field_buf = buffer_init(16);
5166                                                                 buffer_fadd( field_buf, "\"%s\".%s",
5167                                                                         class_itr->key, order_itr->key );
5168                                                                 string = buffer_release(field_buf);
5169                                                         }
5170
5171                                                         if ( (_tmp = jsonObjectGetKeyConst( onode, "direction" )) ) {
5172                                                                 const char* dir = jsonObjectGetString(_tmp);
5173                                                                 if (!strncasecmp(dir, "d", 1)) {
5174                                                                         direction = " DESC";
5175                                                                 } else {
5176                                                                         free(direction);
5177                                                                 }
5178                                                         }
5179                                                 } else {
5180                                                         string = strdup(order_itr->key);
5181                                                         const char* dir = jsonObjectGetString(onode);
5182                                                         if (!strncasecmp(dir, "d", 1)) {
5183                                                                 direction = " DESC";
5184                                                         } else {
5185                                                                 direction = " ASC";
5186                                                         }
5187                                                 }
5188
5189                                                 if (first) {
5190                                                         first = 0;
5191                                                 } else {
5192                                                         buffer_add(order_buf, ", ");
5193                                                 }
5194
5195                                                 buffer_add(order_buf, string);
5196                                                 free(string);
5197
5198                                                 if (direction) {
5199                                                         buffer_add(order_buf, direction);
5200                                                 }
5201                                         }
5202
5203                                         jsonIteratorFree(order_itr);
5204
5205                                 } else {
5206                                         const char* str = jsonObjectGetString(snode);
5207                                         buffer_add(order_buf, str);
5208                                         break;
5209                                 }
5210
5211                         }
5212
5213                         jsonIteratorFree(class_itr);
5214
5215                         string = buffer_release(order_buf);
5216
5217                         if ( *string ) {
5218                                 OSRF_BUFFER_ADD( sql_buf, " ORDER BY " );
5219                                 OSRF_BUFFER_ADD( sql_buf, string );
5220                         }
5221
5222                         free(string);
5223                 }
5224
5225                 if ( (_tmp = jsonObjectGetKeyConst( order_hash, "limit" )) ){
5226                         const char* str = jsonObjectGetString(_tmp);
5227                         buffer_fadd(
5228                                 sql_buf,
5229                                 " LIMIT %d",
5230                                 atoi(str)
5231                         );
5232                 }
5233
5234                 _tmp = jsonObjectGetKeyConst( order_hash, "offset" );
5235                 if (_tmp) {
5236                         const char* str = jsonObjectGetString(_tmp);
5237                         buffer_fadd(
5238                                 sql_buf,
5239                                 " OFFSET %d",
5240                                 atoi(str)
5241                         );
5242                 }
5243         }
5244
5245         if (defaultselhash)
5246                 jsonObjectFree(defaultselhash);
5247         clear_query_stack();
5248
5249         OSRF_BUFFER_ADD_CHAR(sql_buf, ';');
5250         return buffer_release(sql_buf);
5251 }
5252
5253 int doJSONSearch ( osrfMethodContext* ctx ) {
5254         if(osrfMethodVerifyContext( ctx )) {
5255                 osrfLogError( OSRF_LOG_MARK,  "Invalid method context" );
5256                 return -1;
5257         }
5258
5259         osrfLogDebug(OSRF_LOG_MARK, "Received query request");
5260
5261         int err = 0;
5262
5263         // XXX for now...
5264         dbhandle = writehandle;
5265
5266         jsonObject* hash = jsonObjectGetIndex(ctx->params, 0);
5267
5268         int flags = 0;
5269
5270         if ( obj_is_true( jsonObjectGetKey( hash, "distinct" ) ) )
5271                 flags |= SELECT_DISTINCT;
5272
5273         if ( obj_is_true( jsonObjectGetKey( hash, "no_i18n" ) ) )
5274                 flags |= DISABLE_I18N;
5275
5276         osrfLogDebug(OSRF_LOG_MARK, "Building SQL ...");
5277         clear_query_stack();       // a possibly needless precaution
5278         char* sql = buildQuery( ctx, hash, flags );
5279         clear_query_stack();
5280
5281         if (!sql) {
5282                 err = -1;
5283                 return err;
5284         }
5285
5286         osrfLogDebug(OSRF_LOG_MARK, "%s SQL =  %s", MODULENAME, sql);
5287         dbi_result result = dbi_conn_query(dbhandle, sql);
5288
5289         if(result) {
5290                 osrfLogDebug(OSRF_LOG_MARK, "Query returned with no errors");
5291
5292                 if (dbi_result_first_row(result)) {
5293                         /* JSONify the result */
5294                         osrfLogDebug(OSRF_LOG_MARK, "Query returned at least one row");
5295
5296                         do {
5297                                 jsonObject* return_val = oilsMakeJSONFromResult( result );
5298                                 osrfAppRespond( ctx, return_val );
5299                                 jsonObjectFree( return_val );
5300                         } while (dbi_result_next_row(result));
5301
5302                 } else {
5303                         osrfLogDebug(OSRF_LOG_MARK, "%s returned no results for query %s", MODULENAME, sql);
5304                 }
5305
5306                 osrfAppRespondComplete( ctx, NULL );
5307
5308                 /* clean up the query */
5309                 dbi_result_free(result);
5310
5311         } else {
5312                 err = -1;
5313                 osrfLogError(OSRF_LOG_MARK, "%s: Error with query [%s]", MODULENAME, sql);
5314                 osrfAppSessionStatus(
5315                         ctx->session,
5316                         OSRF_STATUS_INTERNALSERVERERROR,
5317                         "osrfMethodException",
5318                         ctx->request,
5319                         "Severe query error -- see error log for more details"
5320                 );
5321         }
5322
5323         free(sql);
5324         return err;
5325 }
5326
5327 // The last parameter, err, is used to report an error condition by updating an int owned by
5328 // the calling code.
5329
5330 // In case of an error, we set *err to -1.  If there is no error, *err is left unchanged.
5331 // It is the responsibility of the calling code to initialize *err before the
5332 // call, so that it will be able to make sense of the result.
5333
5334 // Note also that we return NULL if and only if we set *err to -1.  So the err parameter is
5335 // redundant anyway.
5336 static jsonObject* doFieldmapperSearch ( osrfMethodContext* ctx, osrfHash* class_meta,
5337                 jsonObject* where_hash, jsonObject* query_hash, int* err ) {
5338
5339         // XXX for now...
5340         dbhandle = writehandle;
5341
5342         char* core_class = osrfHashGet( class_meta, "classname" );
5343         char* pkey = osrfHashGet( class_meta, "primarykey" );
5344
5345         const jsonObject* _tmp;
5346
5347         char* sql = buildSELECT( where_hash, query_hash, class_meta, ctx );
5348         if (!sql) {
5349                 osrfLogDebug(OSRF_LOG_MARK, "Problem building query, returning NULL");
5350                 *err = -1;
5351                 return NULL;
5352         }
5353
5354         osrfLogDebug(OSRF_LOG_MARK, "%s SQL =  %s", MODULENAME, sql);
5355
5356         dbi_result result = dbi_conn_query(dbhandle, sql);
5357         if( NULL == result ) {
5358                 osrfLogError(OSRF_LOG_MARK, "%s: Error retrieving %s with query [%s]",
5359                         MODULENAME, osrfHashGet( class_meta, "fieldmapper" ), sql);
5360                 osrfAppSessionStatus(
5361                         ctx->session,
5362                         OSRF_STATUS_INTERNALSERVERERROR,
5363                         "osrfMethodException",
5364                         ctx->request,
5365                         "Severe query error -- see error log for more details"
5366                 );
5367                 *err = -1;
5368                 free(sql);
5369                 return jsonNULL;
5370
5371         } else {
5372                 osrfLogDebug(OSRF_LOG_MARK, "Query returned with no errors");
5373         }
5374
5375         jsonObject* res_list = jsonNewObjectType(JSON_ARRAY);
5376         jsonObject* row_obj = NULL;
5377
5378         if (dbi_result_first_row(result)) {
5379
5380                 // Convert each row to a JSON_ARRAY of column values, and enclose those objects
5381                 // in a JSON_ARRAY of rows.  If two or more rows have the same key value, then
5382                 // eliminate the duplicates.
5383                 osrfLogDebug(OSRF_LOG_MARK, "Query returned at least one row");
5384                 osrfHash* dedup = osrfNewHash();
5385                 do {
5386                         row_obj = oilsMakeFieldmapperFromResult( result, class_meta );
5387                         char* pkey_val = oilsFMGetString( row_obj, pkey );
5388                         if ( osrfHashGet( dedup, pkey_val ) ) {
5389                                 jsonObjectFree( row_obj );
5390                                 free( pkey_val );
5391                         } else {
5392                                 osrfHashSet( dedup, pkey_val, pkey_val );
5393                                 jsonObjectPush( res_list, row_obj );
5394                         }
5395                 } while (dbi_result_next_row(result));
5396                 osrfHashFree(dedup);
5397
5398         } else {
5399                 osrfLogDebug(OSRF_LOG_MARK, "%s returned no results for query %s",
5400                         MODULENAME, sql );
5401         }
5402
5403         /* clean up the query */
5404         dbi_result_free(result);
5405         free(sql);
5406
5407         // If we're asked to flesh, and there's anything to flesh, then flesh.
5408         if (res_list->size && query_hash) {
5409                 _tmp = jsonObjectGetKeyConst( query_hash, "flesh" );
5410                 if (_tmp) {
5411                         // Get the flesh depth
5412                         int flesh_depth = (int) jsonObjectGetNumber( _tmp );
5413                         if ( flesh_depth == -1 || flesh_depth > max_flesh_depth )
5414                                 flesh_depth = max_flesh_depth;
5415
5416                         // We need a non-zero flesh depth, and a list of fields to flesh
5417                         const jsonObject* temp_blob = jsonObjectGetKeyConst( query_hash, "flesh_fields" );
5418                         if ( temp_blob && flesh_depth > 0 ) {
5419
5420                                 jsonObject* flesh_blob = jsonObjectClone( temp_blob );
5421                                 const jsonObject* flesh_fields = jsonObjectGetKeyConst( flesh_blob, core_class );
5422
5423                                 osrfStringArray* link_fields = NULL;
5424                                 osrfHash* links = osrfHashGet( class_meta, "links" );
5425
5426                                 // Make an osrfStringArray of the names of fields to be fleshed
5427                                 if (flesh_fields) {
5428                                         if (flesh_fields->size == 1) {
5429                                                 const char* _t = jsonObjectGetString(
5430                                                         jsonObjectGetIndex( flesh_fields, 0 ) );
5431                                                 if (!strcmp(_t,"*"))
5432                                                         link_fields = osrfHashKeys( links );
5433                                         }
5434
5435                                         if (!link_fields) {
5436                                                 jsonObject* _f;
5437                                                 link_fields = osrfNewStringArray(1);
5438                                                 jsonIterator* _i = jsonNewIterator( flesh_fields );
5439                                                 while ((_f = jsonIteratorNext( _i ))) {
5440                                                         osrfStringArrayAdd( link_fields, jsonObjectGetString( _f ) );
5441                                                 }
5442                                                 jsonIteratorFree(_i);
5443                                         }
5444                                 }
5445
5446                                 osrfHash* fields = osrfHashGet( class_meta, "fields" );
5447
5448                                 // Iterate over the JSON_ARRAY of rows
5449                                 jsonObject* cur;
5450                                 unsigned long res_idx = 0;
5451                                 while ((cur = jsonObjectGetIndex( res_list, res_idx++ ) )) {
5452
5453                                         int i = 0;
5454                                         const char* link_field;
5455
5456                                         // Iterate over the list of fleshable fields
5457                                         while ( (link_field = osrfStringArrayGetString(link_fields, i++)) ) {
5458
5459                                                 osrfLogDebug(OSRF_LOG_MARK, "Starting to flesh %s", link_field);
5460
5461                                                 osrfHash* kid_link = osrfHashGet(links, link_field);
5462                                                 if (!kid_link)
5463                                                         continue;     // Not a link field; skip it
5464
5465                                                 osrfHash* field = osrfHashGet(fields, link_field);
5466                                                 if (!field)
5467                                                         continue;     // Not a field at all; skip it (IDL is ill-formed)
5468
5469                                                 osrfHash* kid_idl = osrfHashGet(oilsIDL(), osrfHashGet(kid_link, "class"));
5470                                                 if (!kid_idl)
5471                                                         continue;   // The class it links to doesn't exist; skip it
5472
5473                                                 const char* reltype = osrfHashGet( kid_link, "reltype" );
5474                                                 if( !reltype )
5475                                                         continue;   // No reltype; skip it (IDL is ill-formed)
5476
5477                                                 osrfHash* value_field = field;
5478
5479                                                 if (   !strcmp( reltype, "has_many" )
5480                                                         || !strcmp( reltype, "might_have" ) ) { // has_many or might_have
5481                                                         value_field = osrfHashGet(
5482                                                                 fields, osrfHashGet( class_meta, "primarykey" ) );
5483                                                 }
5484
5485                                                 osrfStringArray* link_map = osrfHashGet( kid_link, "map" );
5486
5487                                                 if (link_map->size > 0) {
5488                                                         jsonObject* _kid_key = jsonNewObjectType(JSON_ARRAY);
5489                                                         jsonObjectPush(
5490                                                                 _kid_key,
5491                                                                 jsonNewObject( osrfStringArrayGetString( link_map, 0 ) )
5492                                                         );
5493
5494                                                         jsonObjectSetKey(
5495                                                                 flesh_blob,
5496                                                                 osrfHashGet(kid_link, "class"),
5497                                                                 _kid_key
5498                                                         );
5499                                                 };
5500
5501                                                 osrfLogDebug(
5502                                                         OSRF_LOG_MARK,
5503                                                         "Link field: %s, remote class: %s, fkey: %s, reltype: %s",
5504                                                         osrfHashGet(kid_link, "field"),
5505                                                         osrfHashGet(kid_link, "class"),
5506                                                         osrfHashGet(kid_link, "key"),
5507                                                         osrfHashGet(kid_link, "reltype")
5508                                                 );
5509
5510                                                 const char* search_key = jsonObjectGetString(
5511                                                         jsonObjectGetIndex( cur,
5512                                                                 atoi( osrfHashGet(value_field, "array_position") )
5513                                                         )
5514                                                 );
5515
5516                                                 if (!search_key) {
5517                                                         osrfLogDebug(OSRF_LOG_MARK, "Nothing to search for!");
5518                                                         continue;
5519                                                 }
5520
5521                                                 osrfLogDebug(OSRF_LOG_MARK, "Creating param objects...");
5522
5523                                                 // construct WHERE clause
5524                                                 jsonObject* where_clause  = jsonNewObjectType(JSON_HASH);
5525                                                 jsonObjectSetKey(
5526                                                         where_clause,
5527                                                         osrfHashGet(kid_link, "key"),
5528                                                         jsonNewObject( search_key )
5529                                                 );
5530
5531                                                 // construct the rest of the query, mostly
5532                                                 // by copying pieces of the previous level of query
5533                                                 jsonObject* rest_of_query = jsonNewObjectType(JSON_HASH);
5534                                                 jsonObjectSetKey( rest_of_query, "flesh",
5535                                                         jsonNewNumberObject( flesh_depth - 1 + link_map->size )
5536                                                 );
5537
5538                                                 if (flesh_blob)
5539                                                         jsonObjectSetKey( rest_of_query, "flesh_fields",
5540                                                                 jsonObjectClone(flesh_blob) );
5541
5542                                                 if (jsonObjectGetKeyConst(query_hash, "order_by")) {
5543                                                         jsonObjectSetKey( rest_of_query, "order_by",
5544                                                                 jsonObjectClone(jsonObjectGetKeyConst(query_hash, "order_by"))
5545                                                         );
5546                                                 }
5547
5548                                                 if (jsonObjectGetKeyConst(query_hash, "select")) {
5549                                                         jsonObjectSetKey( rest_of_query, "select",
5550                                                                 jsonObjectClone(jsonObjectGetKeyConst(query_hash, "select"))
5551                                                         );
5552                                                 }
5553
5554                                                 // do the query, recursively, to expand the fleshable field
5555                                                 jsonObject* kids = doFieldmapperSearch( ctx, kid_idl,
5556                                                         where_clause, rest_of_query, err);
5557
5558                                                 jsonObjectFree( where_clause );
5559                                                 jsonObjectFree( rest_of_query );
5560
5561                                                 if(*err) {
5562                                                         osrfStringArrayFree(link_fields);
5563                                                         jsonObjectFree(res_list);
5564                                                         jsonObjectFree(flesh_blob);
5565                                                         return jsonNULL;
5566                                                 }
5567
5568                                                 osrfLogDebug(OSRF_LOG_MARK, "Search for %s return %d linked objects",
5569                                                         osrfHashGet(kid_link, "class"), kids->size);
5570
5571                                                 // Traverse the result set
5572                                                 jsonObject* X = NULL;
5573                                                 if ( link_map->size > 0 && kids->size > 0 ) {
5574                                                         X = kids;
5575                                                         kids = jsonNewObjectType(JSON_ARRAY);
5576
5577                                                         jsonObject* _k_node;
5578                                                         unsigned long res_idx = 0;
5579                                                         while ((_k_node = jsonObjectGetIndex( X, res_idx++ ) )) {
5580                                                                 jsonObjectPush(
5581                                                                         kids,
5582                                                                         jsonObjectClone(
5583                                                                                 jsonObjectGetIndex(
5584                                                                                         _k_node,
5585                                                                                         (unsigned long)atoi(
5586                                                                                                 osrfHashGet(
5587                                                                                                         osrfHashGet(
5588                                                                                                                 osrfHashGet(
5589                                                                                                                         osrfHashGet(
5590                                                                                                                                 oilsIDL(),
5591                                                                                                                                 osrfHashGet(kid_link, "class")
5592                                                                                                                         ),
5593                                                                                                                         "fields"
5594                                                                                                                 ),
5595                                                                                                                 osrfStringArrayGetString( link_map, 0 )
5596                                                                                                         ),
5597                                                                                                         "array_position"
5598                                                                                                 )
5599                                                                                         )
5600                                                                                 )
5601                                                                         )
5602                                                                 );
5603                                                         } // end while loop traversing X
5604                                                 }
5605
5606                                                 if (!(strcmp( osrfHashGet(kid_link, "reltype"), "has_a" ))
5607                                                         || !(strcmp( osrfHashGet(kid_link, "reltype"), "might_have" ))) {
5608                                                         osrfLogDebug(OSRF_LOG_MARK, "Storing fleshed objects in %s",
5609                                                                 osrfHashGet(kid_link, "field"));
5610                                                         jsonObjectSetIndex(
5611                                                                 cur,
5612                                                                 (unsigned long)atoi( osrfHashGet( field, "array_position" ) ),
5613                                                                 jsonObjectClone( jsonObjectGetIndex(kids, 0) )
5614                                                         );
5615                                                 }
5616
5617                                                 if (!(strcmp( osrfHashGet(kid_link, "reltype"), "has_many" ))) {
5618                                                         // has_many
5619                                                         osrfLogDebug( OSRF_LOG_MARK, "Storing fleshed objects in %s",
5620                                                                 osrfHashGet( kid_link, "field" ) );
5621                                                         jsonObjectSetIndex(
5622                                                                 cur,
5623                                                                 (unsigned long)atoi( osrfHashGet( field, "array_position" ) ),
5624                                                                 jsonObjectClone( kids )
5625                                                         );
5626                                                 }
5627
5628                                                 if (X) {
5629                                                         jsonObjectFree(kids);
5630                                                         kids = X;
5631                                                 }
5632
5633                                                 jsonObjectFree( kids );
5634
5635                                                 osrfLogDebug( OSRF_LOG_MARK, "Fleshing of %s complete",
5636                                                         osrfHashGet( kid_link, "field" ) );
5637                                                 osrfLogDebug(OSRF_LOG_MARK, "%s", jsonObjectToJSON(cur));
5638
5639                                         } // end while loop traversing list of fleshable fields
5640                                 } // end while loop traversing res_list
5641                                 jsonObjectFree( flesh_blob );
5642                                 osrfStringArrayFree(link_fields);
5643                         }
5644                 }
5645         }
5646
5647         return res_list;
5648 }
5649
5650
5651 static int doUpdate(osrfMethodContext* ctx ) {
5652         if(osrfMethodVerifyContext( ctx )) {
5653                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
5654                 return -1;
5655         }
5656
5657         if( enforce_pcrud )
5658                 timeout_needs_resetting = 1;
5659
5660         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
5661
5662         jsonObject* target = NULL;
5663         if( enforce_pcrud )
5664                 target = jsonObjectGetIndex( ctx->params, 1 );
5665         else
5666                 target = jsonObjectGetIndex( ctx->params, 0 );
5667
5668         if (!verifyObjectClass(ctx, target)) {
5669                 osrfAppRespondComplete( ctx, NULL );
5670                 return -1;
5671         }
5672
5673         if( getXactId( ctx ) == NULL ) {
5674                 osrfAppSessionStatus(
5675                         ctx->session,
5676                         OSRF_STATUS_BADREQUEST,
5677                         "osrfMethodException",
5678                         ctx->request,
5679                         "No active transaction -- required for UPDATE"
5680                 );
5681                 osrfAppRespondComplete( ctx, NULL );
5682                 return -1;
5683         }
5684
5685         // The following test is harmless but redundant.  If a class is
5686         // readonly, we don't register an update method for it.
5687         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
5688                 osrfAppSessionStatus(
5689                         ctx->session,
5690                         OSRF_STATUS_BADREQUEST,
5691                         "osrfMethodException",
5692                         ctx->request,
5693                         "Cannot UPDATE readonly class"
5694                 );
5695                 osrfAppRespondComplete( ctx, NULL );
5696                 return -1;
5697         }
5698
5699         dbhandle = writehandle;
5700         const char* trans_id = getXactId( ctx );
5701
5702         // Set the last_xact_id
5703         int index = oilsIDL_ntop( target->classname, "last_xact_id" );
5704         if (index > -1) {
5705                 osrfLogDebug(OSRF_LOG_MARK, "Setting last_xact_id to %s on %s at position %d",
5706                                 trans_id, target->classname, index);
5707                 jsonObjectSetIndex(target, index, jsonNewObject(trans_id));
5708         }
5709
5710         char* pkey = osrfHashGet(meta, "primarykey");
5711         osrfHash* fields = osrfHashGet(meta, "fields");
5712
5713         char* id = oilsFMGetString( target, pkey );
5714
5715         osrfLogDebug(
5716                 OSRF_LOG_MARK,
5717                 "%s updating %s object with %s = %s",
5718                 MODULENAME,
5719                 osrfHashGet(meta, "fieldmapper"),
5720                 pkey,
5721                 id
5722         );
5723
5724         growing_buffer* sql = buffer_init(128);
5725         buffer_fadd(sql,"UPDATE %s SET", osrfHashGet(meta, "tablename"));
5726
5727         int first = 1;
5728         osrfHash* field_def = NULL;
5729         osrfHashIterator* field_itr = osrfNewHashIterator( fields );
5730         while( (field_def = osrfHashIteratorNext( field_itr ) ) ) {
5731
5732                 // Skip virtual fields, and the primary key
5733                 if( str_is_true( osrfHashGet( field_def, "virtual") ) )
5734                         continue;
5735
5736                 const char* field_name = osrfHashIteratorKey( field_itr );
5737                 if( ! strcmp( field_name, pkey ) )
5738                         continue;
5739
5740                 const jsonObject* field_object = oilsFMGetObject( target, field_name );
5741
5742                 int value_is_numeric = 0;    // boolean
5743                 char* value;
5744                 if (field_object && field_object->classname) {
5745                         value = oilsFMGetString(
5746                                 field_object,
5747                                 (char*)oilsIDLFindPath("/%s/primarykey", field_object->classname)
5748                         );
5749                 } else if( field_object && JSON_BOOL == field_object->type ) {
5750                         if( jsonBoolIsTrue( field_object ) )
5751                                 value = strdup( "t" );
5752                         else
5753                                 value = strdup( "f" );
5754                 } else {
5755                         value = jsonObjectToSimpleString( field_object );
5756                         if( field_object && JSON_NUMBER == field_object->type )
5757                                 value_is_numeric = 1;
5758                 }
5759
5760                 osrfLogDebug( OSRF_LOG_MARK, "Updating %s object with %s = %s",
5761                                 osrfHashGet(meta, "fieldmapper"), field_name, value);
5762
5763                 if (!field_object || field_object->type == JSON_NULL) {
5764                         if ( !(!( strcmp( osrfHashGet(meta, "classname"), "au" ) )
5765                                         && !( strcmp( field_name, "passwd" ) )) ) { // arg at the special case!
5766                                 if( first )
5767                                         first = 0;
5768                                 else
5769                                         OSRF_BUFFER_ADD_CHAR(sql, ',');
5770                                 buffer_fadd( sql, " %s = NULL", field_name );
5771                         }
5772
5773                 } else if ( value_is_numeric || !strcmp( get_primitive( field_def ), "number") ) {
5774                         if( first )
5775                                 first = 0;
5776                         else
5777                                 OSRF_BUFFER_ADD_CHAR(sql, ',');
5778
5779                         const char* numtype = get_datatype( field_def );
5780                         if ( !strncmp( numtype, "INT", 3 ) ) {
5781                                 buffer_fadd( sql, " %s = %ld", field_name, atol(value) );
5782                         } else if ( !strcmp( numtype, "NUMERIC" ) ) {
5783                                 buffer_fadd( sql, " %s = %f", field_name, atof(value) );
5784                         } else {
5785                                 // Must really be intended as a string, so quote it
5786                                 if ( dbi_conn_quote_string(dbhandle, &value) ) {
5787                                         buffer_fadd( sql, " %s = %s", field_name, value );
5788                                 } else {
5789                                         osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]",
5790                                                 MODULENAME, value);
5791                                         osrfAppSessionStatus(
5792                                                 ctx->session,
5793                                                 OSRF_STATUS_INTERNALSERVERERROR,
5794                                                 "osrfMethodException",
5795                                                 ctx->request,
5796                                                 "Error quoting string -- please see the error log for more details"
5797                                         );
5798                                         free(value);
5799                                         free(id);
5800                                         osrfHashIteratorFree( field_itr );
5801                                         buffer_free(sql);
5802                                         osrfAppRespondComplete( ctx, NULL );
5803                                         return -1;
5804                                 }
5805                         }
5806
5807                         osrfLogDebug( OSRF_LOG_MARK, "%s is of type %s", field_name, numtype );
5808
5809                 } else {
5810                         if ( dbi_conn_quote_string(dbhandle, &value) ) {
5811                                 if(first)
5812                                         first = 0;
5813                                 else
5814                                         OSRF_BUFFER_ADD_CHAR( sql, ',' );
5815                                 buffer_fadd( sql, " %s = %s", field_name, value );
5816                         } else {
5817                                 osrfLogError(OSRF_LOG_MARK, "%s: Error quoting string [%s]", MODULENAME, value);
5818                                 osrfAppSessionStatus(
5819                                         ctx->session,
5820                                         OSRF_STATUS_INTERNALSERVERERROR,
5821                                         "osrfMethodException",
5822                                         ctx->request,
5823                                         "Error quoting string -- please see the error log for more details"
5824                                 );
5825                                 free(value);
5826                                 free(id);
5827                                 osrfHashIteratorFree( field_itr );
5828                                 buffer_free(sql);
5829                                 osrfAppRespondComplete( ctx, NULL );
5830                                 return -1;
5831                         }
5832                 }
5833
5834                 free(value);
5835
5836         } // end while
5837
5838         osrfHashIteratorFree( field_itr );
5839
5840         jsonObject* obj = jsonNewObject(id);
5841
5842         if ( strcmp( get_primitive( osrfHashGet( osrfHashGet(meta, "fields"), pkey ) ), "number" ) )
5843                 dbi_conn_quote_string(dbhandle, &id);
5844
5845         buffer_fadd( sql, " WHERE %s = %s;", pkey, id );
5846
5847         char* query = buffer_release(sql);
5848         osrfLogDebug(OSRF_LOG_MARK, "%s: Update SQL [%s]", MODULENAME, query);
5849
5850         dbi_result result = dbi_conn_query(dbhandle, query);
5851         free(query);
5852
5853         if (!result) {
5854                 jsonObjectFree(obj);
5855                 obj = jsonNewObject(NULL);
5856                 osrfLogError(
5857                         OSRF_LOG_MARK,
5858                         "%s ERROR updating %s object with %s = %s",
5859                         MODULENAME,
5860                         osrfHashGet(meta, "fieldmapper"),
5861                         pkey,
5862                         id
5863                 );
5864         }
5865
5866         free(id);
5867         osrfAppRespondComplete( ctx, obj );
5868         jsonObjectFree( obj );
5869         return 0;
5870 }
5871
5872 static int doDelete( osrfMethodContext* ctx ) {
5873         if(osrfMethodVerifyContext( ctx )) {
5874                 osrfLogError( OSRF_LOG_MARK, "Invalid method context" );
5875                 return -1;
5876         }
5877
5878         if( enforce_pcrud )
5879                 timeout_needs_resetting = 1;
5880
5881         osrfHash* meta = osrfHashGet( (osrfHash*) ctx->method->userData, "class" );
5882
5883         if( getXactId( ctx ) == NULL ) {
5884                 osrfAppSessionStatus(
5885                         ctx->session,
5886                         OSRF_STATUS_BADREQUEST,
5887                         "osrfMethodException",
5888                         ctx->request,
5889                         "No active transaction -- required for DELETE"
5890                 );
5891                 osrfAppRespondComplete( ctx, NULL );
5892                 return -1;
5893         }
5894
5895         // The following test is harmless but redundant.  If a class is
5896         // readonly, we don't register a delete method for it.
5897         if( str_is_true( osrfHashGet( meta, "readonly" ) ) ) {
5898                 osrfAppSessionStatus(
5899                         ctx->session,
5900                         OSRF_STATUS_BADREQUEST,
5901                         "osrfMethodException",
5902                         ctx->request,
5903                         "Cannot DELETE readonly class"
5904                 );
5905                 osrfAppRespondComplete( ctx, NULL );
5906                 return -1;
5907         }
5908
5909         dbhandle = writehandle;
5910
5911         char* pkey = osrfHashGet(meta, "primarykey");
5912
5913         int _obj_pos = 0;
5914         if( enforce_pcrud )
5915                 _obj_pos = 1;
5916
5917         char* id;
5918         if (jsonObjectGetIndex(ctx->params, _obj_pos)->classname) {
5919                 if (!verifyObjectClass(ctx, jsonObjectGetIndex( ctx->params, _obj_pos ))) {
5920                         osrfAppRespondComplete( ctx, NULL );
5921                         return -1;
5922                 }
5923
5924                 id = oilsFMGetString( jsonObjectGetIndex(ctx->params, _obj_pos), pkey );
5925         } else {
5926                 if( enforce_pcrud && !verifyObjectPCRUD( ctx, NULL )) {
5927                         osrfAppRespondComplete( ctx, NULL );
5928                         return -1;
5929                 }
5930                 id = jsonObjectToSimpleString(jsonObjectGetIndex(ctx->params, _obj_pos));
5931         }
5932
5933         osrfLogDebug(
5934                 OSRF_LOG_MARK,
5935                 "%s deleting %s object with %s = %s",
5936                 MODULENAME,
5937                 osrfHashGet(meta, "fieldmapper"),
5938                 pkey,
5939                 id
5940         );
5941
5942         jsonObject* obj = jsonNewObject(id);
5943
5944         if ( strcmp( get_primitive( osrfHashGet( osrfHashGet(meta, "fields"), pkey ) ), "number" ) )
5945                 dbi_conn_quote_string(writehandle, &id);
5946
5947         dbi_result result = dbi_conn_queryf(writehandle, "DELETE FROM %s WHERE %s = %s;",
5948                 osrfHashGet(meta, "tablename"), pkey, id);
5949
5950         if (!result) {
5951                 jsonObjectFree(obj);
5952                 obj = jsonNewObject(NULL);
5953                 osrfLogError(
5954                         OSRF_LOG_MARK,
5955                         "%s ERROR deleting %s object with %s = %s",
5956                         MODULENAME,
5957                         osrfHashGet(meta, "fieldmapper"),
5958                         pkey,
5959                         id
5960                 );
5961         }
5962
5963         free(id);
5964
5965         osrfAppRespondComplete( ctx, obj );
5966         jsonObjectFree( obj );
5967         return 0;
5968 }
5969
5970 /**
5971         @brief Translate a row returned from the database into a jsonObject of type JSON_ARRAY.
5972         @param result An iterator for a result set; we only look at the current row.
5973         @param @meta Pointer to the class metadata for the core class.
5974         @return Pointer to the resulting jsonObject if successful; otherwise NULL.
5975
5976         If a column is not defined in the IDL, or if it has no array_position defined for it in
5977         the IDL, or if it is defined as virtual, ignore it.
5978
5979         Otherwise, translate the column value into a jsonObject of type JSON_NULL, JSON_NUMBER,
5980         or JSON_STRING.  Then insert this jsonObject into the JSON_ARRAY according to its
5981         array_position in the IDL.
5982
5983         A field defined in the IDL but not represented in the returned row will leave a hole
5984         in the JSON_ARRAY.  In effect it will be treated as a null value.
5985
5986         In the resulting JSON_ARRAY, the field values appear in the sequence defined by the IDL,
5987         regardless of their sequence in the SELECT statement.  The JSON_ARRAY is assigned the
5988         classname corresponding to the @a meta argument.
5989
5990         The calling code is responsible for freeing the the resulting jsonObject by calling
5991         jsonObjectFree().
5992 */
5993 static jsonObject* oilsMakeFieldmapperFromResult( dbi_result result, osrfHash* meta) {
5994         if(!(result && meta)) return jsonNULL;
5995
5996         jsonObject* object = jsonNewObjectType( JSON_ARRAY );
5997         jsonObjectSetClass(object, osrfHashGet(meta, "classname"));
5998         osrfLogInternal(OSRF_LOG_MARK, "Setting object class to %s ", object->classname);
5999
6000         osrfHash* fields = osrfHashGet(meta, "fields");
6001
6002         int columnIndex = 1;
6003         const char* columnName;
6004
6005         /* cycle through the columns in the row returned from the database */
6006         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
6007
6008                 osrfLogInternal(OSRF_LOG_MARK, "Looking for column named [%s]...", (char*)columnName);
6009
6010                 int fmIndex = -1;  // Will be set to the IDL's sequence number for this field
6011
6012                 /* determine the field type and storage attributes */
6013                 unsigned short type = dbi_result_get_field_type_idx(result, columnIndex);
6014                 int attr            = dbi_result_get_field_attribs_idx(result, columnIndex);
6015
6016                 // Fetch the IDL's sequence number for the field.  If the field isn't in the IDL,
6017                 // or if it has no sequence number there, or if it's virtual, skip it.
6018                 osrfHash* _f = osrfHashGet( fields, (char*) columnName );
6019                 if( _f ) {
6020
6021                         if ( str_is_true( osrfHashGet(_f, "virtual") ) )
6022                                 continue;   // skip this column: IDL says it's virtual
6023
6024                         const char* pos = (char*)osrfHashGet(_f, "array_position");
6025                         if ( !pos )      // IDL has no sequence number for it.  This shouldn't happen,
6026                                 continue;    // since we assign sequence numbers dynamically as we load the IDL.
6027
6028                         fmIndex = atoi( pos );
6029                         osrfLogInternal(OSRF_LOG_MARK, "... Found column at position [%s]...", pos);
6030                 } else {
6031                         continue;     // This field is not defined in the IDL
6032                 }
6033
6034                 // Stuff the column value into a slot in the JSON_ARRAY, indexed according to the
6035                 // sequence number from the IDL (which is likely to be different from the sequence
6036                 // of columns in the SELECT clause).
6037                 if (dbi_result_field_is_null_idx(result, columnIndex)) {
6038                         jsonObjectSetIndex( object, fmIndex, jsonNewObject(NULL) );
6039                 } else {
6040
6041                         switch( type ) {
6042
6043                                 case DBI_TYPE_INTEGER :
6044
6045                                         if( attr & DBI_INTEGER_SIZE8 )
6046                                                 jsonObjectSetIndex( object, fmIndex,
6047                                                         jsonNewNumberObject(dbi_result_get_longlong_idx(result, columnIndex)));
6048                                         else
6049                                                 jsonObjectSetIndex( object, fmIndex,
6050                                                         jsonNewNumberObject(dbi_result_get_int_idx(result, columnIndex)));
6051
6052                                         break;
6053
6054                                 case DBI_TYPE_DECIMAL :
6055                                         jsonObjectSetIndex( object, fmIndex,
6056                                                         jsonNewNumberObject(dbi_result_get_double_idx(result, columnIndex)));
6057                                         break;
6058
6059                                 case DBI_TYPE_STRING :
6060
6061                                         jsonObjectSetIndex(
6062                                                 object,
6063                                                 fmIndex,
6064                                                 jsonNewObject( dbi_result_get_string_idx(result, columnIndex) )
6065                                         );
6066
6067                                         break;
6068
6069                                 case DBI_TYPE_DATETIME : {
6070
6071                                         char dt_string[256] = "";
6072                                         struct tm gmdt;
6073
6074                                         // Fetch the date column as a time_t
6075                                         time_t _tmp_dt = dbi_result_get_datetime_idx(result, columnIndex);
6076
6077                                         // Translate the time_t to a human-readable string
6078                                         if (!(attr & DBI_DATETIME_DATE)) {
6079                                                 gmtime_r( &_tmp_dt, &gmdt );
6080                                                 strftime(dt_string, sizeof(dt_string), "%T", &gmdt);
6081                                         } else if (!(attr & DBI_DATETIME_TIME)) {
6082                                                 localtime_r( &_tmp_dt, &gmdt );
6083                                                 strftime(dt_string, sizeof(dt_string), "%F", &gmdt);
6084                                         } else {
6085                                                 localtime_r( &_tmp_dt, &gmdt );
6086                                                 strftime(dt_string, sizeof(dt_string), "%FT%T%z", &gmdt);
6087                                         }
6088
6089                                         jsonObjectSetIndex( object, fmIndex, jsonNewObject(dt_string) );
6090
6091                                         break;
6092                                 }
6093                                 case DBI_TYPE_BINARY :
6094                                         osrfLogError( OSRF_LOG_MARK,
6095                                                 "Can't do binary at column %s : index %d", columnName, columnIndex);
6096                         } // End switch
6097                 }
6098                 ++columnIndex;
6099         } // End while
6100
6101         return object;
6102 }
6103
6104 static jsonObject* oilsMakeJSONFromResult( dbi_result result ) {
6105         if(!result) return jsonNULL;
6106
6107         jsonObject* object = jsonNewObject(NULL);
6108
6109         time_t _tmp_dt;
6110         char dt_string[256];
6111         struct tm gmdt;
6112
6113         int fmIndex;
6114         int columnIndex = 1;
6115         int attr;
6116         unsigned short type;
6117         const char* columnName;
6118
6119         /* cycle through the column list */
6120         while( (columnName = dbi_result_get_field_name(result, columnIndex)) ) {
6121
6122                 osrfLogInternal(OSRF_LOG_MARK, "Looking for column named [%s]...", (char*)columnName);
6123
6124                 fmIndex = -1; // reset the position
6125
6126                 /* determine the field type and storage attributes */
6127                 type = dbi_result_get_field_type_idx(result, columnIndex);
6128                 attr = dbi_result_get_field_attribs_idx(result, columnIndex);
6129
6130                 if (dbi_result_field_is_null_idx(result, columnIndex)) {
6131                         jsonObjectSetKey( object, columnName, jsonNewObject(NULL) );
6132                 } else {
6133
6134                         switch( type ) {
6135
6136                                 case DBI_TYPE_INTEGER :
6137
6138                                         if( attr & DBI_INTEGER_SIZE8 )
6139                                                 jsonObjectSetKey( object, columnName,
6140                                                                 jsonNewNumberObject(dbi_result_get_longlong_idx(
6141                                                                                 result, columnIndex)) );
6142                                         else
6143                                                 jsonObjectSetKey( object, columnName,
6144                                                                 jsonNewNumberObject(dbi_result_get_int_idx(result, columnIndex)) );
6145                                         break;
6146
6147                                 case DBI_TYPE_DECIMAL :
6148                                         jsonObjectSetKey( object, columnName,
6149                                                         jsonNewNumberObject(dbi_result_get_double_idx(result, columnIndex)) );
6150                                         break;
6151
6152                                 case DBI_TYPE_STRING :
6153                                         jsonObjectSetKey( object, columnName,
6154                                                         jsonNewObject(dbi_result_get_string_idx(result, columnIndex)) );
6155                                         break;
6156
6157                                 case DBI_TYPE_DATETIME :
6158
6159                                         memset(dt_string, '\0', sizeof(dt_string));
6160                                         memset(&gmdt, '\0', sizeof(gmdt));
6161
6162                                         _tmp_dt = dbi_result_get_datetime_idx(result, columnIndex);
6163
6164
6165                                         if (!(attr & DBI_DATETIME_DATE)) {
6166                                                 gmtime_r( &_tmp_dt, &gmdt );
6167                                                 strftime(dt_string, sizeof(dt_string), "%T", &gmdt);
6168                                         } else if (!(attr & DBI_DATETIME_TIME)) {
6169                                                 localtime_r( &_tmp_dt, &gmdt );
6170                                                 strftime(dt_string, sizeof(dt_string), "%F", &gmdt);
6171                                         } else {
6172                                                 localtime_r( &_tmp_dt, &gmdt );
6173                                                 strftime(dt_string, sizeof(dt_string), "%FT%T%z", &gmdt);
6174                                         }
6175
6176                                         jsonObjectSetKey( object, columnName, jsonNewObject(dt_string) );
6177                                         break;
6178
6179                                 case DBI_TYPE_BINARY :
6180                                         osrfLogError( OSRF_LOG_MARK,
6181                                                 "Can't do binary at column %s : index %d", columnName, columnIndex );
6182                         }
6183                 }
6184                 ++columnIndex;
6185         } // end while loop traversing result
6186
6187         return object;
6188 }
6189
6190 // Interpret a string as true or false
6191 static int str_is_true( const char* str ) {
6192         if( NULL == str || strcasecmp( str, "true" ) )
6193                 return 0;
6194         else
6195                 return 1;
6196 }
6197
6198 // Interpret a jsonObject as true or false
6199 static int obj_is_true( const jsonObject* obj ) {
6200         if( !obj )
6201                 return 0;
6202         else switch( obj->type )
6203         {
6204                 case JSON_BOOL :
6205                         if( obj->value.b )
6206                                 return 1;
6207                         else
6208                                 return 0;
6209                 case JSON_STRING :
6210                         if( strcasecmp( obj->value.s, "true" ) )
6211                                 return 0;
6212                         else
6213                                 return 1;
6214                 case JSON_NUMBER :          // Support 1/0 for perl's sake
6215                         if( jsonObjectGetNumber( obj ) == 1.0 )
6216                                 return 1;
6217                         else
6218                                 return 0;
6219                 default :
6220                         return 0;
6221         }
6222 }
6223
6224 // Translate a numeric code into a text string identifying a type of
6225 // jsonObject.  To be used for building error messages.
6226 static const char* json_type( int code ) {
6227         switch ( code )
6228         {
6229                 case 0 :
6230                         return "JSON_HASH";
6231                 case 1 :
6232                         return "JSON_ARRAY";
6233                 case 2 :
6234                         return "JSON_STRING";
6235                 case 3 :
6236                         return "JSON_NUMBER";
6237                 case 4 :
6238                         return "JSON_NULL";
6239                 case 5 :
6240                         return "JSON_BOOL";
6241                 default :
6242                         return "(unrecognized)";
6243         }
6244 }
6245
6246 // Extract the "primitive" attribute from an IDL field definition.
6247 // If we haven't initialized the app, then we must be running in
6248 // some kind of testbed.  In that case, default to "string".
6249 static const char* get_primitive( osrfHash* field ) {
6250         const char* s = osrfHashGet( field, "primitive" );
6251         if( !s ) {
6252                 if( child_initialized )
6253                         osrfLogError(
6254                                 OSRF_LOG_MARK,
6255                                 "%s ERROR No \"datatype\" attribute for field \"%s\"",
6256                                 MODULENAME,
6257                                 osrfHashGet( field, "name" )
6258                         );
6259
6260                 s = "string";
6261         }
6262         return s;
6263 }
6264
6265 // Extract the "datatype" attribute from an IDL field definition.
6266 // If we haven't initialized the app, then we must be running in
6267 // some kind of testbed.  In that case, default to to NUMERIC,
6268 // since we look at the datatype only for numbers.
6269 static const char* get_datatype( osrfHash* field ) {
6270         const char* s = osrfHashGet( field, "datatype" );
6271         if( !s ) {
6272                 if( child_initialized )
6273                         osrfLogError(
6274                                 OSRF_LOG_MARK,
6275                                 "%s ERROR No \"datatype\" attribute for field \"%s\"",
6276                                 MODULENAME,
6277                                 osrfHashGet( field, "name" )
6278                         );
6279                 else
6280                         s = "NUMERIC";
6281         }
6282         return s;
6283 }
6284
6285 /**
6286         @brief Determine whether a string is potentially a valid SQL identifier.
6287         @param s The identifier to be tested.
6288         @return 1 if the input string is potentially a valid SQL identifier, or 0 if not.
6289
6290         Purpose: to prevent certain kinds of SQL injection.  To that end we don't necessarily
6291         need to follow all the rules exactly, such as requiring that the first character not
6292         be a digit.
6293
6294         We allow leading and trailing white space.  In between, we do not allow punctuation
6295         (except for underscores and dollar signs), control characters, or embedded white space.
6296
6297         More pedantically we should allow quoted identifiers containing arbitrary characters, but
6298         for the foreseeable future such quoted identifiers are not likely to be an issue.
6299 */
6300 static int is_identifier( const char* s) {
6301         if( !s )
6302                 return 0;
6303
6304         // Skip leading white space
6305         while( isspace( (unsigned char) *s ) )
6306                 ++s;
6307
6308         if( !s )
6309                 return 0;   // Nothing but white space?  Not okay.
6310
6311         // Check each character until we reach white space or
6312         // end-of-string.  Letters, digits, underscores, and
6313         // dollar signs are okay. With the exception of periods
6314         // (as in schema.identifier), control characters and other
6315         // punctuation characters are not okay.  Anything else
6316         // is okay -- it could for example be part of a multibyte
6317         // UTF8 character such as a letter with diacritical marks,
6318         // and those are allowed.
6319         do {
6320                 if( isalnum( (unsigned char) *s )
6321                         || '.' == *s
6322                         || '_' == *s
6323                         || '$' == *s )
6324                         ;  // Fine; keep going
6325                 else if(   ispunct( (unsigned char) *s )
6326                                 || iscntrl( (unsigned char) *s ) )
6327                         return 0;
6328                         ++s;
6329         } while( *s && ! isspace( (unsigned char) *s ) );
6330
6331         // If we found any white space in the above loop,
6332         // the rest had better be all white space.
6333
6334         while( isspace( (unsigned char) *s ) )
6335                 ++s;
6336
6337         if( *s )
6338                 return 0;   // White space was embedded within non-white space
6339
6340         return 1;
6341 }
6342
6343 /**
6344         @brief Determine whether to accept a character string as a comparison operator.
6345         @param op The candidate comparison operator.
6346         @return 1 if the string is acceptable as a comparison operator, or 0 if not.
6347
6348         We don't validate the operator for real.  We just make sure that it doesn't contain
6349         any semicolons or white space (with special exceptions for a few specific operators).
6350         The idea is to block certain kinds of SQL injection.  If it has no semicolons or white
6351         space but it's still not a valid operator, then the database will complain.
6352
6353         Another approach would be to compare the string against a short list of approved operators.
6354         We don't do that because we want to allow custom operators like ">100*", which at this
6355         writing would be difficult or impossible to express otherwise in a JSON query.
6356 */
6357 static int is_good_operator( const char* op ) {
6358         if( !op ) return 0;   // Sanity check
6359
6360         const char* s = op;
6361         while( *s ) {
6362                 if( isspace( (unsigned char) *s ) ) {
6363                         // Special exceptions for SIMILAR TO, IS DISTINCT FROM,
6364                         // and IS NOT DISTINCT FROM.
6365                         if( !strcasecmp( op, "similar to" ) )
6366                                 return 1;
6367                         else if( !strcasecmp( op, "is distinct from" ) )
6368                                 return 1;
6369                         else if( !strcasecmp( op, "is not distinct from" ) )
6370                                 return 1;
6371                         else
6372                                 return 0;
6373                 }
6374                 else if( ';' == *s )
6375                         return 0;
6376                 ++s;
6377         }
6378         return 1;
6379 }
6380
6381 /**
6382         @name Query Frame Management
6383
6384         The following machinery supports a stack of query frames for use by SELECT().
6385
6386         A query frame caches information about one level of a SELECT query.  When we enter
6387         a subquery, we push another query frame onto the stack, and pop it off when we leave.
6388
6389         The query frame stores information about the core class, and about any joined classes
6390         in the FROM clause.
6391
6392         The main purpose is to map table aliases to classes and tables, so that a query can
6393         join to the same table more than once.  A secondary goal is to reduce the number of
6394         lookups in the IDL by caching the results.
6395 */
6396 /*@{*/
6397
6398 #define STATIC_CLASS_INFO_COUNT 3
6399
6400 static ClassInfo static_class_info[ STATIC_CLASS_INFO_COUNT ];
6401
6402 /**
6403         @brief Allocate a ClassInfo as raw memory.
6404         @return Pointer to the newly allocated ClassInfo.
6405
6406         Except for the in_use flag, which is used only by the allocation and deallocation
6407         logic, we don't initialize the ClassInfo here.
6408 */
6409 static ClassInfo* allocate_class_info( void ) {
6410         // In order to reduce the number of mallocs and frees, we return a static
6411         // instance of ClassInfo, if we can find one that we're not already using.
6412         // We rely on the fact that the compiler will implicitly initialize the
6413         // static instances so that in_use == 0.
6414
6415         int i;
6416         for( i = 0; i < STATIC_CLASS_INFO_COUNT; ++i ) {
6417                 if( ! static_class_info[ i ].in_use ) {
6418                         static_class_info[ i ].in_use = 1;
6419                         return static_class_info + i;
6420                 }
6421         }
6422
6423         // The static ones are all in use.  Malloc one.
6424
6425         return safe_malloc( sizeof( ClassInfo ) );
6426 }
6427
6428 /**
6429         @brief Free any malloc'd memory owned by a ClassInfo, returning it to a pristine state.
6430         @param info Pointer to the ClassInfo to be cleared.
6431 */
6432 static void clear_class_info( ClassInfo* info ) {
6433         // Sanity check
6434         if( ! info )
6435                 return;
6436
6437         // Free any malloc'd strings
6438
6439         if( info->alias != info->alias_store )
6440                 free( info->alias );
6441
6442         if( info->class_name != info->class_name_store )
6443                 free( info->class_name );
6444
6445         free( info->source_def );
6446
6447         info->alias = info->class_name = info->source_def = NULL;
6448         info->next = NULL;
6449 }
6450
6451 /**
6452         @brief Free a ClassInfo and everything it owns.
6453         @param info Pointer to the ClassInfo to be freed.
6454 */
6455 static void free_class_info( ClassInfo* info ) {
6456         // Sanity check
6457         if( ! info )
6458                 return;
6459
6460         clear_class_info( info );
6461
6462         // If it's one of the static instances, just mark it as not in use
6463
6464         int i;
6465         for( i = 0; i < STATIC_CLASS_INFO_COUNT; ++i ) {
6466                 if( info == static_class_info + i ) {
6467                         static_class_info[ i ].in_use = 0;
6468                         return;
6469                 }
6470         }
6471
6472         // Otherwise it must have been malloc'd, so free it
6473
6474         free( info );
6475 }
6476
6477 /**
6478         @brief Populate an already-allocated ClassInfo.
6479         @param info Pointer to the ClassInfo to be populated.
6480         @param alias Alias for the class.  If it is NULL, or an empty string, use the class
6481         name for an alias.
6482         @param class Name of the class.
6483         @return Zero if successful, or 1 if not.
6484
6485         Populate the ClassInfo with copies of the alias and class name, and with pointers to
6486         the relevant portions of the IDL for the specified class.
6487 */
6488 static int build_class_info( ClassInfo* info, const char* alias, const char* class ) {
6489         // Sanity checks
6490         if( ! info ){
6491                 osrfLogError( OSRF_LOG_MARK,
6492                                           "%s ERROR: No ClassInfo available to populate", MODULENAME );
6493                 info->alias = info->class_name = info->source_def = NULL;
6494                 info->class_def = info->fields = info->links = NULL;
6495                 return 1;
6496         }
6497
6498         if( ! class ) {
6499                 osrfLogError( OSRF_LOG_MARK,
6500                                           "%s ERROR: No class name provided for lookup", MODULENAME );
6501                 info->alias = info->class_name = info->source_def = NULL;
6502                 info->class_def = info->fields = info->links = NULL;
6503                 return 1;
6504         }
6505
6506         // Alias defaults to class name if not supplied
6507         if( ! alias || ! alias[ 0 ] )
6508                 alias = class;
6509
6510         // Look up class info in the IDL
6511         osrfHash* class_def = osrfHashGet( oilsIDL(), class );
6512         if( ! class_def ) {
6513                 osrfLogError( OSRF_LOG_MARK,
6514                                           "%s ERROR: Class %s not defined in IDL", MODULENAME, class );
6515                 info->alias = info->class_name = info->source_def = NULL;
6516                 info->class_def = info->fields = info->links = NULL;
6517                 return 1;
6518         } else if( str_is_true( osrfHashGet( class_def, "virtual" ) ) ) {
6519                 osrfLogError( OSRF_LOG_MARK,
6520                                           "%s ERROR: Class %s is defined as virtual", MODULENAME, class );
6521                 info->alias = info->class_name = info->source_def = NULL;
6522                 info->class_def = info->fields = info->links = NULL;
6523                 return 1;
6524         }
6525
6526         osrfHash* links = osrfHashGet( class_def, "links" );
6527         if( ! links ) {
6528                 osrfLogError( OSRF_LOG_MARK,
6529                                           "%s ERROR: No links defined in IDL for class %s", MODULENAME, class );
6530                 info->alias = info->class_name = info->source_def = NULL;
6531                 info->class_def = info->fields = info->links = NULL;
6532                 return 1;
6533         }
6534
6535         osrfHash* fields = osrfHashGet( class_def, "fields" );
6536         if( ! fields ) {
6537                 osrfLogError( OSRF_LOG_MARK,
6538                                           "%s ERROR: No fields defined in IDL for class %s", MODULENAME, class );
6539                 info->alias = info->class_name = info->source_def = NULL;
6540                 info->class_def = info->fields = info->links = NULL;
6541                 return 1;
6542         }
6543
6544         char* source_def = getRelation( class_def );
6545         if( ! source_def )
6546                 return 1;
6547
6548         // We got everything we need, so populate the ClassInfo
6549         if( strlen( alias ) > ALIAS_STORE_SIZE )
6550                 info->alias = strdup( alias );
6551         else {
6552                 strcpy( info->alias_store, alias );
6553                 info->alias = info->alias_store;
6554         }
6555
6556         if( strlen( class ) > CLASS_NAME_STORE_SIZE )
6557                 info->class_name = strdup( class );
6558         else {
6559                 strcpy( info->class_name_store, class );
6560                 info->class_name = info->class_name_store;
6561         }
6562
6563         info->source_def = source_def;
6564
6565         info->class_def = class_def;
6566         info->links     = links;
6567         info->fields    = fields;
6568
6569         return 0;
6570 }
6571
6572 #define STATIC_FRAME_COUNT 3
6573
6574 static QueryFrame static_frame[ STATIC_FRAME_COUNT ];
6575
6576 /**
6577         @brief Allocate a QueryFrame as raw memory.
6578         @return Pointer to the newly allocated QueryFrame.
6579
6580         Except for the in_use flag, which is used only by the allocation and deallocation
6581         logic, we don't initialize the QueryFrame here.
6582 */
6583 static QueryFrame* allocate_frame( void ) {
6584         // In order to reduce the number of mallocs and frees, we return a static
6585         // instance of QueryFrame, if we can find one that we're not already using.
6586         // We rely on the fact that the compiler will implicitly initialize the
6587         // static instances so that in_use == 0.
6588
6589         int i;
6590         for( i = 0; i < STATIC_FRAME_COUNT; ++i ) {
6591                 if( ! static_frame[ i ].in_use ) {
6592                         static_frame[ i ].in_use = 1;
6593                         return static_frame + i;
6594                 }
6595         }
6596
6597         // The static ones are all in use.  Malloc one.
6598
6599         return safe_malloc( sizeof( QueryFrame ) );
6600 }
6601
6602 /**
6603         @brief Free a QueryFrame, and all the memory it owns.
6604         @param frame Pointer to the QueryFrame to be freed.
6605 */
6606 static void free_query_frame( QueryFrame* frame ) {
6607         // Sanity check
6608         if( ! frame )
6609                 return;
6610
6611         clear_class_info( &frame->core );
6612
6613         // Free the join list
6614         ClassInfo* temp;
6615         ClassInfo* info = frame->join_list;
6616         while( info ) {
6617                 temp = info->next;
6618                 free_class_info( info );
6619                 info = temp;
6620         }
6621
6622         frame->join_list = NULL;
6623         frame->next = NULL;
6624
6625         // If the frame is a static instance, just mark it as unused
6626         int i;
6627         for( i = 0; i < STATIC_FRAME_COUNT; ++i ) {
6628                 if( frame == static_frame + i ) {
6629                         static_frame[ i ].in_use = 0;
6630                         return;
6631                 }
6632         }
6633
6634         // Otherwise it must have been malloc'd, so free it
6635
6636         free( frame );
6637 }
6638
6639 /**
6640         @brief Search a given QueryFrame for a specified alias.
6641         @param frame Pointer to the QueryFrame to be searched.
6642         @param target The alias for which to search.
6643         @return Pointer to the ClassInfo for the specified alias, if found; otherwise NULL.
6644 */
6645 static ClassInfo* search_alias_in_frame( QueryFrame* frame, const char* target ) {
6646         if( ! frame || ! target ) {
6647                 return NULL;
6648         }
6649
6650         ClassInfo* found_class = NULL;
6651
6652         if( !strcmp( target, frame->core.alias ) )
6653                 return &(frame->core);
6654         else {
6655                 ClassInfo* curr_class = frame->join_list;
6656                 while( curr_class ) {
6657                         if( strcmp( target, curr_class->alias ) )
6658                                 curr_class = curr_class->next;
6659                         else {
6660                                 found_class = curr_class;
6661                                 break;
6662                         }
6663                 }
6664         }
6665
6666         return found_class;
6667 }
6668
6669 /**
6670         @brief Push a new (blank) QueryFrame onto the stack.
6671 */
6672 static void push_query_frame( void ) {
6673         QueryFrame* frame = allocate_frame();
6674         frame->join_list = NULL;
6675         frame->next = curr_query;
6676
6677         // Initialize the ClassInfo for the core class
6678         ClassInfo* core = &frame->core;
6679         core->alias = core->class_name = core->source_def = NULL;
6680         core->class_def = core->fields = core->links = NULL;
6681
6682         curr_query = frame;
6683 }
6684
6685 /**
6686         @brief Pop a QueryFrame off the stack and destroy it.
6687 */
6688 static void pop_query_frame( void ) {
6689         // Sanity check
6690         if( ! curr_query )
6691                 return;
6692
6693         QueryFrame* popped = curr_query;
6694         curr_query = popped->next;
6695
6696         free_query_frame( popped );
6697 }
6698
6699 /**
6700         @brief Populate the ClassInfo for the core class.
6701         @param alias Alias for the core class.  If it is NULL or an empty string, we use the
6702         class name as an alias.
6703         @param class_name Name of the core class.
6704         @return Zero if successful, or 1 if not.
6705
6706         Populate the ClassInfo of the core class with copies of the alias and class name, and
6707         with pointers to the relevant portions of the IDL for the core class.
6708 */
6709 static int add_query_core( const char* alias, const char* class_name ) {
6710
6711         // Sanity checks
6712         if( ! curr_query ) {
6713                 osrfLogError( OSRF_LOG_MARK,
6714                                           "%s ERROR: No QueryFrame available for class %s", MODULENAME, class_name );
6715                 return 1;
6716         } else if( curr_query->core.alias ) {
6717                 osrfLogError( OSRF_LOG_MARK,
6718                                           "%s ERROR: Core class %s already populated as %s",
6719                                           MODULENAME, curr_query->core.class_name, curr_query->core.alias );
6720                 return 1;
6721         }
6722
6723         build_class_info( &curr_query->core, alias, class_name );
6724         if( curr_query->core.alias )
6725                 return 0;
6726         else {
6727                 osrfLogError( OSRF_LOG_MARK,
6728                                           "%s ERROR: Unable to look up core class %s", MODULENAME, class_name );
6729                 return 1;
6730         }
6731 }
6732
6733 /**
6734         @brief Search the current QueryFrame for a specified alias.
6735         @param target The alias for which to search.
6736         @return A pointer to the corresponding ClassInfo, if found; otherwise NULL.
6737 */
6738 static inline ClassInfo* search_alias( const char* target ) {
6739         return search_alias_in_frame( curr_query, target );
6740 }
6741
6742 /**
6743         @brief Search all levels of query for a specified alias, starting with the current query.
6744         @param target The alias for which to search.
6745         @return A pointer to the corresponding ClassInfo, if found; otherwise NULL.
6746 */
6747 static ClassInfo* search_all_alias( const char* target ) {
6748         ClassInfo* found_class = NULL;
6749         QueryFrame* curr_frame = curr_query;
6750
6751         while( curr_frame ) {
6752                 if(( found_class = search_alias_in_frame( curr_frame, target ) ))
6753                         break;
6754                 else
6755                         curr_frame = curr_frame->next;
6756         }
6757
6758         return found_class;
6759 }
6760
6761 /**
6762         @brief Add a class to the list of classes joined to the current query.
6763         @param alias Alias of the class to be added.  If it is NULL or an empty string, we use
6764         the class name as an alias.
6765         @param classname The name of the class to be added.
6766         @return A pointer to the ClassInfo for the added class, if successful; otherwise NULL.
6767 */
6768 static ClassInfo* add_joined_class( const char* alias, const char* classname ) {
6769
6770         if( ! classname || ! *classname ) {    // sanity check
6771                 osrfLogError( OSRF_LOG_MARK, "Can't join a class with no class name" );
6772                 return NULL;
6773         }
6774
6775         if( ! alias )
6776                 alias = classname;
6777
6778         const ClassInfo* conflict = search_alias( alias );
6779         if( conflict ) {
6780                 osrfLogError( OSRF_LOG_MARK,
6781                                           "%s ERROR: Table alias \"%s\" conflicts with class \"%s\"",
6782                                           MODULENAME, alias, conflict->class_name );
6783                 return NULL;
6784         }
6785
6786         ClassInfo* info = allocate_class_info();
6787
6788         if( build_class_info( info, alias, classname ) ) {
6789                 free_class_info( info );
6790                 return NULL;
6791         }
6792
6793         // Add the new ClassInfo to the join list of the current QueryFrame
6794         info->next = curr_query->join_list;
6795         curr_query->join_list = info;
6796
6797         return info;
6798 }
6799
6800 /**
6801         @brief Destroy all nodes on the query stack.
6802 */
6803 static void clear_query_stack( void ) {
6804         while( curr_query )
6805                 pop_query_frame();
6806 }
6807
6808 /*@}*/