text: /*
   * Ctable search routines
   *
   * $Id$
   *
   */
  
  #include "ctable.h"
  
  #include "ctable_qsort.c"
  
  #include "boyer_moore.c"
  
  #include "jsw_rand.c"
  
  #include "ctable_lists.c"
  
  #include "ctable_batch.c"
  
  #include "jsw_slib.c"
  
  #include "speedtableHash.c"
  
  #include <time.h>
  
  // forward references
  CTABLE_INTERNAL struct cursor *
  ctable_CreateEmptyCursor(Tcl_Interp *interp, CTable *ctable, char *name);
  CTABLE_INTERNAL struct cursor *
  ctable_CreateCursor(Tcl_Interp *interp, CTable *ctable, CTableSearch *search);
  CTABLE_INTERNAL int
  ctable_CreateCursorCommand(Tcl_Interp *interp, struct cursor *cursor);
  CTABLE_INTERNAL Tcl_Obj *
  ctable_CursorToName(struct cursor *cursor);
  
  //#define INDEXDEBUG
  // #define MEGADEBUG
  // #define SEARCHDEBUG
  #ifndef TRACKFIELD
  #define TRACKFIELD 1
  #endif
  // debugging routines - verify that every skiplist contains an entry for
  // every row.
  CTABLE_INTERNAL void
  ctable_verifyField(CTable *ctable, int field, int verbose)
  {
      ctable_BaseRow *row = NULL;
      ctable_BaseRow *found = NULL;
      ctable_BaseRow *walk = NULL;
      jsw_skip_t     *skip = ctable->skipLists[field];
      int             index = ctable->creator->fields[field]->indexNumber;
  
      if(verbose) fprintf(stderr, "Verifying field %s\n", ctable->creator->fields[field]->name);
      CTABLE_LIST_FOREACH (ctable->ll_head, row, 0) {
  	if (verbose) fprintf(stderr, "  Searching 0x%lx for 0x%lx\n", (long)skip, (long)row);
  	// make sure the entry can be found.
  	jsw_sreset(skip);
  	jsw_sfind (skip, row);
  	found = jsw_srow(skip);
  	if(found != row) {
  	    int count = 0;
  	    if(!verbose) {
  		fprintf(stderr, "Verifying field %s\n", ctable->creator->fields[field]->name);
  		fprintf(stderr, "  Searching skip 0x%lx for row 0x%lx\n", (long)skip, (long)row);
  	    }
  	    fprintf(stderr, "    Walking from 0x%lx\n", (long)found);
              // walk walkRow through the linked list of rows off this skip list
              CTABLE_LIST_FOREACH (found, walk, index) {
  		fprintf(stderr, "      ... 0x%lx\n", (long)walk);
  		count++;
  	        if(row == walk)
  		    break;
  	    }
  	    if(row != walk)
  	        Tcl_Panic("row 0x%lx not found in table", (long)row);
  	    fprintf(stderr, "    Found after %d links\n", count);
  	}
      }
      if(verbose) fprintf(stderr, "Field %s OK\n", ctable->creator->fields[field]->name);
  }
  
  CTABLE_INTERNAL void
  ctable_verify (Tcl_Interp *interp, CTable *ctable, int verbose) {
      int field;
  
      if(verbose) fprintf(stderr, "Verify start.\n");
      for(field = 0; field < ctable->creator->nFields; field++) {
  	if(ctable->creator->fields[field]->canBeNull == 0) {
  	    if(ctable->skipLists[field]) {
  	        ctable_verifyField(ctable, field, verbose);
  	    } else if(verbose) {
  		fprintf(stderr, "No index for field %s\n", ctable->creator->fields[field]->name);
  	    }
  	} else if(verbose) {
  	    fprintf(stderr, "Skipping nullable field %s\n", ctable->creator->fields[field]->name);
  	}
      }
      if(verbose) fprintf(stderr, "Verify end.\n");
  }
  
  
  /*
   * ctable_ParseFieldList - given a Tcl list object and a pointer to an array
   * of integer field numbers and a pointer to an integer for field counts,
   * install the field count into the field count and allocate an array of 
   * integers for the corresponding field indexes and fill that array with the 
   * field numbers corresponding to the field names in the list.
   *
   * It is up to the caller to free the memory pointed to through the
   * fieldList argument.
   *
   * return TCL_OK if all went according to plan, else TCL_ERROR.
   *
   */
  CTABLE_INTERNAL int
  ctable_ParseFieldList (Tcl_Interp *interp, Tcl_Obj *fieldListObj, CONST char **fieldNames, int **fieldListPtr, int *fieldCountPtr) {
      int             nFields;
      Tcl_Obj       **fieldsObjv;
      int             i;
      int            *fieldList;
  
      // the fields they want us to retrieve
      if (Tcl_ListObjGetElements (interp, fieldListObj, &nFields, &fieldsObjv) == TCL_ERROR) {
        return TCL_ERROR;
      }
  
      *fieldCountPtr = nFields;
      *fieldListPtr = fieldList = (int *)ckalloc (sizeof (int) * nFields);
  
      for (i = 0; i < nFields; i++) {
  	if (Tcl_GetIndexFromObj (interp, fieldsObjv[i], fieldNames, "field", TCL_EXACT, &fieldList[i]) != TCL_OK) {
  	    ckfree ((char *)fieldList);
  	    *fieldListPtr = NULL;
  	    return TCL_ERROR;
  	  }
      }
      return TCL_OK;
  }
  
  //
  // ctable_ParseSortFieldList - given a Tcl list object, and a pointer to a
  // ctable sort structure, store the number of fields in the list in the
  // sort structure's field count.  allocate an array of integers for the
  // field numbers and directions and store them into the sort structure passed.
  //
  // Strip the prepending dash of each field, if present, and do the lookup
  // and store the field number in the corresponding field number array.
  //
  // If the dash was present set the corresponding direction in the direction
  // array to -1 else set it to 1.
  //
  // It is up to the caller to free the memory pointed to through the
  // fieldList argument.
  //
  // return TCL_OK if all went according to plan, else TCL_ERROR.
  //
  //
  CTABLE_INTERNAL int
  ctable_ParseSortFieldList (Tcl_Interp *interp, Tcl_Obj *fieldListObj, CONST char **fieldNames, CTableSort *sort) {
      int             nFields;
      Tcl_Obj       **fieldsObjv;
      Tcl_Obj        *fieldNameObj;
      int             i;
      char           *fieldName;
  
      // the fields they want us to retrieve
      if (Tcl_ListObjGetElements (interp, fieldListObj, &nFields, &fieldsObjv) == TCL_ERROR) {
        return TCL_ERROR;
      }
  
      sort->nFields = nFields;
      sort->fields =  (int *)ckalloc (sizeof (int) * nFields);
      sort->directions =  (int *)ckalloc (sizeof (int) * nFields);
  
      for (i = 0; i < nFields; i++) {
  	int fieldNameNeedsFreeing = 0;
  
          fieldName = Tcl_GetString (fieldsObjv[i]);
  	if (fieldName[0] == '-') {
  	    sort->directions[i] = -1;
  	    fieldName++;
  	    fieldNameObj = Tcl_NewStringObj (fieldName, -1);
  	    fieldNameNeedsFreeing = 1;
  	} else {
  	    fieldNameObj = fieldsObjv[i];
  	    sort->directions[i] = 1;
  	    fieldNameNeedsFreeing = 0;
  	}
  
  	if (Tcl_GetIndexFromObj (interp, fieldNameObj, fieldNames, "field", TCL_EXACT, &sort->fields[i]) != TCL_OK) {
  	    if (fieldNameNeedsFreeing) {
  		Tcl_DecrRefCount (fieldNameObj);
  	    }
  	    ckfree ((char *)sort->fields);
  	    ckfree ((char *)sort->directions);
  	    sort->fields = NULL;
  	    sort->directions = NULL;
  	    return TCL_ERROR;
  	}
  
  	if (fieldNameNeedsFreeing) {
  	    Tcl_DecrRefCount (fieldNameObj);
  	}
      }
      return TCL_OK;
  }
  
  //
  // ctable_searchMatchPatternCheck - examine the match pattern to determine
  // a strategy for matching the pattern.
  //
  // If it's anchored, i.e. doesn't start with an asterisk, that's good to know, 
  // we'll be real fast.
  //
  // If it's a pattern, we'll use more full blown pattern matching.
  //
  // If it's unanchored, we'll use Boyer-Moore to go as fast as we can.
  //
  // There are oppportunities for optimization here, check out Peter's
  // speedtable query optimizer for optimizations the determination of
  // which can be made here instead.  Shrug.
  //
  CTABLE_INTERNAL int
  ctable_searchMatchPatternCheck (char *s) {
      char c;
  if(!s) Tcl_Panic("ctable_searchMatchPatternCheck called with null");
  
      int firstCharIsStar = 0;
      int lastCharIsStar = 0;
  
      if (*s == '\0') {
  	return CTABLE_STRING_MATCH_ANCHORED;
      }
  
      if (*s++ == '*') {
  	firstCharIsStar = 1;
      }
  
      while ((c = *s++) != '\0') {
  	switch (c) {
  	  case '*':
  	    if (*s == '\0') {
  		lastCharIsStar = 1;
  	    } else {
  		// some other * in the middle, too fancy
  		return CTABLE_STRING_MATCH_PATTERN;
  	    }
  	    break;
  
  	  case '?':
  	  case '[':
  	  case ']':
  	  case '\\':
  	    return CTABLE_STRING_MATCH_PATTERN;
  
  	}
      }
  
      if (firstCharIsStar) {
  	if (lastCharIsStar) {
  	    return CTABLE_STRING_MATCH_UNANCHORED;
  	}
  	// we could do a reverse anchored search here but i don't
  	// think this comes up often enough (*pattern) to warrant it
  	return CTABLE_STRING_MATCH_PATTERN;
      }
  
      if (lastCharIsStar) {
  	// first char is not star, last char is
  	return CTABLE_STRING_MATCH_ANCHORED;
      }
  
      // first char is not star, last char is not star, and there are
      // no other metacharacters in the string. This is bad because they
      // should be using "=" not "match" but we'll use pattern because
      // that will actually do the right thing.  alternatively we could
      // add another string match pattern type
      return CTABLE_STRING_MATCH_PATTERN;
  }
  
  CTABLE_INTERNAL int ctable_CreateInRows(Tcl_Interp *interp, CTable *ctable, CTableSearchComponent *component)
  {
      int i;
  
      if(component->inListRows || component->inCount == 0)
  	return TCL_OK;
  
      component->inListRows = (ctable_BaseRow **)ckalloc(component->inCount * sizeof (ctable_BaseRow *));
  
      // Since the main loop may abort, make sure this is clean
      for(i = 0; i < component->inCount; i++) {
  	component->inListRows[i] = NULL;
      }
  
      for(i = 0; i < component->inCount; i++) {
  	component->inListRows[i] = (*ctable->creator->make_empty_row) (ctable);
  
  	if ((*ctable->creator->set) (interp, ctable, component->inListObj[i], component->inListRows[i], component->fieldID, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  	    Tcl_AddErrorInfo (interp, "\n    while processing \"in\" compare function");
  	    return TCL_ERROR;
  	}
      }
      return TCL_OK;
  }
  
  CTABLE_INTERNAL void ctable_FreeInRows(CTable *ctable, CTableSearchComponent *component)
  {
      if(component->inListRows) {
  	int i;
  	for(i = 0; i < component->inCount; i++) {
  	    if(component->inListRows[i]) {
  	        ctable->creator->delete_row (ctable, component->inListRows[i], CTABLE_INDEX_PRIVATE);
  	    }
  	}
  	ckfree((char*)component->inListRows);
  	component->inListRows = NULL;
      }
  }
  
  static int
  ctable_ParseSearch (Tcl_Interp *interp, CTable *ctable, Tcl_Obj *componentListObj, CONST char **fieldNames, CTableSearch *search) {
      Tcl_Obj    **componentList;
      int          componentIdx;
      int          componentListCount;
  
      Tcl_Obj    **termList;
      int          termListCount;
  
      int          field;
  
      CTableSearchComponent  *components;
      CTableSearchComponent  *component;
  
      static CONST char *searchTerms[] = CTABLE_SEARCH_TERMS;
  
      if (Tcl_ListObjGetElements (interp, componentListObj, &componentListCount, &componentList) == TCL_ERROR) {
          return TCL_ERROR;
      }
  
      if (componentListCount == 0) {
          search->components = NULL;
  	return TCL_OK;
      }
  
      search->nComponents = componentListCount;
  
      components = (CTableSearchComponent *)ckalloc (componentListCount * sizeof (CTableSearchComponent));
  
      search->components = components;
  
      for (componentIdx = 0; componentIdx < componentListCount; componentIdx++) {
          int term;
  
  	if (Tcl_ListObjGetElements (interp, componentList[componentIdx], &termListCount, &termList) == TCL_ERROR) {
  	  err:
  	    ckfree ((char *)components);
  	    search->components = NULL;
  	    return TCL_ERROR;
  	}
  
  	if (termListCount < 2) {
  	    // would be cool to support regexps here too
  	    Tcl_WrongNumArgs (interp, 0, termList, "term field ?value..?");
  	    goto err;
  	}
  
  	if (Tcl_GetIndexFromObj (interp, termList[0], searchTerms, "term", TCL_EXACT, &term) != TCL_OK) {
  	    goto err;
  	}
  
  	if (Tcl_GetIndexFromObj (interp, termList[1], fieldNames, "field", TCL_EXACT, &field) != TCL_OK) {
  	    goto err;
  	}
  
  	component = &components[componentIdx];
  
  	component->comparisonType = term;
  	component->fieldID = field;
  	component->clientData = NULL;
  	component->row1 = NULL;
  	component->row2 = NULL;
  	component->row3 = NULL;
  	component->inListObj = NULL;
  	component->inListRows = NULL;
  	component->inCount = 0;
  	component->compareFunction = ctable->creator->fields[field]->compareFunction;
  
  	if (term == CTABLE_COMP_FALSE || term == CTABLE_COMP_TRUE || term == CTABLE_COMP_NULL || term == CTABLE_COMP_NOTNULL) {
  	    if (termListCount != 2) {
  		Tcl_AppendResult (interp, "false, true, null and notnull search expressions must have only two fields", (char *) NULL);
  		goto err;
  	    }
  	} else {
  
  	    if (term == CTABLE_COMP_IN) {
  	        if (termListCount != 3) {
  		    Tcl_AppendResult (interp, "term \"", Tcl_GetString (termList[0]), "\" require 3 arguments (term, field, list)", (char *) NULL);
  		    goto err;
  		}
  
  		if (Tcl_ListObjGetElements (interp, termList[2], &component->inCount, &component->inListObj) == TCL_ERROR) {
  		    goto err;
  		}
  
  	    } else if (term == CTABLE_COMP_RANGE) {
  	        ctable_BaseRow *row;
  
  	        if (termListCount != 4) {
  		    Tcl_AppendResult (interp, "term \"", Tcl_GetString (termList[0]), "\" require 4 arguments (term, field, lowValue, highValue)", (char *) NULL);
  		    goto err;
  		}
  
  		row = (*ctable->creator->make_empty_row) (ctable);
  		if ((*ctable->creator->set) (interp, ctable, termList[2], row, field, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  		    goto err;
  		}
  		component->row1 = row;
  
  		row = (*ctable->creator->make_empty_row) (ctable);
  		if ((*ctable->creator->set) (interp, ctable, termList[3], row, field, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  		    goto err;
  		}
  		component->row2 = row;
  
  		continue;
  
  	    } else if (termListCount != 3) {
  		Tcl_AppendResult (interp, "term \"", Tcl_GetString (termList[0]), "\" require 3 arguments (term, field, value)", (char *) NULL);
  		goto err;
  	    }
  
  	    if ((term == CTABLE_COMP_MATCH) || (term == CTABLE_COMP_NOTMATCH) || (term == CTABLE_COMP_MATCH_CASE) || (term == CTABLE_COMP_NOTMATCH_CASE)) {
  		// Check if field that supports string matches
  		int ftype = ctable->creator->fieldTypes[field];
  		if(ftype != CTABLE_TYPE_FIXEDSTRING && ftype != CTABLE_TYPE_VARSTRING && ftype != CTABLE_TYPE_KEY) {
  		    Tcl_AppendResult (interp, "term \"", Tcl_GetString (termList[1]), "\" must be a string type for \"", Tcl_GetString (termList[0]), "\" operation", (char *) NULL);
  		    goto err;
  		}
  
  		struct ctableSearchMatchStruct *sm = (struct ctableSearchMatchStruct *)ckalloc (sizeof (struct ctableSearchMatchStruct));
  
  		sm->type = ctable_searchMatchPatternCheck (Tcl_GetString (termList[2]));
  		sm->nocase = ((term == CTABLE_COMP_MATCH) || (term == CTABLE_COMP_NOTMATCH));
  
  		if (sm->type == CTABLE_STRING_MATCH_UNANCHORED) {
  		    char *needle;
  		    int len;
  
  		    needle = Tcl_GetStringFromObj (termList[2], &len);
  		    boyer_moore_setup (sm, (unsigned char *)needle + 1, len - 2, sm->nocase);
  		} else if(sm->type == CTABLE_STRING_MATCH_ANCHORED && term == CTABLE_COMP_MATCH_CASE) {
  		    int len;
  		    char *needle = Tcl_GetStringFromObj (termList[2], &len);
  		    char *prefix = (char *) ckalloc(len+1);
  		    int i;
  
  		    /* stash the prefix of the match into row2 */
  		    for(i = 0; i < len; i++) {
  			if(needle[i] == '*') {
  			    break;
  			}
  			prefix[i] = needle[i];
  			break;
  		    }
  
  		    // This test should never fail.
  		    if(i > 0) {
  		        ctable_BaseRow *row;
  		        prefix[i] = '\0';
  
  		        row = (*ctable->creator->make_empty_row) (ctable);
  	    		if ((*ctable->creator->set) (interp, ctable, Tcl_NewStringObj (prefix, -1), row, field, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  			    goto err;
  	    	        }
  	    		component->row2 = row;
  
  			// Now set up row3 as the first non-matching string
  			prefix[i-1]++;
  
  		        row = (*ctable->creator->make_empty_row) (ctable);
  	    		if ((*ctable->creator->set) (interp, ctable, Tcl_NewStringObj (prefix, -1), row, field, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  			    goto err;
  	    	        }
  	    		component->row3 = row;
  
  		    }
  		    ckfree(prefix);
  		}
  
  		component->clientData = sm;
  	    }
  
  	    /* stash what we want to compare to into a row as in "range"
  	     */
  	    if (term != CTABLE_COMP_IN) {
  		ctable_BaseRow *row;
  		row = (*ctable->creator->make_empty_row) (ctable);
  		if ((*ctable->creator->set) (interp, ctable, termList[2], row, field, CTABLE_INDEX_PRIVATE) == TCL_ERROR) {
  			goto err;
  		}
  		component->row1 = row;
  	    }
  	}
      }
  
      // it worked, leave the components allocated
      return TCL_OK;
  }
  
  static int
  ctable_ParseFilters (Tcl_Interp *interp, CTable *ctable, Tcl_Obj *filterListObj, CTableSearch *search) {
      Tcl_Obj               **filterList;
      int                     filterListCount;
  
      CTableSearchFilter     *filters = NULL;
  
      int i;
  
      // Assume nothing, trust no one.
      search->filters = NULL;
      search->nFilters = 0;
  
      if (Tcl_ListObjGetElements (interp, filterListObj, &filterListCount, &filterList) == TCL_ERROR)
          goto abend;
  
      // Nothing to do
      if (filterListCount == 0) {
          return TCL_OK;
      }
  
  
      // List is {{filtername filtervalue} {filtername filtervalue}} .. if passing a list to a filter, pass a list
  
      // make room, make room
      filters = (CTableSearchFilter *)ckalloc (filterListCount * sizeof (CTableSearchFilter));
  
      // step through list, looking for filter names and filling in the structs
      for (i = 0; i < filterListCount; i++) {
          Tcl_Obj    **termList;
          int          termListCount;
          int item;
  
          if (Tcl_ListObjGetElements (interp, filterList[i], &termListCount, &termList) == TCL_ERROR || termListCount != 2) {
            Tcl_AppendResult (interp, "each term of the filter must be a nested list with 2 items", (char *) NULL);
            goto abend;
          }
  
  
          if (Tcl_GetIndexFromObj (interp, termList[0], ctable->creator->filterNames, "filter", TCL_EXACT, &item) != TCL_OK)
              goto abend;
  
          filters[i].filterFunction = ctable->creator->filterFunctions[item];
          filters[i].filterObject = termList[1];
      }
  
      // Register the filter list we've created
      search->filters = filters;
      search->nFilters = filterListCount;
      return TCL_OK;
  
  abend:
      if(filters)
          ckfree ((char *)filters);
      return TCL_ERROR;
  }
  
  //
  // ctable_SearchAction - Perform the search action on a row that's matched
  //  the search criteria.
  //
  static int
  ctable_SearchAction (Tcl_Interp *interp, CTable *ctable, CTableSearch *search, ctable_BaseRow *row) {
      char           *key;
      ctable_CreatorTable *creator = ctable->creator;
  
      key = row->hashEntry.key;
  
      if (search->action == CTABLE_SEARCH_ACTION_WRITE_TABSEP) {
  	Tcl_DString     dString;
  
  	Tcl_DStringInit (&dString);
  
  	// string-append the specified fields, or all fields, tab separated
  
  	if (search->nRetrieveFields < 0) {
  	    (*creator->dstring_append_get_tabsep) (key, row, creator->publicFieldList, creator->nPublicFields, &dString, search->noKeys, search->sepstr, search->quoteType, search->nullString);
  	} else {
  	    (*creator->dstring_append_get_tabsep) (key, row, search->retrieveFields, search->nRetrieveFields, &dString, search->noKeys, search->sepstr, search->quoteType, search->nullString);
  	}
  
  	// write the line out
  
  	if (Tcl_WriteChars (search->tabsepChannel, Tcl_DStringValue (&dString), Tcl_DStringLength (&dString)) < 0) {
  	    return TCL_ERROR;
  	}
  
  	Tcl_DStringFree (&dString);
  	return TCL_OK;
      }
  
      // if there's a code body to eval...
  
      if (search->codeBody != NULL) {
  	Tcl_Obj *listObj = NULL;
  	int      evalResult;
  
  	// generate the list of requested fields, or all fields, in
  	// "get style" (value only), "array get style" (key-value
  	// pairs with nulls suppressed) and "array get with nulls"
  	// style, (all requested fields, null or not)
  
  	switch (search->action) {
  
  	  case CTABLE_SEARCH_ACTION_GET: {
  	    if (search->nRetrieveFields < 0) {
  		listObj = creator->gen_list (interp, row);
  	    } else {
  	       int i;
  
  	       listObj = Tcl_NewObj ();
  	       for (i = 0; i < search->nRetrieveFields; i++) {
  		   creator->lappend_field (interp, listObj, row, search->retrieveFields[i]);
  	       }
  	    }
  	    break;
  	  }
  
  	  case CTABLE_SEARCH_ACTION_ARRAY_WITH_NULLS:
  	  case CTABLE_SEARCH_ACTION_ARRAY: {
  	    int result = TCL_OK;
  
  	    // Clear array before filling it in. Ignore failure because it's
  	    // OK for the array not to exist at this point.
  	    Tcl_UnsetVar2(interp, Tcl_GetString(search->rowVarNameObj), NULL, 0);
  
  	    if (search->nRetrieveFields < 0) {
  	       int i;
  
  	       for (i = 0; i < creator->nFields; i++) {
  		   if (is_hidden_field(creator,i) && !is_key_field(creator,i,search->noKeys)) {
  		       continue;
  		   }
  	           if (search->action == CTABLE_SEARCH_ACTION_ARRAY) {
  		       result = creator->array_set (interp, search->rowVarNameObj, row, i);
  		   } else {
  		       result = creator->array_set_with_nulls (interp, search->rowVarNameObj, row, i);
  		   }
  	       }
  	    } else {
  	       int i;
  
  	       for (i = 0; i < search->nRetrieveFields; i++) {
  	           if (search->action == CTABLE_SEARCH_ACTION_ARRAY) {
  		       result = creator->array_set (interp, search->rowVarNameObj, row, search->retrieveFields[i]);
  		   } else {
  		       result = creator->array_set_with_nulls (interp, search->rowVarNameObj, row, search->retrieveFields[i]);
  		   }
  	       }
  	    }
  
  	    if (result != TCL_OK) {
  	        return result;
  	    }
  	    break;
  	  }
  
  	  case CTABLE_SEARCH_ACTION_ARRAY_GET: {
  	    int i;
  
  	    listObj = Tcl_NewObj ();
  	    if (search->nRetrieveFields < 0) {
  		for (i = 0; i < creator->nFields; i++) {
  		    if (is_hidden_field(creator,i) && !is_key_field(creator,i,search->noKeys)) {
  			continue;
  		    }
  		    creator->lappend_nonnull_field_and_name (interp, listObj, row, i);
  		 }
  	    } else {
  		for (i = 0; i < search->nRetrieveFields; i++) {
  		    creator->lappend_nonnull_field_and_name (interp, listObj, row, search->retrieveFields[i]);
  		}
  	    }
  	    break;
  	  }
  
  	  case CTABLE_SEARCH_ACTION_ARRAY_GET_WITH_NULLS: {
  	    int i;
  
  	    listObj = Tcl_NewObj ();
  	    if (search->nRetrieveFields < 0) {
  		for (i = 0; i < creator->nFields; i++) {
  		    if (is_hidden_field(creator,i) && !is_key_field(creator,i,search->noKeys)) {
  			continue;
  		    }
  		    creator->lappend_field_and_name (interp, listObj, row, i);
  		}
  	    } else {
  		listObj = Tcl_NewObj ();
  		for (i = 0; i < search->nRetrieveFields; i++) {
  		    creator->lappend_field_and_name (interp, listObj, row, search->retrieveFields[i]);
  		}
  	    }
  	    break;
  	  }
  
  	  default: {
  	      if (search->keyVarNameObj == NULL) {
  	          Tcl_Panic ("software failure - unhandled search action");
  	      }
  	  }
  	}
  
  	// if the key var is defined, set the key into it
  	if (search->keyVarNameObj != NULL) {
  	    if (Tcl_ObjSetVar2 (interp, search->keyVarNameObj, (Tcl_Obj *)NULL, Tcl_NewStringObj (key, -1), TCL_LEAVE_ERR_MSG) == (Tcl_Obj *) NULL) {
  		return TCL_ERROR;
  	    }
  	}
  
  	// set the returned list into the value var
  	if ((listObj != NULL) && (Tcl_ObjSetVar2 (interp, search->rowVarNameObj, (Tcl_Obj *)NULL, listObj, TCL_LEAVE_ERR_MSG) == (Tcl_Obj *) NULL)) {
  	    return TCL_ERROR;
  	}
  
  	// evaluate the code body
  	//
  	// By using a Tcl object for the code body, the code body will be
          // on-the-fly compiled by Tcl once and cached on subsequent
  	// evals.  Cool.
  	//
  	evalResult = Tcl_EvalObjEx (interp, search->codeBody, 0);
  	switch (evalResult) {
  	  case TCL_ERROR:
  	    Tcl_AddErrorInfo (interp, "\n   while processing search code body");
  	    return TCL_ERROR;
  
  	  case TCL_OK:
  	  case TCL_CONTINUE:
  	  case TCL_BREAK:
  	    Tcl_ResetResult(interp);
  	  case TCL_RETURN:
  	    return evalResult;
  	}
      }
  
      return TCL_OK;
  }
  
  //
  // ctable_checkForKey - check for a "key" field, and if there set the internal
  // noKeys flag to suppress the separate output of the "_key" field.
  //
  CTABLE_INTERNAL void ctable_checkForKey(CTable *ctable, CTableSearch *search)
  {
      if (!search->noKeys) {
  	int		     i;
          int                 *fields;
          int                  nFields;
          ctable_CreatorTable *creator = ctable->creator;
  
          if (search->nRetrieveFields < 0) {
  	    fields = creator->publicFieldList;
  	    nFields = creator->nPublicFields;
          } else {
  	    nFields = search->nRetrieveFields;
  	    fields = search->retrieveFields;
          }
  
          for (i = 0; i < nFields; i++) {
  	    if(fields[i] == creator->keyField) {
  		search->noKeys = 1;
  		break;
  	    }
  	}
      }
  }
  
  //
  // ctable_WriteFieldNames - write field names from a search structure to
  //   the specified channel, tab-separated
  //
  //
  static int
  ctable_WriteFieldNames (Tcl_Interp *interp, CTable *ctable, CTableSearch *search)
  {
      int i;
      Tcl_DString          dString;
      int                 *fields;
      int                  nFields;
      ctable_CreatorTable *creator = ctable->creator;
  
      Tcl_DStringInit (&dString);
  
      if (search->nRetrieveFields < 0) {
  	fields = creator->publicFieldList;
  	nFields = creator->nPublicFields;
      } else {
  	nFields = search->nRetrieveFields;
  	fields = search->retrieveFields;
      }
  
      // Generate key field if necessary
      if (!search->noKeys)
          Tcl_DStringAppend (&dString, "_key", 4);
  
      for (i = 0; i < nFields; i++) {
  	if (!search->noKeys || i != 0) {
  	    Tcl_DStringAppend(&dString, search->sepstr, -1);
  	}
  
  	Tcl_DStringAppend(&dString, creator->fields[fields[i]]->name, -1);
      }
      Tcl_DStringAppend(&dString, "\n", 1);
  
      if (Tcl_WriteChars (search->tabsepChannel, Tcl_DStringValue (&dString), Tcl_DStringLength (&dString)) < 0) {
  	Tcl_DStringFree (&dString);
  	return TCL_ERROR;
      }
  
      Tcl_DStringFree (&dString);
      return TCL_OK;
  }
  
  //
  // ctable_PerformTransactions - atomic actions taken at the end of a search
  //
  static int
  ctable_PerformTransaction (Tcl_Interp *interp, CTable *ctable, CTableSearch *search)
  {
      int fieldIndex, rowIndex;
      ctable_CreatorTable *creator = ctable->creator;
  
      if(search->tranType == CTABLE_SEARCH_TRAN_NONE || search->tranType == CTABLE_SEARCH_TRAN_CURSOR) {
  	return TCL_OK;
      }
  
      if(search->tranType == CTABLE_SEARCH_TRAN_DELETE) {
  
        // walk the result and delete the matched rows
        for (rowIndex = search->offset; rowIndex < search->offsetLimit; rowIndex++) {
  	  (*creator->delete_row) (ctable, search->tranTable[rowIndex], CTABLE_INDEX_NORMAL);
  	  ctable->count--;
        }
  
        return TCL_OK;
  
      }
  
      if(search->tranType == CTABLE_SEARCH_TRAN_UPDATE) {
  
  	int objc;
  	Tcl_Obj **objv;
  	int *updateFields;
  
  	// Convert the tranData field into a list
          if (Tcl_ListObjGetElements (interp, search->tranData, &objc, &objv) == TCL_ERROR) {
  updateParseError:
  	    Tcl_AppendResult (interp, " parsing argument to -update", (char *)NULL);
  	    return TCL_ERROR;
  	}
  
  	if(objc & 1) {
  	    Tcl_AppendResult (interp, "-update list must contain an even number of elements", (char *)NULL);
              return TCL_ERROR;
          }
  
  	// get some space for the field indices
  	updateFields = (int *)ckalloc (sizeof (int) * (objc/2));
  
  	// Convert the names to field indices
  	for(fieldIndex = 0; fieldIndex < objc/2; fieldIndex++) {
  	    if (Tcl_GetIndexFromObj (interp, objv[fieldIndex*2], creator->fieldNames, "field", TCL_EXACT, &updateFields[fieldIndex]) != TCL_OK) {
  	        ckfree((char*)updateFields);
  		goto updateParseError;
  	    }
  	}
  
  	// Perform update
          for (rowIndex = search->offset; rowIndex < search->offsetLimit; rowIndex++) {
  	    ctable_BaseRow *row = search->tranTable[rowIndex];
  
  	    for(fieldIndex = 0; fieldIndex < objc/2; fieldIndex++) {
  		if((*creator->set)(interp, ctable, objv[fieldIndex*2+1], row, updateFields[fieldIndex], CTABLE_INDEX_NORMAL) == TCL_ERROR) {
  		    ckfree((char*)updateFields);
  		    Tcl_AppendResult (interp, " (update may be incomplete)", (char *)NULL);
  		    return TCL_ERROR;
  		}
  	    }
  	}
  
  	ckfree((char*)updateFields);
  
          return TCL_OK;
      }
  
      Tcl_AppendResult(interp, "internal error - unknown transaction type %d", search->tranType);
  
      return TCL_ERROR;
  }
  
  //
  // ctable_search_poll - called periodically from a search to avoid blocking
  // the Tcl event loop.
  //
  CTABLE_INTERNAL int ctable_search_poll(Tcl_Interp *interp, CTable *ctable, CTableSearch *search)
  {
      if(search->pollCodeBody) {
  	int result = Tcl_EvalObjEx (interp, search->pollCodeBody, 0);
  	if(result == TCL_ERROR) {
  	    Tcl_BackgroundError(interp);
  	    Tcl_ResetResult(interp);
  	}
      } else {
  	Tcl_DoOneEvent(0);
      }
      return TCL_OK;
  }
  
  //
  // ctable_PostSearchCommonActions - actions taken at the end of a search
  //
  // If results sorting is required, we sort the results.
  //
  // We interpret start and offset, if set, to limit rows returned.
  //
  // We walk the sort results, calling ctable_SearchAction on each.
  //
  // If transaction processing is required, we call ctable_PerformTransactions
  //
  //
  static int
  ctable_PostSearchCommonActions (Tcl_Interp *interp, CTable *ctable, CTableSearch *search)
  {
      int walkIndex;
      ctable_CreatorTable *creator = ctable->creator;
      int actionResult = TCL_OK;
  
      // if there was nothing matched, or we're before the offset, we're done
      if(search->matchCount == 0 || search->offset > search->matchCount) {
  	// And we didn't match anything
  	search->matchCount = 0;
  	return TCL_OK;
      }
  
      // if we're not sorting or performing a transaction, we're done -- we
      // did 'em all on the fly
      if (search->tranTable == NULL) {
  	// But account for the offset in the matchCount
  	search->matchCount -= search->offset;
          return TCL_OK;
      }
  
      // figure out the last row they could want, if it's more than what's
      // there, set it down to what came back
      if ((search->offsetLimit == 0) || (search->offsetLimit > search->matchCount)) {
          search->offsetLimit = search->matchCount;
      }
  
      if(search->sortControl.nFields) {	// sorting
        ctable_qsort_r (search->tranTable, search->matchCount, sizeof (ctable_HashEntry *), &search->sortControl, (cmp_t*) creator->sort_compare);
      }
  
      if (search->tranType == CTABLE_SEARCH_TRAN_CURSOR) {
  	search->cursor = ctable_CreateCursor(interp, ctable, search);
      } else if(search->bufferResults == CTABLE_BUFFER_DEFER) { // we deferred the operation to here
          // walk the result
          for (walkIndex = search->offset; walkIndex < search->offsetLimit; walkIndex++) {
  	    if(search->pollInterval && --search->nextPoll <= 0) {
  		if(ctable_search_poll(interp, ctable, search) == TCL_ERROR)
  		    return TCL_ERROR;
  		search->nextPoll = search->pollInterval;
  	    }
  	    /* here is where we want to take the match actions
  	     * when we are sorting or otherwise buffering
  	     */
  	    actionResult = ctable_SearchAction (interp, ctable, search, search->tranTable[walkIndex]);
  	    switch (actionResult) {
  	        case TCL_CONTINUE:
  	        case TCL_OK: {
  	            actionResult = TCL_OK;
  	            break;
  	        }
  	        case TCL_BREAK:
  	        case TCL_RETURN: {
  		    // We DO count this row as a match, to be consistent with
  		    // the non-deferred case.
  		    search->offsetLimit = walkIndex + 1;
  	            goto normal_return;
  	        }
  	        case TCL_ERROR: {
  	            return TCL_ERROR;
  	        }
  	    }
          }
      }
  
  normal_return:
      // Calculate the final match count
      search->matchCount = search->offsetLimit - search->offset;
  
      // Finally, perform any pending transaction.
      if(search->tranType != CTABLE_SEARCH_TRAN_NONE && search->tranType !=  CTABLE_SEARCH_TRAN_CURSOR) {
  	if (ctable_PerformTransaction(interp, ctable, search) == TCL_ERROR) {
  	    return TCL_ERROR;
  	}
      }
  
      return actionResult;
  }
  
  //
  // ctable_SearchCompareRow - perform comparisons on a row
  //
  INLINE static int
  ctable_SearchCompareRow (Tcl_Interp *interp, CTable *ctable, CTableSearch *search, ctable_BaseRow *row)
  {
      int   compareResult;
      int   actionResult;
  
      // Handle polling
      if(search->pollInterval && --search->nextPoll <= 0) {
  	if(ctable_search_poll(interp, ctable, search) == TCL_ERROR)
  	    return TCL_ERROR;
  	search->nextPoll = search->pollInterval;
      }
  
      // if we have a match pattern (for the key) and it doesn't match,
      // skip this row
  
      if (search->pattern != (char *) NULL) {
  	if (!Tcl_StringCaseMatch (row->hashEntry.key, search->pattern, 1)) {
  	    return TCL_CONTINUE;
  	}
      }
  
      // check filters
      if (search->nFilters) {
  	int i;
  	int filterResult = TCL_OK;
  
  	for(i = 0; i < search->nFilters; i++) {
  	    filterFunction_t f = search->filters[i].filterFunction;
  	    filterResult = (*f) (interp, ctable, row, search->filters[i].filterObject, search->sequence);
  	    if(filterResult != TCL_OK)
  		return filterResult;
  	}
      }
  
      //
      // run the supplied compare routine
      //
      compareResult = (*ctable->creator->search_compare) (interp, search, row);
      if (compareResult == TCL_CONTINUE) {
  	return TCL_CONTINUE;
      }
  
      if (compareResult == TCL_ERROR) {
  	return TCL_ERROR;
      }
  
      // It's a Match 
      // Are we sorting? Plop the match in the sort table and return
      // Increment count in this block to make sure it's incremented in all
      // paths.
  
      if (search->tranTable == NULL) {
  	++search->matchCount;
      } else {
  	/* We are buffering the results (eg, for a sort or a transaction)
  	 * so just return, we'll do the heavy lifting later. */
  	assert (search->matchCount <= ctable->count);
  	search->tranTable[search->matchCount++] = row;
  
  	// If buffering for an important reason (eg, sorting), defer
  	if(search->bufferResults == CTABLE_BUFFER_DEFER)
  	    return TCL_CONTINUE;
      }
  
      // We're not deferring the results, let's figure out what to do as we
      // match. If we haven't met the start point, blow it off.
      if (search->matchCount <= search->offset) {
  	return TCL_CONTINUE;
      }
  
      if (search->action == CTABLE_SEARCH_ACTION_NONE) {
  	// we're only counting -- if there is a limit and it's been 
  	// met, we're done
  	if ((search->limit != 0) && (search->matchCount >= search->offsetLimit)) {
  	    return TCL_BREAK;
  	}
  
  	// the limit hasn't been exceeded or there isn't one,
  	// so we keep counting -- but we continue here because
  	// we don't need to do any processing on the line
  	return TCL_CONTINUE;
      }
  
      /* we want to take the match actions here -- we're here when we aren't
       * buffering (at least not for an "important" reason)
       */
      actionResult = ctable_SearchAction (interp, ctable, search, row);
      if (actionResult == TCL_ERROR) {
  	return TCL_ERROR;
      }
  
      if ((actionResult == TCL_CONTINUE) || (actionResult == TCL_OK)) {
  	// if there was a limit and we've met it, we're done
  	if ((search->limit != 0) && (search->matchCount >= search->offsetLimit)) {
  	    return TCL_BREAK;
  	}
  	return TCL_CONTINUE;
      }
  
      if ((actionResult == TCL_BREAK) || (actionResult == TCL_RETURN)) {
  	return actionResult;
      }
  
      Tcl_Panic("software failure - unhandled SearchAction return");
      return TCL_ERROR;
  }
  
  enum skipStart_e {
      SKIP_START_NONE, SKIP_START_GE_ROW1, SKIP_START_GT_ROW1, SKIP_START_EQ_ROW1, SKIP_START_RESET
  };
  enum skipEnd_e {
      SKIP_END_NONE, SKIP_END_NE_ROW1, SKIP_END_GE_ROW1, SKIP_END_GT_ROW1, SKIP_END_GE_ROW2
  };
  enum skipNext_e {
      SKIP_NEXT_NONE, SKIP_NEXT_ROW, SKIP_NEXT_MATCH, SKIP_NEXT_IN_LIST
  };
  
  // skiplist scanning rules table.
  //   scan starts at skipStart, goes to skipEnd, and uses skipNext to traverse.
  //   score is based on an estimate of how much a scan on this field is likely
  //   to limit the search.
  // rationale:
  //   <, <=, or >= are 2 points for being constrained at one end
  //   > is only 1 point because we need to loop over "=" at start
  //   range is 4 points for being constrained at both ends
  //   = is 6 points for being a "minimum range"
  //   in is 100 points (and win) because it HAS to be done over a list
  // plus one point for being the sort field, because that saves creating and
  //   sorting the transaction table.
  //
  static struct {
      enum skipStart_e	skipStart;
      enum skipEnd_e	skipEnd;
      enum skipNext_e	skipNext;
      int			score;
  } skipTypes[] = {
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // FALSE
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // TRUE
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // NULL
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // NOTNULL
    {SKIP_START_RESET,	SKIP_END_GE_ROW1, SKIP_NEXT_ROW,   2 }, // LT
    {SKIP_START_RESET,	SKIP_END_GT_ROW1, SKIP_NEXT_ROW,   2 }, // LE
    {SKIP_START_EQ_ROW1,	SKIP_END_GT_ROW1, SKIP_NEXT_ROW,   6 }, // EQ
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // NE
    {SKIP_START_GE_ROW1,	SKIP_END_NONE,	  SKIP_NEXT_ROW,   2 }, // GE
    {SKIP_START_GT_ROW1,	SKIP_END_NONE,	  SKIP_NEXT_ROW,   1 }, // GT
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // MATCH
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // NOTMATCH
    {SKIP_START_GE_ROW1,	SKIP_END_GE_ROW2, SKIP_NEXT_MATCH, 3 }, // MATCH_CASE
    {SKIP_START_NONE,	SKIP_END_NONE,	  SKIP_NEXT_NONE, -2 }, // NOTMATCH_CASE
    {SKIP_START_GE_ROW1,	SKIP_END_GE_ROW2, SKIP_NEXT_ROW,   4 }, // RANGE
    {SKIP_START_RESET,	SKIP_END_NONE, SKIP_NEXT_IN_LIST,  6 }  // IN
  };
  
  #define SORT_SCORE 1 // being able to sort is worth 1 point
       // TODO: see if this should be higher
       //       if you change this, change the "not implemented"
       //       value in the table above to -1 - SORT_SCORE
  
  enum walkType_e { WALK_DEFAULT, WALK_SKIP, WALK_HASH_EQ, WALK_HASH_IN };
  
  static enum walkType_e hashTypes[] = {
    WALK_DEFAULT, WALK_DEFAULT, WALK_DEFAULT, WALK_DEFAULT, // FALSE..NOTNULL
    WALK_DEFAULT, WALK_DEFAULT, WALK_HASH_EQ, WALK_DEFAULT, // LT..NE
    WALK_DEFAULT, WALK_DEFAULT, WALK_DEFAULT, WALK_DEFAULT, // GT..NOTMATCH
    WALK_DEFAULT, WALK_DEFAULT, WALK_DEFAULT, WALK_HASH_IN  // MATCH_CASE..IN
  };
  
  #ifdef WITH_SHARED_TABLES
  struct restart_t {
      ctable_BaseRow	  *row1;
      ctable_BaseRow	  *row2;
      enum skipStart_e	   skipStart;
      enum skipEnd_e	   skipEnd;
      fieldCompareFunction_t compareFunction;
  };
  
  CTABLE_INTERNAL int ctable_SearchRestartNeeded(ctable_BaseRow *row, struct restart_t *restart)
  {
      // check to see if we're still following the right skiplist, by seeing
      // if the value we're following still satisfies the criteria
      switch(restart->skipEnd) {
  	case SKIP_END_GE_ROW1: {
  	    if (restart->compareFunction (row, restart->row1) >= 0)
  		return 1;
  	    break;
  	}
  	case SKIP_END_GT_ROW1: {
  	    if (restart->compareFunction (row, restart->row1) > 0)
  		return 1;
  	    break;
  	}
  	case SKIP_END_GE_ROW2: {
  	    if (restart->compareFunction (row, restart->row2) >= 0)
  		return 1;
  	    break;
  	}
  	default: {
  	    break;
  	}
      }
      switch(restart->skipStart) {
          case SKIP_START_GE_ROW1: {
  	    if (restart->compareFunction (row, restart->row1) < 0)
  		return 1;
  	    break;
  	}
  	case SKIP_START_GT_ROW1: {
  	    if (restart->compareFunction (row, restart->row1) <= 0)
  		return 1;
  	    break;
  	}
  	case SKIP_START_EQ_ROW1: {
  	    if (restart->compareFunction (row, restart->row1) != 0)
  		return 1;
  	    break;
  	}
          case SKIP_START_NONE:
  	case SKIP_START_RESET: {
  	    break;
  	}
      }
      // Got through here... no restart needed
      return 0;
  }
  #endif
  
  //
  // ctable_PrepareTransactions - set up buffering for transactions if needed
  //
  CTABLE_INTERNAL void ctable_PrepareTransactions(CTable *ctable, CTableSearch *search)
  {
      // buffer results if:
      //   We've got search actions, AND...
      //     We're searching, or...
      //     We're a reader table, or...
      //     We explicitly requested bufering.
      if (search->action == CTABLE_SEARCH_ACTION_NONE) {
  	search->bufferResults = CTABLE_BUFFER_NONE;
      } else if(search->action == CTABLE_SEARCH_ACTION_CURSOR) {
  	search->bufferResults = CTABLE_BUFFER_DEFER;
      } else if(search->sortControl.nFields > 0) {
  	search->bufferResults = CTABLE_BUFFER_DEFER;
  #ifdef WITH_SHARED_TABLES
      } else if(ctable->share_type == CTABLE_SHARED_READER) {
  	search->bufferResults = CTABLE_BUFFER_DEFER;
  #endif
      } else if(search->bufferResults < 0) {
      	search->bufferResults = CTABLE_BUFFER_NONE;
      }
  
      // If we're not buffering for any other reason, buffer for a transaction
      // but set bufferResults to PROVISIONAL so we can still terminate the
      // search early
      if(search->bufferResults == CTABLE_BUFFER_NONE && search->tranType != CTABLE_SEARCH_TRAN_NONE) {
  	search->bufferResults = CTABLE_BUFFER_PROVISIONAL;
      }
  
      // if we're buffering,
      // allocate a space for the search results that we'll then sort from
      if (search->bufferResults != CTABLE_BUFFER_NONE) {
  	search->tranTable = (ctable_BaseRow **)ckalloc (sizeof (ctable_BaseRow *) * ctable->count);
      }
  }
  
  //
  // ctable_PerformSearch - perform the search
  //
  // write field names if we need to
  //
  // for each row in the table, apply search compare test in turn
  //    the first one that comes up that the row should be excluded ends
  //    looking at that one
  //
  //    if nothing excluded it, we pick it -- this can mean taking the action
  //    immediately or, if sorting, picking the object for sorting and then
  //    taking the action
  //
  //
  CTABLE_INTERNAL int
  ctable_PerformSearch (Tcl_Interp *interp, CTable *ctable, CTableSearch *search) {
      int           	  compareResult;
      int			  finalResult = TCL_OK;
  
      ctable_CreatorTable	 *creator = ctable->creator;
  
      ctable_BaseRow	  *row = NULL;
      ctable_BaseRow	  *row1 = NULL;
      ctable_BaseRow	  *walkRow;
      ctable_BaseRow        *row2 = NULL;
      char	  	  *key = NULL;
  
      int			   bestScore = 0;
  
      jsw_skip_t   	  *skipList = NULL;
      int           	   skipField = 0;
      int			   inOrderWalk = 0;
  
      fieldCompareFunction_t compareFunction = NULL;
      int                    indexNumber = -1;
      int                    comparisonType = 0;
  
      int			   sortField = -1;
  
      enum walkType_e	   walkType	= WALK_DEFAULT;
  
      enum skipStart_e	   skipStart = SKIP_START_NONE;
      enum skipEnd_e	   skipEnd = SKIP_END_NONE;
      enum skipNext_e	   skipNext = SKIP_NEXT_NONE;
  
      int			   myCount;
  
      int			   inIndex = 0;
      Tcl_Obj		 **inListObj = NULL;
      ctable_BaseRow	 **inListRows = NULL;
      int			   inCount = 0;
  
      int			   canUseHash = 1;
  
      CTableSearch         *s;
  
  #ifdef WITH_SHARED_TABLES
      int			   firstTime = 1;
      int			   locked_cycle = LOST_HORIZON;
      int			   num_restarts = 0;
  
      jsw_skip_t   	  *skipListCopy = NULL;
  #endif
  
      if (search->writingTabsepIncludeFieldNames) {
  	ctable_WriteFieldNames (interp, ctable, search);
      }
  
  #ifdef WITH_SHARED_TABLES
      if (firstTime) {
  	firstTime = 0;
      } else {
  restart_search:
  	num_restarts++;
  
  	// Check for indefinite deferral
  	if(MAX_RESTARTS > 0 && num_restarts > MAX_RESTARTS) {
  	    Tcl_AppendResult (interp, "restart count exceeded", (char *) NULL);
  	    finalResult = TCL_ERROR;
  	    goto clean_and_return;
  	}
  
  	// re-initialise and de-allocate and clean up, we're going through the
  	// while exercise again...
  
          if (search->tranTable != NULL) {
  	    ckfree ((char *)search->tranTable);
  	    search->tranTable = NULL;
          }
  
          if(skipListCopy) {
  	    jsw_free_private_copy(skipListCopy);
  	    skipListCopy = NULL;
  	}
  
  
          if (ctable->share_type == CTABLE_SHARED_READER) {
  	    read_unlock(ctable->share);
  	    locked_cycle = LOST_HORIZON;
  	}
  
  	row = NULL;
  	row1 = NULL;
  	row2 = NULL;
  	key = NULL;
  	bestScore = 0;
  
  	skipList = NULL;
          skipField = 0;
          inOrderWalk = 0;
  
          compareFunction = NULL;
          indexNumber = -1;
          comparisonType = 0;
  
          sortField = -1;
  
          walkType = WALK_DEFAULT;
  
          skipStart = SKIP_START_NONE;
          skipEnd = SKIP_END_NONE;
          skipNext = SKIP_NEXT_NONE;
      
          inIndex = 0;
          inListObj = NULL;
          inCount = 0;
      }
  
      if (ctable->share_type == CTABLE_SHARED_READER) {
  	// make sure the dummy ctable is up to date.
  	locked_cycle = read_lock(ctable->share);
  
          ctable->count = ctable->share_ctable->count;
  	ctable->skipLists = ctable->share_ctable->skipLists;
  	ctable->ll_head = ctable->share_ctable->ll_head;
      }
  #endif
  
      search->matchCount = 0;
      search->alreadySearched = -1;
      if (search->tranTable != NULL) {
  	ckfree ((char *)search->tranTable);
  	search->tranTable = NULL;
      }
      search->offsetLimit = search->offset + search->limit;
  
      if (ctable->count == 0) {
  #ifdef WITH_SHARED_TABLES
          if(locked_cycle != LOST_HORIZON)
  	    read_unlock(ctable->share);
  #endif
  	if (search->cursor) {
  	    ctable_CreateCursorCommand(interp, search->cursor);
  	    Tcl_SetObjResult (interp, ctable_CursorToName (search->cursor));
  	} else if (search->cursorName) {
  	    struct cursor *cursor = ctable_CreateEmptyCursor(interp, ctable, search->cursorName);
  	    search->cursorName = NULL;
  	    ctable_CreateCursorCommand(interp, cursor);
  	    Tcl_SetObjResult (interp, ctable_CursorToName (cursor));
  	} else {
  	    Tcl_SetObjResult (interp, Tcl_NewIntObj (0));
  	}
          return TCL_OK;
      }
  
      // Check to see if we're sorting on a single field
      if (search->sortControl.nFields == 1) {
  	sortField = search->sortControl.fields[0];
      }
  
      // If there's no components in the search, make sure the index is NONE
      if(!search->nComponents) {
  	search->reqIndexField = CTABLE_SEARCH_INDEX_NONE;
      }
  
      // Check if we can use the hash table.
  #ifdef WITH_SHARED_TABLES
      // fprintf(stderr, "ctable->share_type=%d\n", ctable->share_type);
      if(ctable->share_type == CTABLE_SHARED_READER) {
  	// fprintf(stderr, "READER TABLE\n");
  	canUseHash = 0;
      }
  #endif
  
      // if they're asking for an index search, look for the best search
      // in the list of comparisons
      //
      // See the skipTypes table for the scores, but basically the tighter
      // the constraint, the better.
      //
      // If we can use a hash table, we use it, because it's either JUST
      // a simple hash lookup, or it's "in" which has to be handled here.
      //
      // If we're doing a client search, we don't have access to the
      // hashtable, so we can't do a hash search.
      if (search->reqIndexField != CTABLE_SEARCH_INDEX_NONE && search->nComponents > 0) {
  	int index = 0;
  	int trynum;
  
  	if(search->reqIndexField != CTABLE_SEARCH_INDEX_ANY) {
  	    while(index < search->nComponents) {
  	        if(search->components[index].fieldID == search->reqIndexField)
  		    break;
  		index++;
  	    }
  	}
  
  	// Look for the best usable search field starting with the requested
  	// one
  	for(trynum = 0; trynum < search->nComponents; trynum++, index++) {
  	    if(index >= search->nComponents)
  	        index = 0;
  	    CTableSearchComponent *component = &search->components[index];
  	    int field = component->fieldID;
  	    int score;
  
  	    comparisonType = component->comparisonType;
  
  	    // If it's the key, then see if it's something we can walk
  	    // using a hash.
  	    if(field == creator->keyField && canUseHash) {
  		walkType_e tryWalkType = hashTypes[comparisonType];
  
  		if (tryWalkType != WALK_DEFAULT) {
  		    walkType = tryWalkType;
  
  		    if(walkType == WALK_HASH_IN) {
  			inOrderWalk = 0; // TODO: check if list in order
  		        inListObj = component->inListObj;
  		        inCount = component->inCount;
  		    } else { //    WALK_HASH_EQ
  			inOrderWalk = 1; // degenerate case, only one result.
  			row1 = (ctable_BaseRow*) component->row1;
  			key = row1->hashEntry.key;
  		    }
  
  		    search->alreadySearched = index;
  
  		    // Always use a hash if it's available, because it's
  		    // either '=' (with only one result) or 'in' (with at
  		    // most inCount results).
  		    break;
  		}
  	    }
  
  	    // Do we have an index on this puppy?
  	    if(!ctable->skipLists[field]) {
  		continue;
  	    }
  
  	    // Special case - if it's a match and not anchored, skip
  	    if(skipTypes[comparisonType].skipNext == SKIP_NEXT_MATCH) {
  		if(component->row2 == NULL) {
  		    continue;
  		}
  	    }
  
  	    // If we have previous searches, walk back through the previous searches to see if we're already using
  	    // this field
  	    for(s = search->previousSearch; s; s = s->previousSearch) {
  		if(s->searchField == field) {
  		    break;
  		}
  	    }
  	    if(s) {
  		continue;
  	    }
  
  	    score = skipTypes[comparisonType].score;
  
  	    // Prefer to avoid sort
  	    if(field == sortField) score += SORT_SCORE;
  
  	    // We already found a better option than this one, skip it
  	    if (bestScore > score)
  		continue;
  
  	    // Got a new best candidate, save the world.
  	    skipField = field;
  	    search->searchField = field;
  
  	    skipNext  = skipTypes[comparisonType].skipNext;
  	    skipStart = skipTypes[comparisonType].skipStart;
  	    skipEnd   = skipTypes[comparisonType].skipEnd;
  
  	    search->alreadySearched = index;
  	    skipList = ctable->skipLists[field];
  	    walkType = WALK_SKIP;
  
              compareFunction = creator->fields[field]->compareFunction;
              indexNumber = creator->fields[field]->indexNumber;
  
  	    switch(skipNext) {
  	        case SKIP_NEXT_MATCH: {
  		    // For a match, we use row2 and row3 as row1 and row2,
  		    // otherwise treat it as a range
  		    inOrderWalk = 1;
  		    row1 = (ctable_BaseRow*)component->row2;
  		    row2 = component->row3;
  		    skipNext = SKIP_NEXT_ROW;
  		    break;
  	        }
  		case SKIP_NEXT_ROW: {
  		    inOrderWalk = 1;
  		    row1 = (ctable_BaseRow*) component->row1;
  		    row2 = component->row2;
  		    break;
  		}
  		case SKIP_NEXT_IN_LIST: {
  		    inOrderWalk = 0; // TODO: check if list in order
  		    inListObj = component->inListObj;
  		    inCount = component->inCount;
  		    if (ctable_CreateInRows(interp, ctable, component) == TCL_ERROR) {
  			finalResult = TCL_ERROR;
  			goto clean_and_return;
  		    }
  		    inListRows = component->inListRows;
  		    row1 = (ctable_BaseRow*) component->row1;
  		    break;
  		}
  		default: { // Can't happen
  		    Tcl_Panic("skipNext has unexpected value %d (comparisonType == %d, skipStart == %d, skipEnd == %d, score == %d)", skipNext, comparisonType, skipStart, skipEnd, score);
  		}
  	    }
  
  	    // Score of 100 means it's mandatory
  	    if (bestScore >= 100)
  	        break;
          }
      }
  
  
      // if we're sorting on the field we're searching, AND we can eliminate
      // the sort because we know we're walking in order, then eliminate the
      // sort step
      if (inOrderWalk) {
          if (search->sortControl.nFields == 1) {
  	    if(sortField == skipField) {
  		search->sortControl.nFields = 0;
  	    }
  	}
      }
  
      // Prepare transaction buffering if necessary
      ctable_PrepareTransactions(ctable, search);
  
      // Prepare for background operations
      if(search->pollInterval) search->nextPoll = search->pollInterval;
  
  #ifdef INDEXDEBUG
  fprintf(stderr, "Starting search.\n");
  fprintf(stderr, "  walkType == %d\n", walkType);
  fprintf(stderr, "  sortField == %d\n", sortField);
  fprintf(stderr, "  skipField == %d\n", skipField);
  fprintf(stderr, "  skipList == 0x%0ld\n", (long)skipList);
  if(skipList) {
  fprintf(stderr, "    skipNext == %d\n", skipNext);
  fprintf(stderr, "    skipStart == %d\n", skipStart);
  fprintf(stderr, "    skipEnd == %d\n", skipEnd);
  }
  fprintf(stderr, "  search->sortControl.nFields == %d\n", search->sortControl.nFields);
  int ii;
  fprintf(stderr, "  search->nComponents == %d\n", search->nComponents);
  for(ii = 0; ii < search->nComponents; ii++) {
  CTableSearchComponent *component = &search->components[ii];
  fprintf(stderr, "    search->components[%d].fieldID == %d\n", ii, component->fieldID);
  fprintf(stderr, "    search->components[%d].comparisonType == %d\n", ii, component->comparisonType);
  }
  #endif
  
      if (walkType == WALK_DEFAULT) {
  #ifdef INDEXDEBUG
  fprintf(stderr, "WALK_DEFAULT\n");
  #endif
  	// walk the hash table links.
  	CTABLE_LIST_FOREACH (ctable->ll_head, row, 0) {
  	    compareResult = ctable_SearchCompareRow (interp, ctable, search, row);
  	    if ((compareResult == TCL_CONTINUE) || (compareResult == TCL_OK))
  		continue;
  
  	    if (compareResult == TCL_BREAK)
  		break;
  
  	    if (compareResult == TCL_RETURN) {
  		finalResult = TCL_RETURN;
  		break;
  	    }
  
  	    if (compareResult == TCL_ERROR) {
  		finalResult = TCL_ERROR;
  		goto clean_and_return;
  	    }
          }
      } else if(walkType == WALK_HASH_EQ || walkType == WALK_HASH_IN) {
  #ifdef INDEXDEBUG
  fprintf(stderr, "WALK_HASH_*\n");
  #endif
  	// Just look up the necessary hash table values
  
  	// This loop is a little complex because for "=" the key is
  	// already set, otherwise it needs to be loaded from the index
  	// list. This would actually be simpler with a goto. :)
  	while (inIndex < inCount || key) {
  	    // If we don't have a key, get one.
  	    if (!key)
  		key = Tcl_GetString(inListObj[inIndex++]);
  
  	    // Look it up
  	    row2 = creator->find_row(ctable, key);
  
  	    // Throw away this key
  	    key = NULL;
  
  	    // If we didn't find the prize, try again
  	    if(!row2)
  		continue;
  
  	    row1 = (ctable_BaseRow *)row2;
  
  	    compareResult = ctable_SearchCompareRow (interp, ctable, search, row1);
  	    if ((compareResult == TCL_CONTINUE) || (compareResult == TCL_OK))
  		continue;
  
  	    if (compareResult == TCL_BREAK)
  		break;
  
  	    if (compareResult == TCL_RETURN) {
  		finalResult = TCL_RETURN;
  		break;
  	    }
  	    if (compareResult == TCL_ERROR) {
  		finalResult = TCL_ERROR;
  		goto clean_and_return;
  	    }
          }
      } else {
  #ifdef INDEXDEBUG
  fprintf(stderr, "WALK_SKIP\n");
  #endif
  
          //
          // walk the skip list
          //
          // all walks are "tailored", meaning the first search term is of an
          // indexed row and it's of a type where we can cut off the search
          // past a point, see if we're past the cutoff point and if we are
          // terminate the search.
          //
  #ifdef WITH_SHARED_TABLES
  	struct restart_t main_restart;
  	struct restart_t loop_restart;
  	cell_t main_cycle = LOST_HORIZON;
  	cell_t loop_cycle = LOST_HORIZON;
  
  #ifdef SANITY_CHECKS
  	creator->sanity_check_pointer(ctable, (void *)skipList, CTABLE_INDEX_NORMAL, "ctablePerformSearch : skipList");
  #endif
  
  	if(ctable->share_type == CTABLE_SHARED_READER) {
  	    // Save the cycle at the time we started the search
  	    main_cycle = ctable->share->map->cycle;
  #ifdef SANITY_CHECKS
  	    if(main_cycle == LOST_HORIZON)
  		Tcl_Panic("Master is not updating the garbage collect cycle!");
  #endif
  	    // save the main restart condition
  	    main_restart.row1 = row1;
  	    main_restart.row2 = (ctable_BaseRow*)row2;
  	    main_restart.skipStart = skipStart;
  	    main_restart.skipEnd = skipEnd;
  	    main_restart.compareFunction = compareFunction;
  
  	    // clone the skiplist
  	    skipListCopy = jsw_private_copy(skipList, getpid(), compareFunction);
  	    if(skipListCopy)
  		skipList = skipListCopy;
  	}
  #endif
  
  	// Find the row to start walking on
  	switch(skipStart) {
  
  	    case SKIP_START_EQ_ROW1: {
  if(!row1) Tcl_Panic("Can't happen! Row1 is null for '=' comparison.");
  		jsw_sfind (skipList, row1);
  		break;
  	    }
  
  	    case SKIP_START_GE_ROW1: {
  if(!row1) Tcl_Panic("Can't happen! Row1 is null for '>=' comparison.");
  		jsw_sfind_equal_or_greater (skipList, row1);
  		break;
  	    }
  
  	    case SKIP_START_GT_ROW1: {
  if(!row1) Tcl_Panic("Can't happen! Row1 is null for '>' comparison.");
  		jsw_sfind_equal_or_greater (skipList, row1);
  		while (1) {
                      if ((row = jsw_srow (skipList)) == NULL)
  		        goto search_complete;
  	    	    if (compareFunction (row, row1) > 0)
  			break;
  		    jsw_snext(skipList);
  		}
  		break;
  	    }
  
  	    case SKIP_START_RESET: {
  		jsw_sreset(skipList);
  		break;
  	    }
  
  	    default: { // can't happen
  		Tcl_Panic("skipStart has unexpected value %d", skipStart);
  	    }
  	}
  
  #ifdef MEGADEBUG
  #ifdef WITH_SHARED_TABLES
  if(ctable->share_type == CTABLE_SHARED_READER)
      fprintf(stderr, "Following skiplist 0x%lX\n", (long)skipList);
  #endif
  #endif
  
  	// Count is only to make sure we blow up on infinite loops
  	// It's compared less than OR equal to the largest possible value
  	// it could have.
  	for(myCount = 0; myCount <= ctable->count; myCount++) {
  	    if(skipNext == SKIP_NEXT_IN_LIST) {
  		// We're looking at a list of entries rather than a range,
  		// so we have to loop here searching for each in turn instead
  		// of just following the skiplist
  		while(1) {
  		  // If we're at the end, game over.
  		  if(inIndex >= inCount)
  		      goto search_complete;
  
  #ifdef WITH_SHARED_TABLES
  		  if(main_cycle != LOST_HORIZON) {
  	            // Save the cycle at the time we started the walk
  	            main_cycle = ctable->share->map->cycle;
  		  }
  #endif
  
  		  row = inListRows[inIndex++];
  if(!row) Tcl_Panic("Can't happen, null row in 'in' comparison");
  
  		  // If there's a match for this row, break out of the loop
                    if (jsw_sfind (skipList, row) != NULL)
  		      break;
  	        }
  	    }
  
  	    // Now we can fetch whatever we found.
              if ((row = jsw_srow (skipList)) == NULL)
  		goto search_complete;
  #ifdef SANITY_CHECKS
  	creator->sanity_check_pointer(ctable, (void *)row, CTABLE_INDEX_NORMAL, "ctablePerformSearch : row");
  #endif
  
  #ifdef WITH_SHARED_TABLES
  	    // If we're a reader and this row has changed since we started
  	    // then check if it changed the skiplist we're following, if so...
  	    // go back and restart the search
  	    if(main_cycle != LOST_HORIZON) {
  	        // Save the cycle at the time we started the loop
  	        loop_cycle = ctable->share->map->cycle;
  		if(row->_row_cycle != LOST_HORIZON) {
  		    int delta = row->_row_cycle - main_cycle;
  		    if(delta > 0) {
  		        if(ctable_SearchRestartNeeded(row, &main_restart)) {
  #ifdef MEGADEBUG
  if(num_restarts == 0) fprintf(stderr, "%d: main restart: main_cycle=%ld; row->_row_cycle=%ld; delta %d\n", getpid(), (long)main_cycle, (long)row->_row_cycle, delta);
  #endif
  			    goto restart_search;
  			}
  		    }
  #ifdef SANITY_CHECKS
  		} else {
  		    Tcl_Panic("Master is not copying the garbage collect cycle to the row!");
  #endif
  		}
  
  	        // save the loop restart condition - exact match for this field
  	        loop_restart.row1 = row;
  	        loop_restart.row2 = row;
  	        loop_restart.skipStart = SKIP_START_EQ_ROW1;
  	        loop_restart.skipEnd = SKIP_END_NE_ROW1;
  	        loop_restart.compareFunction = compareFunction;
  	    }
  #endif
  
  	    // if at end or past any terminating condition, break
  	    switch(skipEnd) {
  	        case SKIP_END_GE_ROW1: {
  		    if (compareFunction (row, row1) >= 0)
  			goto search_complete;
  		    break;
  		}
  	        case SKIP_END_GT_ROW1: {
  		    if (compareFunction (row, row1) > 0)
  			goto search_complete;
  		    break;
  		}
  	        case SKIP_END_GE_ROW2: {
  		    if (compareFunction (row, row2) >= 0)
  			goto search_complete;
  		    break;
  		}
  		default: {
  		    // may not be a terminating condition, or it may
  		    // have been taken care of above
  	        }
  	    }
  
              // walk walkRow through the linked list of rows off this skip list
  	    // node. This is not the safe foreach routine because we don't
  	    // change the skiplist during the search, and if someone else
  	    // does it we need to restart the transaction anyway.
  
              CTABLE_LIST_FOREACH (row, walkRow, indexNumber) {
  
  #ifdef WITH_SHARED_TABLES
  		// If we're a reader and this row has changed since we started
  		// then check if it changed the list we're following, if so...
  		// go back and restart the search
  	        if(loop_cycle != LOST_HORIZON) {
  		    if(row->_row_cycle != LOST_HORIZON) {
  		        int delta = row->_row_cycle - main_cycle;
  		        if(delta > 0) {
  		            if(ctable_SearchRestartNeeded(row, &loop_restart)) {
  #ifdef MEGADEBUG
  if(num_restarts == 0) fprintf(stderr, "%d: loop restart: loop_cycle=%ld; row->_row_cycle=%ld; delta=%d\n", getpid(), (long)loop_cycle, (long)row->_row_cycle, delta);
  #endif
  			        goto restart_search;
  			    }
  		        }
  #ifdef SANITY_CHECKS
  		    } else {
  		        Tcl_Panic("Master is not copying the garbage collect cycle to the row!");
  #endif
  		    }
  	        }
  #endif
  
  	        compareResult = ctable_SearchCompareRow (interp, ctable, search, walkRow);
  	        if ((compareResult == TCL_CONTINUE) || (compareResult == TCL_OK)) continue;
  
  	        if (compareResult == TCL_BREAK)
  		    goto search_complete;
  
  		if (compareResult == TCL_RETURN) {
  		    finalResult = TCL_RETURN;
  	            goto search_complete;
  	        }
  
  	        if (compareResult == TCL_ERROR) {
  	            finalResult = TCL_ERROR;
  	            goto clean_and_return;
  	        }
  	    }
  
  	    if(skipNext == SKIP_NEXT_ROW)
  		jsw_snext(skipList);
  	}
  	// Should never just fall out of the loop...
  	Tcl_AppendResult (interp, "infinite search loop", (char *) NULL);
  	finalResult = TCL_ERROR;
  	goto clean_and_return;
      }
  
    // We only jump to this on success, so we got to the end of the loop
    // or we broke out of it early
    search_complete:
  
      // We're no longer walking a skiplist, so make a note of that so it can be re-used.
      search->searchField = -1;
  
      switch (ctable_PostSearchCommonActions (interp, ctable, search)) {
  	case TCL_ERROR: {
  	    finalResult = TCL_ERROR;
  	    break;
  	}
  	case TCL_RETURN: {
  	    finalResult = TCL_RETURN;
  	    break;
  	}
      }
  
    // We only jump to this on an error
    clean_and_return:
      if (search->tranTable != NULL) {
  	ckfree ((char *)search->tranTable);
  	search->tranTable = NULL;
      }
  
      if (finalResult != TCL_ERROR && (search->codeBody == NULL || finalResult != TCL_RETURN)) {
  	if(search->cursor) {
  	    // We got here so we can create the command
  	    ctable_CreateCursorCommand(interp, search->cursor);
  	    Tcl_SetObjResult (interp, ctable_CursorToName (search->cursor));
  	} else if (search->cursorName) {
  	    struct cursor *cursor = ctable_CreateEmptyCursor(interp, ctable, search->cursorName);
  	    search->cursorName = NULL;
  	    ctable_CreateCursorCommand(interp, cursor);
  	    Tcl_SetObjResult (interp, ctable_CursorToName (cursor));
  	} else {
  	    Tcl_SetObjResult (interp, Tcl_NewIntObj (search->matchCount));
  	}
      }
  
  #ifdef WITH_SHARED_TABLES
      if(search->cursor)
  	search->cursor->lockCycle = locked_cycle;
      else if(locked_cycle != LOST_HORIZON)
  	read_unlock(ctable->share);
  
      if(skipListCopy)
  	jsw_free_private_copy(skipListCopy);
  
  #ifdef MEGADEBUG
  if(num_restarts) fprintf(stderr, "%d: Restarted search %d times\n", getpid(), num_restarts);
  #endif
  #endif
  
      return finalResult;
  }
  
  //
  // ctable_SetupSearch - prepare to search by parsing the command line arguments
  // specified when the ctables "search" method is invoked.
  //
  //
  static int
  ctable_SetupSearch (Tcl_Interp *interp, CTable *ctable, Tcl_Obj *CONST objv[], int objc, CTableSearch *search, int indexField, CTableSearch *previous_search) {
      int             i;
      int             searchTerm = 0;
      CONST char    **fieldNames = ctable->creator->fieldNames;
      int		    quick_count;
  
      static int staticSequence = 0;
  
      static CONST char *searchOptions[] = {"-array", "-array_with_nulls", "-array_get", "-array_get_with_nulls", "-code", "-compare", "-countOnly", "-fields", "-get", "-glob", "-key", "-with_field_names", "-limit", "-nokeys", "-offset", "-sort", "-write_tabsep", "-tab", "-delete", "-update", "-buffer", "-index", "-poll_code", "-poll_interval", "-quote", "-null", "-filter", "-cursor", (char *)NULL};
  
      enum searchOptions {SEARCH_OPT_ARRAY_NAMEOBJ, SEARCH_OPT_ARRAYWITHNULLS_NAMEOBJ, SEARCH_OPT_ARRAYGET_NAMEOBJ, SEARCH_OPT_ARRAYGETWITHNULLS_NAMEOBJ, SEARCH_OPT_CODE, SEARCH_OPT_COMPARE, SEARCH_OPT_COUNTONLY, SEARCH_OPT_FIELDS, SEARCH_OPT_GET_NAMEOBJ, SEARCH_OPT_GLOB, SEARCH_OPT_KEYVAR_NAMEOBJ, SEARCH_OPT_WITH_FIELD_NAMES, SEARCH_OPT_LIMIT, SEARCH_OPT_DONT_INCLUDE_KEY, SEARCH_OPT_OFFSET, SEARCH_OPT_SORT, SEARCH_OPT_WRITE_TABSEP, SEARCH_OPT_TAB, SEARCH_OPT_DELETE, SEARCH_OPT_UPDATE, SEARCH_OPT_BUFFER, SEARCH_OPT_INDEX, SEARCH_OPT_POLL_CODE, SEARCH_OPT_POLL_INTERVAL, SEARCH_OPT_QUOTE_TYPE, SEARCH_OPT_NULL_STRING, SEARCH_OPT_FILTER, SEARCH_OPT_CURSOR};
      if (objc < 2) {
        wrong_args:
  	Tcl_WrongNumArgs (interp, 2, objv, "?-array_get varName? ?-array_get_with_nulls varName? ?-code codeBody? ?-compare list? ?-filter list? ?-countOnly 0|1? ?-fields fieldList? ?-get varName? ?-glob pattern? ?-key varName? ?-with_field_names 0|1?  ?-limit limit? ?-nokeys 0|1? ?-offset offset? ?-sort {?-?field1..}? ?-write_tabsep channel? ?-tab value? ?-delete 0|1? ?-update {fields value...}? ?-buffer 0|1? ?-poll_interval interval? ?-poll_code codeBody? ?-quote type?");
  	return TCL_ERROR;
      }
  
      // Quick count of table rows for early optimizations where ONLY the
      // number of rows is being used. This quick_count is a snapshot that
      // is safe to use for shared readers because reading it is an atomic
      // operation.
  #ifdef WITH_SHARED_TABLES
      if (ctable->share_type == CTABLE_SHARED_READER)
  	quick_count = ctable->share_ctable->count;
      else
  #endif
  	quick_count = ctable->count;
  
  // TODO figure out how to handle the case where a cursor command is being used. In the meantime we can't use this shortcut
  #if 0
      // if there are no rows in the table, the search won't turn up
      // anything, so skip all that
      if (quick_count == 0)
      {
  	search->filters = NULL;
  	search->components = NULL; // keep ctable_searchTeardown happy
  	Tcl_SetObjResult (interp, Tcl_NewIntObj (0));
  	return TCL_RETURN;
      }
  #endif
  
      // initialize search control structure
      search->ctable = ctable;
      search->previousSearch = previous_search;
      search->action = CTABLE_SEARCH_ACTION_NONE;
      search->nComponents = 0;
      search->components = NULL;
      search->nFilters = 0;
      search->filters = NULL;
      search->countMax = 0;
      search->offset = 0;
      search->limit = 0;
      search->pattern = NULL;
      search->pollInterval = 0;
      search->nextPoll = -1;
      search->pollCodeBody = NULL;
      search->sortControl.fields = NULL;
      search->sortControl.directions = NULL;
      search->sortControl.nFields = 0;
      search->retrieveFields = NULL;
      search->nRetrieveFields = -1;   // -1 = all, 0 = none
      search->noKeys = 0;
      search->rowVarNameObj = NULL;
      search->keyVarNameObj = NULL;
      search->codeBody = NULL;
      search->writingTabsepIncludeFieldNames = 0;
      search->tranType = CTABLE_SEARCH_TRAN_NONE;
      search->tranData = NULL;
      search->reqIndexField = indexField;
      search->bufferResults = CTABLE_BUFFER_DEFAULT;
      search->sepstr = "\t";
      search->nullString = NULL;
      search->quoteType = CTABLE_QUOTE_NONE;
      search->matchCount = 0;
      search->alreadySearched = -1;
      search->tranTable = NULL;
      search->offsetLimit = search->offset + search->limit;
      search->cursorName = NULL;
      search->cursor = NULL;
      search->searchField = -1;
  
      // Give each search a unique non-zero sequence number
      if(++staticSequence == 0) ++staticSequence;
      search->sequence = staticSequence;
  
      for (i = 2; i < objc; ) {
  	if (Tcl_GetIndexFromObj (interp, objv[i++], searchOptions, "search option", TCL_EXACT, &searchTerm) != TCL_OK) {
  	    return TCL_ERROR;
  	}
  
  	//  all the arguments require one parameter
  	if (i >= objc) {
  	    goto wrong_args;
  	}
  
  	switch (searchTerm) {
  	  case SEARCH_OPT_SORT: {
  	    // the fields they want us to sort on
  	    if (ctable_ParseSortFieldList (interp, objv[i++], fieldNames, &search->sortControl) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing sort options", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_FIELDS: {
  	    // the fields they want us to retrieve
  	    if (ctable_ParseFieldList (interp, objv[i++], fieldNames, &search->retrieveFields, &search->nRetrieveFields) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search fields", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    break;
  	  }
  
  	  case SEARCH_OPT_DONT_INCLUDE_KEY: {
  	    if (Tcl_GetBooleanFromObj (interp, objv[i++], &search->noKeys) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search -nokeys", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_INDEX: {
  	    if (Tcl_GetIndexFromObj (interp, objv[i++], fieldNames, "field", TCL_EXACT, &search->reqIndexField) != TCL_OK) {
  		Tcl_AppendResult (interp, " while processing search index", (char *) NULL);
  		return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_WITH_FIELD_NAMES: {
  	    if (Tcl_GetBooleanFromObj (interp, objv[i++], &search->writingTabsepIncludeFieldNames) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search -with_field_names", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_ARRAY_NAMEOBJ: {
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  	    search->rowVarNameObj = objv[i++];
  	    search->action = CTABLE_SEARCH_ACTION_ARRAY;
  	    break;
  	  }
  
  	  case SEARCH_OPT_ARRAYWITHNULLS_NAMEOBJ: {
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  	    search->rowVarNameObj = objv[i++];
  	    search->action = CTABLE_SEARCH_ACTION_ARRAY_WITH_NULLS;
  	    break;
  	  }
  
  	  case SEARCH_OPT_ARRAYGET_NAMEOBJ: {
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  	    search->rowVarNameObj = objv[i++];
  	    search->action = CTABLE_SEARCH_ACTION_ARRAY_GET;
  	    break;
  	  }
  
  	  case SEARCH_OPT_ARRAYGETWITHNULLS_NAMEOBJ: {
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  	    search->rowVarNameObj = objv[i++];
  	    search->action = CTABLE_SEARCH_ACTION_ARRAY_GET_WITH_NULLS;
  	    break;
  	  }
  
  	  case SEARCH_OPT_KEYVAR_NAMEOBJ: {
  	    search->keyVarNameObj = objv[i++];
  	    break;
            }
  
  	  case SEARCH_OPT_GET_NAMEOBJ: {
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  	    search->rowVarNameObj = objv[i++];
  	    search->action = CTABLE_SEARCH_ACTION_GET;
  	    break;
            }
  
  	  case SEARCH_OPT_GLOB: {
  	    search->pattern = Tcl_GetString (objv[i++]);
  	    break;
  	  }
  
  	  case SEARCH_OPT_COMPARE: {
  	    if (ctable_ParseSearch (interp, ctable, objv[i++], fieldNames, search) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search compare", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_FILTER: {
  	    if (ctable_ParseFilters (interp, ctable, objv[i++], search) == TCL_ERROR) {
  		Tcl_AppendResult (interp, " while processing search filter", (char *)NULL);
  		return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_COUNTONLY: {
  #if 0
  	    int countOnly;
  
  	    if (Tcl_GetBooleanFromObj (interp, objv[i++], &countOnly) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search countOnly", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    if (countOnly) {
  		if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		    goto actionOverload;
  		search->action = CTABLE_SEARCH_ACTION_COUNT;
  	    }
  #else
  	    i++; // skip argument
  #endif
  	    break;
  	  }
  
  	  case SEARCH_OPT_OFFSET: {
  	    if (Tcl_GetIntFromObj (interp, objv[i++], &search->offset) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search offset", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    if (search->offset < 0) {
  	        Tcl_AppendResult (interp, "Search offset cannot be negative", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_LIMIT: {
  	    if (Tcl_GetIntFromObj (interp, objv[i++], &search->limit) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing search limit", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    if (search->limit < 0) {
  	        Tcl_AppendResult (interp, "Search limit cannot be negative", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_CODE: {
  	      search->codeBody = objv[i++];
  	      break;
  	  }
  
  	  case SEARCH_OPT_POLL_CODE: {
  	      search->pollCodeBody = objv[i++];
  	      if (!search->pollInterval)
  		  search->pollInterval = CTABLE_DEFAULT_POLL_INTERVAL;
  	      break;
  	  }
  
  	  case SEARCH_OPT_POLL_INTERVAL: {
  	    int poll_interval;
  	    if (Tcl_GetIntFromObj (interp, objv[i++], &poll_interval) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing poll_interval option", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    search->pollInterval = poll_interval;
  	    break;
  	  }
  	
  	  case SEARCH_OPT_DELETE: {
  	    int do_delete;
  	    if (Tcl_GetIntFromObj (interp, objv[i++], &do_delete) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing delete option", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    if(do_delete) {
  	      if(ctable->cursors) {
  	        Tcl_AppendResult(interp, "Can not delete while cursors are active.", NULL);
  	        Tcl_SetErrorCode (interp, "speedtables", "no_delete_with_cursors", NULL);
  	        return TCL_ERROR;
  	      }
  	      if(previous_search) {
  	        Tcl_AppendResult(interp, "Can not delete in nested search.", NULL);
  	        Tcl_SetErrorCode (interp, "speedtables", "no_delete_inside_search", NULL);
  	        return TCL_ERROR;
  	      }
  #ifdef WITH_SHARED_TABLES
  	      if(ctable->share_type == CTABLE_SHARED_READER) {
  		Tcl_AppendResult (interp, "Can't modify read-only tables.", (char *)NULL);
  		return TCL_ERROR;
  	      }
  #endif
  	      if(search->tranType != CTABLE_SEARCH_TRAN_NONE &&
  	         search->tranType != CTABLE_SEARCH_TRAN_DELETE
  	      ) {
  		Tcl_AppendResult (interp, "Can not combine -delete with other transaction options", (char *)NULL);
  		return TCL_ERROR;
  	      }
  	      search->tranType = CTABLE_SEARCH_TRAN_DELETE;
  	    } else {
  	      if(search->tranType == CTABLE_SEARCH_TRAN_DELETE)
  		search->tranType = CTABLE_SEARCH_TRAN_NONE;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_UPDATE: {
  #ifdef WITH_SHARED_TABLES
  	    if(ctable->share_type == CTABLE_SHARED_READER) {
  		Tcl_AppendResult (interp, "Can't modify read-only tables.", (char *)NULL);
  		return TCL_ERROR;
  	    }
  #endif
  	    if(search->tranType != CTABLE_SEARCH_TRAN_NONE &&
  	       search->tranType != CTABLE_SEARCH_TRAN_UPDATE
  	    ) {
  		Tcl_AppendResult (interp, "Can not combine -update with other transaction options", (char *)NULL);
  		return TCL_ERROR;
  	    }
  	    search->tranType = CTABLE_SEARCH_TRAN_UPDATE;
  	    search->tranData = objv[i++];
  	    break;
  	  }
  
  	  case SEARCH_OPT_BUFFER: {
  	    int do_buffer;
  	    if (Tcl_GetIntFromObj (interp, objv[i++], &do_buffer) == TCL_ERROR) {
  	        Tcl_AppendResult (interp, " while processing buffer option", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    search->bufferResults = do_buffer;
  	    break;
  	  }
  
  	  case SEARCH_OPT_TAB: {
  	    search->sepstr = Tcl_GetString(objv[i++]);
  	    break;
  	  }
  
  	  case SEARCH_OPT_NULL_STRING: {
  	    search->nullString = Tcl_GetString(objv[i++]);
  	    break;
  	  }
  
  	  case SEARCH_OPT_QUOTE_TYPE: {
  	    search->quoteType = ctable_parseQuoteType(interp, objv[i++]);
  	    if(search->quoteType < 0) {
  		Tcl_AppendResult (interp, "in argument to -quote", NULL);
  		return TCL_ERROR;
  	    }
  	    break;
  	  }
  
  	  case SEARCH_OPT_WRITE_TABSEP: {
  	    int        mode;
  	    char      *channelName;
  
  	    if (search->action != CTABLE_SEARCH_ACTION_NONE)
  		goto actionOverload;
  
  	    channelName = Tcl_GetString (objv[i++]);
  	    if ((search->tabsepChannel = Tcl_GetChannel (interp, channelName, &mode)) == NULL) {
  	        Tcl_AppendResult (interp, " while processing write_tabsep channel", (char *) NULL);
  	        return TCL_ERROR;
  	    }
  
  	    if (!(mode & TCL_WRITABLE)) {
  		Tcl_AppendResult (interp, "channel \"", channelName, "\" not writable", (char *)NULL);
  		return TCL_ERROR;
  	    }
  
  	    search->action = CTABLE_SEARCH_ACTION_WRITE_TABSEP;
  
  	    break;
  	  }
  
  	  case SEARCH_OPT_CURSOR: {
  	    char *cursorName = NULL;
  	    struct cursor *c;
  
              if(search->tranType != CTABLE_SEARCH_TRAN_NONE || search->action != CTABLE_SEARCH_ACTION_NONE) {
  		Tcl_AppendResult (interp, "Can not combine -cursor with other operations", (char *)NULL);
  	    }
  
  	    cursorName = Tcl_GetString (objv[i++]);
  
  	    if (strcmp (cursorName, "#auto") == 0) {
  		char *tableName = NULL;
  		int tableNameLength;
  		static unsigned long int auto_cursor_id = 0;
  
  		// use command name of the ctable as the base of the cursor name
  		tableName = Tcl_GetStringFromObj (objv[0], &tableNameLength);
  		tableNameLength += 42+2;
  		cursorName = (char *) ckalloc (tableNameLength);
  		snprintf(cursorName, tableNameLength, "%s_C%lu", tableName, ++auto_cursor_id);
  	    } else {
  		char *tmp = (char *)ckalloc (strlen(cursorName) + 1);
  		strcpy(tmp, cursorName);
  		cursorName = tmp;
  	    }
  
  	    for (c = ctable->cursors; c; c = c->nextCursor) {
  		if(strcmp(c->cursorName, cursorName) == 0) {
  		    Tcl_AppendResult (interp, "Cursor name must not duplicate an existing cursor on the same table", (char *)NULL);
  		    ckfree(cursorName);
  	            return TCL_ERROR;
  		}
  	    }
  
  	    search->tranType = CTABLE_SEARCH_TRAN_CURSOR;
  	    search->action = CTABLE_SEARCH_ACTION_CURSOR;
  	    search->cursorName = cursorName;
  	  }
  	}
      }
  
      // If we have a code body, make sure we're not doing a write_tabsep or returning a cursor, make
      // sure we have a row variable or a key variable, and that we're not
      // leaving the search action "none"
      if (search->codeBody != NULL) {
  	if (search->action == CTABLE_SEARCH_ACTION_WRITE_TABSEP || search->action == CTABLE_SEARCH_ACTION_CURSOR) {
  	    Tcl_AppendResult (interp, "Both -code and -write_tabsep or -cursor specified", (char *)NULL);
  	    goto errorReturn;
  	}
  	if (search->rowVarNameObj == NULL && search->keyVarNameObj == NULL) {
  	    Tcl_AppendResult (interp, "Code block specified, but none of -key, -get, -array, -array_get, -array_with_nulls, or -array_get_with_nulls provided", NULL);
  	    goto errorReturn;
  	}
  	if(search->action == CTABLE_SEARCH_ACTION_NONE)
  	    search->action = CTABLE_SEARCH_ACTION_CODE;
      }
  
      // If we're doing a transaction, make sure we're not leaving the search
      // action "none"
      if(search->tranType != CTABLE_SEARCH_TRAN_NONE) {
  	if(search->action == CTABLE_SEARCH_ACTION_NONE)
  	    search->action = CTABLE_SEARCH_ACTION_TRANSACTION_ONLY;
      }
  
      // If there's nothing going on in the search, then skip the search and
      // return quick_count (calculated earlier).
      if(search->action == CTABLE_SEARCH_ACTION_NONE) {
  	if(search->nComponents == 0 && search->nFilters == 0 && search->nRetrieveFields <= 0 && search->codeBody == NULL && search->pattern == NULL && search->rowVarNameObj == NULL && search->keyVarNameObj == NULL) {
  
  	    if (search->offset) {
  		quick_count -= search->offset;
  		if (quick_count < 0) {
  		    quick_count = 0;
  		}
  	    }
  
  	    if (search->limit) {
  		if (quick_count > search->limit) {
  		    quick_count = search->limit;
  		}
  	    }
  
  	    Tcl_SetObjResult (interp, Tcl_NewIntObj (quick_count));
  	    return TCL_RETURN;
  	}
      }
  
      if (search->action == CTABLE_SEARCH_ACTION_WRITE_TABSEP) {
  	ctable_checkForKey(ctable, search);
      } else {
  	if(search->writingTabsepIncludeFieldNames) {
  	    Tcl_AppendResult (interp, "can't use -with_field_names without -write_tabsep", (char *) NULL);
  	    return TCL_ERROR;
  	}
  
          if (search->codeBody == NULL && search->sortControl.nFields) {
  	    Tcl_AppendResult (interp, "Sorting must be accompanied by -code or -write_tabsep", (char *) NULL);
  	    return TCL_ERROR;
          }
      }
  
      return TCL_OK;
  
    actionOverload: 
  
      Tcl_AppendResult (interp, "only one of -array, -array_with_nulls, -array_get, -array_get_with_nulls, or -write_tabsep must be specified", (char *) NULL);
  
    errorReturn:
  
      return TCL_ERROR;
  }
  
  //
  // ctable_elapsed_time - calculate the time interval between two timespecs and store in a new
  // timespec
  //
  void
  ctable_elapsed_time (struct timespec *oldtime, struct timespec *newtime, struct timespec *elapsed) {
      elapsed->tv_sec = newtime->tv_sec - oldtime->tv_sec;
      elapsed->tv_nsec = newtime->tv_nsec - oldtime->tv_nsec;
  
      if (elapsed->tv_nsec < 0) {
  	elapsed->tv_nsec += 1000000000;
  	elapsed->tv_sec--;
      }
  }
  
  #ifdef CTABLES_CLOCK
  //
  // ctable_performance_callback - callback routine for performance of search calls
  //
  void
  ctable_performance_callback (Tcl_Interp *interp, CTable *ctable, Tcl_Obj *CONST objv[], int objc, struct timespec *startTimeSpec, int loggingMatchCount) {
      struct timespec endTimeSpec;
      struct timespec elapsedTimeSpec;
      Tcl_Obj *cmdObjv[4];
      double cpu;
      int i;
  
      // calculate elapsed cpu
  
      clock_gettime (CTABLES_CLOCK, &endTimeSpec);
      ctable_elapsed_time (startTimeSpec, &endTimeSpec, &elapsedTimeSpec);
      cpu = (elapsedTimeSpec.tv_sec + (elapsedTimeSpec.tv_nsec / 1000000000.0));
  
      if (cpu < ctable->performanceCallbackThreshold) {
  	return;
      }
  
      cmdObjv[0] = Tcl_NewStringObj (ctable->performanceCallback, -1);
      cmdObjv[1] = Tcl_NewListObj (objc, objv);
      cmdObjv[2] = Tcl_NewIntObj (loggingMatchCount);
      cmdObjv[3] = Tcl_NewDoubleObj (cpu);
  
      for (i = 0; i < 4; i++) {
  	Tcl_IncrRefCount (cmdObjv[i]);
      }
  
      if (Tcl_EvalObjv (interp, 4, cmdObjv, 0) == TCL_ERROR) {
  	Tcl_BackgroundError (interp);
      }
  
      for (i = 0; i < 4; i++) {
  	Tcl_DecrRefCount (cmdObjv[i]);
      }
  
  }
  #endif
  
  //
  // ctable_TeardownSearch - tear down (free) a search structure and the
  //  stuff within it.
  //
  static void
  ctable_TeardownSearch (CTableSearch *search) {
      int i;
  
      if (search->filters) {
  	ckfree((char*)search->filters);
  	search->filters = NULL;
      }
  
      if (search->components == NULL) {
          return;
      }
  
      // teardown components
      for (i = 0; i < search->nComponents; i++) {
  	CTableSearchComponent  *component = &search->components[i];
  
  	if (component->row1 != NULL) {
  	    search->ctable->creator->delete_row (search->ctable, component->row1, CTABLE_INDEX_PRIVATE);
  	}
  
  	if (component->row2 != NULL) {
  	    search->ctable->creator->delete_row (search->ctable, component->row2, CTABLE_INDEX_PRIVATE);
  	}
  
  	if (component->row3 != NULL) {
  	    search->ctable->creator->delete_row (search->ctable, component->row3, CTABLE_INDEX_PRIVATE);
  	}
  
  	if (component->clientData != NULL) {
  	    // this needs to be pluggable
  	    if ((component->comparisonType == CTABLE_COMP_MATCH) || (component->comparisonType == CTABLE_COMP_NOTMATCH) || (component->comparisonType == CTABLE_COMP_MATCH_CASE) || (component->comparisonType == CTABLE_COMP_NOTMATCH_CASE)) {
  		struct ctableSearchMatchStruct *sm = (struct ctableSearchMatchStruct*) component->clientData;
  		if (sm->type == CTABLE_STRING_MATCH_UNANCHORED) {
  		    boyer_moore_teardown (sm);
  		}
  	    }
  
  	    ckfree ((char*)component->clientData);
  	}
  
  	ctable_FreeInRows(search->ctable, component);
      }
  
      ckfree ((char *)search->components);
      search->components = NULL;
  
      if (search->sortControl.fields != NULL) {
          ckfree ((char *)search->sortControl.fields);
  	search->sortControl.fields = NULL;
          ckfree ((char *)search->sortControl.directions);
  	search->sortControl.directions = NULL;
      }
  
      if (search->retrieveFields != NULL) {
  	ckfree ((char *)search->retrieveFields);
  	search->retrieveFields = NULL;
      }
  
      if (search->tranTable != NULL) {
          ckfree ((char *)search->tranTable);
          search->tranTable = NULL;
      }
  }
  
  //
  // ctable_SetupAndPerformSearch - setup and perform a (possibly) skiplist search
  //   on a table.
  //
  // Uses a skip list index as the outer loop.  Still brute force unless the
  // foremost compare routine is tailorable, however even so, much faster
  // than a hash table walk.
  //
  //
  CTABLE_INTERNAL int
  ctable_SetupAndPerformSearch (Tcl_Interp *interp, Tcl_Obj *CONST objv[], int objc, CTable *ctable, int indexField) {
      CTableSearch    search;
      int result;
  #ifdef CTABLES_CLOCK
      struct timespec startTimeSpec;
      int loggingMatchCount = 0;
  #endif
  
  
  #ifdef CTABLES_CLOCK
      if (ctable->performanceCallbackEnable) {
  	clock_gettime (CTABLES_CLOCK, &startTimeSpec);
      }
  #endif
  
      // flag this search in progress
      CTableSearch *previous_search = ctable->searches;
      ctable->searches = &search;
  
      result = ctable_SetupSearch (interp, ctable, objv, objc, &search, indexField, previous_search);
      if (result == TCL_ERROR) {
          ctable->searches = previous_search;
          return TCL_ERROR;
      }
  
      // return from "setup" means "search optimized away"
      if (result == TCL_RETURN) {
  	result = TCL_OK;
      } else {
          result = ctable_PerformSearch (interp, ctable, &search);
      }
  
  #ifdef CTABLES_CLOCK
      if (ctable->performanceCallbackEnable) {
  	loggingMatchCount = search.matchCount;
      }
  #endif
  
      ctable_TeardownSearch (&search);
  
      ctable->searches = previous_search;
  
  #ifdef CTABLES_CLOCK
      if (ctable->performanceCallbackEnable) {
  	Tcl_Obj *saveResultObj = Tcl_GetObjResult (interp);
  	Tcl_IncrRefCount (saveResultObj);
  	ctable_performance_callback (interp, ctable, objv, objc, &startTimeSpec, loggingMatchCount);
  	Tcl_SetObjResult (interp, saveResultObj);
      }
  #endif
  
      return result;
  }
  
  //
  // ctable_DropIndex - delete all the rows in a row's index, free the
  // structure and set the field's pointer to the skip list to NULL
  //
  // "final" means "we're destroying the ctable". This allows us to avoid
  // deleting structures in shared memory that are going away anyway.
  //
  CTABLE_INTERNAL void
  ctable_DropIndex (CTable *ctable, int field, int final) {
      jsw_skip_t *skip = ctable->skipLists[field];
      ctable_BaseRow *row;
      int listIndexNumber = ctable->creator->fields[field]->indexNumber;
  
      if (skip == NULL) return;
  
      // Forget I had a skiplist
      ctable->skipLists[field] = NULL;
  
      // Delete the skiplist
      jsw_sdelete_skiplist (skip, final);
  
      // Don't need to reset the bucket list if it's just going to be deleted
      if(final) return;
  
      // Walk the table and erase the bucket lists for the row
      // We're walking the main index for the table so it's safe to
      // clear the index for $field
      CTABLE_LIST_FOREACH (ctable->ll_head, row, 0) {
  	row->_ll_nodes[listIndexNumber].next = NULL;
  	row->_ll_nodes[listIndexNumber].prev = NULL;
  	row->_ll_nodes[listIndexNumber].head = NULL;
      }
  }
  
  //
  // ctable_DropAllIndexes - delete all of a table's indexes
  //
  CTABLE_INTERNAL void
  ctable_DropAllIndexes (CTable *ctable, int final) {
      int field;
  
      for (field = 0; field < ctable->creator->nFields; field++) {
          ctable_DropIndex (ctable, field, final);
      }
  }
  
  //
  // ctable_IndexCount -- set the Tcl interpreter obj result to the
  //                      number of items in the index
  //
  // mostly just going to be used as a cross-check for testing to make sure
  // inserts and deletes into indexes corresponding to changes in rows works
  // properly
  //
  CTABLE_INTERNAL int
  ctable_IndexCount (Tcl_Interp *interp, CTable *ctable, int field) {
      jsw_skip_t *skip = ctable->skipLists[field];
      int         count;
  
      if (skip == NULL) {
  	Tcl_AppendResult (interp, "that field does not have an index", (char *) NULL);
  	return TCL_ERROR;
      }
  
      count = (int)jsw_ssize(skip);
      Tcl_SetObjResult (interp, Tcl_NewIntObj (count));
      return TCL_OK;
  }
  
  CTABLE_INTERNAL int
  ctable_DumpIndex (CTable *ctable, int field) {
      jsw_skip_t *skip = ctable->skipLists[field];
      ctable_BaseRow *row;
      Tcl_Obj    *utilityObj = Tcl_NewObj ();
      CONST char *s;
  
      if (skip == NULL) {
          return TCL_OK;
      }
  
      jsw_dump_head (skip);
  
      for (jsw_sreset (skip); (row = jsw_srow (skip)) != NULL; jsw_snext(skip)) {
          s = ctable->creator->get_string (row, field, NULL, utilityObj);
  	jsw_dump (s, skip, ctable->creator->fields[field]->indexNumber);
      }
  
      Tcl_DecrRefCount (utilityObj);
  
      return TCL_OK;
  }
  
  
  //
  // ctable_ListIndex - return a list of all of the index values 
  //
  // warning - can be hugely inefficient if you have a zillion elements
  // but useful for testing
  //
  CTABLE_INTERNAL int
  ctable_ListIndex (Tcl_Interp *interp, CTable *ctable, int fieldNum) {
      jsw_skip_t *skip = ctable->skipLists[fieldNum];
      ctable_BaseRow    *p;
      Tcl_Obj    *resultObj = Tcl_GetObjResult (interp);
  
      if (skip == NULL) {
          return TCL_OK;
      }
  
      for (jsw_sreset (skip); (p = jsw_srow (skip)) != NULL; jsw_snext(skip)) {
  
          if (ctable->creator->lappend_field (interp, resultObj, p, fieldNum) == TCL_ERROR) {
  	    Tcl_AppendResult (interp, " while walking index fields", (char *) NULL);
  	    return TCL_ERROR;
  	}
      }
  
      return TCL_OK;
  }
  
  CTABLE_INTERNAL INLINE void
  ctable_RemoveFromIndex (CTable *ctable, void *vRow, int field) {
      jsw_skip_t *skip = ctable->skipLists[field];
      ctable_BaseRow *row = (ctable_BaseRow*) vRow;
      int index = ctable->creator->fields[field]->indexNumber;
  
  #ifdef SEARCHDEBUG
  printf("ctable_RemoveFromIndex row 0x%lx, field %d (%s) skip == 0x%lx\n", (long)row, field, ctable->creator->fieldNames[field], (long unsigned int)skip);
  if(field == TRACKFIELD) {
    printf("BEFORE=  ");
    ctable_DumpIndex (ctable, field);
  }
  #endif
      // jsw_dump_head(skip);
  
      if (skip == NULL) {
  //printf("no skiplist\n");
          return;
      }
  
      // invariant: prev is never NULL if in list
      if(row->_ll_nodes[index].prev == NULL) {
  //printf("not in list\n");
  	return;
      }
      if (ctable_ListRemoveMightBeTheLastOne (row, index)) {
  //printf("i might be the last one, field %d\n", field);
          // it might be the last one, see if it really was
  //printf ("row->_ll_nodes[index].head %lx\n", (long unsigned int)row->_ll_nodes[index].head);
  	if (*row->_ll_nodes[index].head == NULL) {
  //printf("erasing last entry field %d\n", field);
              // put the pointer back so the compare routine will have
  	    // something to match
              *row->_ll_nodes[index].head = row;
  	    if (!jsw_serase (skip, row)) {
  		fprintf (stderr, "Attempted to remove non-existent field %s\n", ctable->creator->fields[field]->name);
  	    }
  	    *row->_ll_nodes[index].head = NULL; // don't think this is needed, but do it anyway
  	}
      }
  #ifdef SEARCHDEBUG
  if(field == TRACKFIELD) {
    printf("AFTER=  ");
    ctable_DumpIndex (ctable, field);
  }
  fflush(stdout);
  #endif
  
      return;
  }
  
  //
  // ctable_RemoveFromAllIndexes -- remove a row from all of the indexes it's
  // in -- this does a bidirectional linked list remove for each 
  //
  //
  //
  CTABLE_INTERNAL void
  ctable_RemoveFromAllIndexes (CTable *ctable, ctable_BaseRow *row) {
      int         field;
      
      // everybody's in index 0, take this guy out
      ctable_ListRemove (row, 0);
  
      // NB slightly gross, we shouldn't have to look at all of the fields
      // to even see which ones could be indexed but the programmer is
      // in a hurry
      for (field = 0; field < ctable->creator->nFields; field++) {
  	if (ctable->skipLists[field] != NULL) {
  	    ctable_RemoveFromIndex (ctable, row, field);
  	}
      }
  }
  
  //
  // ctable_InsertIntoIndex - for the given field of the given row of the given
  // ctable, insert this row into that table's field's index if there is an
  // index on that field.
  //
  CTABLE_INTERNAL INLINE int
  ctable_InsertIntoIndex (Tcl_Interp *interp, CTable *ctable, ctable_BaseRow *row, int field) {
      jsw_skip_t *skip = ctable->skipLists[field];
      ctable_FieldInfo *f;
      Tcl_Obj *utilityObj;
      ctable_CreatorTable *creator = ctable->creator;
      int index = creator->fields[field]->indexNumber;
  
      if (skip == NULL) {
      return TCL_OK;
      }
  
      // invariant: prev is always NULL if not in list
      if(row->_ll_nodes[index].prev != NULL) {
  	Tcl_Panic ("Double insert row for field %s", ctable->creator->fields[field]->name);
      }
  
  #ifdef SANITY_CHECKS
      creator->sanity_check_pointer(ctable, (void *)row, CTABLE_INDEX_NORMAL, "ctable_InsertIntoIndex : row");
      creator->sanity_check_pointer(ctable, (void *)skip, CTABLE_INDEX_NORMAL, "ctable_InsertIntoIndex : skip");
  #endif
  
      f = creator->fields[field];
  
  # ifdef SEARCHDEBUG
  // dump info about row being inserted
  utilityObj = Tcl_NewObj();
  printf("ctable_InsertIntoIndex row 0x%lx, field %d, field name %s, index %d, value '%s'\n", (long)row, field, f->name, f->indexNumber, ctable->creator->get_string (row, field, NULL, utilityObj));
  fflush(stdout);
  Tcl_DecrRefCount (utilityObj);
  if(field == TRACKFIELD) {
    printf("BEFORE=  ");
    ctable_DumpIndex (ctable, field);
  }
  #endif
  
      if (!jsw_sinsert_linked (skip, row, f->indexNumber, f->unique)) {
  
  	utilityObj = Tcl_NewObj();
  	Tcl_AppendResult (interp, "unique check failed for field \"", f->name, "\", value \"", ctable->creator->get_string (row, field, NULL, utilityObj), "\"", (char *) NULL);
  	Tcl_DecrRefCount (utilityObj);
          return TCL_ERROR;
      }
  
  # ifdef SEARCHDEBUG
      // ctable_verifyField(ctable, field, 0);
  if(field == TRACKFIELD) {
    printf("AFTER=  ");
    ctable_DumpIndex (ctable, field);
  }
  fflush(stdout);
  #endif
  
      return TCL_OK;
  }
  
  //
  // ctable_CreateIndex - create an index on a specified field of a specified
  // ctable.
  //
  CTABLE_INTERNAL int
  ctable_CreateIndex (Tcl_Interp *interp, CTable *ctable, int field, int depth) {
      ctable_BaseRow *row;
  
      jsw_skip_t      *skip;
  
      // make sure the field has an index set up for it
      // in the linked list nodes of the row.
      if (ctable->creator->fields[field]->indexNumber < 0) {
  	Tcl_AppendResult (interp, "can't create an index on field '", ctable->creator->fields[field]->name, "' that hasn't been defined as having an index", (char *)NULL);
  	return TCL_ERROR;
      }
  
      // make sure we're allowed to do this
  #ifdef WITH_SHARED_TABLES
      if(ctable->share_type == CTABLE_SHARED_READER) {
  	Tcl_AppendResult (interp, "can't create an index on a read-only table", (char *)NULL);
  	return TCL_ERROR;
      }
  #endif
  
      // if there's already a skip list, just say "fine"
      if ((skip = ctable->skipLists[field]) != NULL) {
          return TCL_OK;
      }
  
  #ifdef WITH_SHARED_TABLES
      if(ctable->share_type == CTABLE_SHARED_MASTER)
          skip = jsw_snew (depth, ctable->creator->fields[field]->compareFunction, ctable->share);
      else
  #endif
          skip = jsw_snew (depth, ctable->creator->fields[field]->compareFunction, NULL);
  
      // we should plug the list in last, so that concurrent users don't
      // walk an incomplete skiplist, but ctable_InsertIntoIndex needs this
      // TODO: make ctable_InsertIntoIndex a wrapper around a new "ctable_InsertIntoSkiplist"?
      ctable->skipLists[field] = skip;
  
      // Walk the whole table to create the index
      CTABLE_LIST_FOREACH (ctable->ll_head, row, 0) {
  	// we want to be able to call out to an error handler rather
  	// than fail and unwind the stack.
  	// (not here so much as in read_tabsep because here we just unwind
  	// and undo the new index if we get an error)
  	if (ctable_InsertIntoIndex (interp, ctable, row, field) == TCL_ERROR) {
  	    // you can't leave them with a partial index or there will
  	    // be heck to pay later when queries don't find all the
  	    // rows, etc
  	    jsw_sdelete_skiplist (skip, 0);
  	    ctable->skipLists[field] = NULL;
  	    Tcl_AppendResult (interp, " while creating index", (char *) NULL);
  	    return TCL_ERROR;
  	}
      }
  
      return TCL_OK;
  }
  
  CTABLE_INTERNAL int
  ctable_LappendIndexLowAndHi (Tcl_Interp *interp, CTable *ctable, int field) {
      jsw_skip_t            *skip = ctable->skipLists[field];
      ctable_BaseRow        *row;
      Tcl_Obj               *resultObj = Tcl_GetObjResult (interp);
  
      if (skip == NULL) {
  	Tcl_AppendResult (interp, "that field isn't indexed", (char *)NULL);
  	return TCL_ERROR;
      }
  
      jsw_sreset (skip);
      row = jsw_srow (skip);
  
      if (row == NULL) {
          return TCL_OK;
      }
  
      if (ctable->creator->lappend_field (interp, resultObj, row, ctable->creator->fieldList[field]) == TCL_ERROR) {
          return TCL_ERROR;
      }
  
      jsw_findlast (skip);
      row = jsw_srow (skip);
  
      if (ctable->creator->lappend_field (interp, resultObj, row, ctable->creator->fieldList[field]) == TCL_ERROR) {
          return TCL_ERROR;
      }
  
      return TCL_OK;
  }
  
  CTABLE_INTERNAL int
  ctable_DestroyCursor(Tcl_Interp *interp, struct cursor *cursor)
  {
      if(!cursor->ownerTable) return TCL_OK; // being deleted
      CTable *ctable = cursor->ownerTable;
      cursor->ownerTable = NULL;
  
      struct cursor *prev = NULL;
      struct cursor *next = ctable->cursors;
      while(next && next != cursor) {
  	prev = next;
  	next = next->nextCursor;
      }
      if(!next) return TCL_OK; // already deleted
  
      // Remove from ctable
      if(prev)
  	prev->nextCursor = next->nextCursor;
      else
  	ctable->cursors = next->nextCursor;
  
      if(cursor->tranTable) {
  	ckfree(cursor->tranTable);
  	cursor->tranTable = NULL;
      }
  
      if(cursor->cursorName) {
  	ckfree(cursor->cursorName);
  	cursor->cursorName = NULL;
      }
  
      if(interp && cursor->commandInfo) {
  	Tcl_Command tmp = cursor->commandInfo;
  	cursor->commandInfo = NULL;
  	Tcl_DeleteCommandFromToken(interp, tmp);
      }
  
  #ifdef WITH_SHARED_TABLES
      // destroying a read-locked cursor? Tag the whole ctable as locked
      if(cursor->lockCycle != LOST_HORIZON)
  	ctable->cursorLock = 1;
  
      if(!ctable->cursors) {
  	// if the table was tagged as locked, and this is the last cursor, unlock it
  	if(ctable->cursorLock) {
  	    read_unlock(ctable->share);
  	    ctable->cursorLock = 0;
  	}
          end_write(ctable);
      }
  #endif
  
      ckfree(cursor);
  
      return TCL_OK;
  }
  
  CTABLE_INTERNAL struct cursor *
  ctable_CreateEmptyCursor(Tcl_Interp *interp, CTable *ctable, char *cursorName)
  {
  	// ALLOCATE and build new cursor
          struct cursor *cursor = (struct cursor *)ckalloc(sizeof (struct cursor));
  
  	// CREATE empty transaction table (1 element, empty)
  	cursor->tranTable = (ctable_BaseRow **)ckalloc (sizeof (ctable_BaseRow *));
  	cursor->tranTable[0] = (ctable_BaseRow *)NULL;
  	cursor->tranIndex = 0;
  
  	// INITIALIZE default values
          cursor->cursorName = cursorName;
          cursor->offset = 0;
          cursor->offsetLimit = 0;
  	cursor->commandInfo = NULL;
  #ifdef WITH_SHARED_TABLES
  	cursor->lockCycle = LOST_HORIZON;
  #endif
  
  	// INSERT cursor into cursor list
          cursor->nextCursor = ctable->cursors;
          ctable->cursors = cursor;
  
  	// SAVE ctable link in cursor
          cursor->ownerTable = ctable;
  
  	// MARK cursor as valid
  	cursor->cursorState = CTABLE_CURSOR_OK;
  
  	return cursor;
  }
  
  CTABLE_INTERNAL struct cursor *
  ctable_CreateCursor(Tcl_Interp *interp, CTable *ctable, CTableSearch *search)
  {
  	// Defense - if it's already been created from this search, or the search doesn't need a cursor, drop it
  	if (!search->tranTable || !search->cursorName) return NULL;
  
  	// ALLOCATE and build new cursor
          struct cursor *cursor = (struct cursor *)ckalloc(sizeof (struct cursor));
  
  	// MOVE transaction table naem cursor name to cursor
          cursor->tranTable = search->tranTable;
          search->tranTable = NULL;
          cursor->cursorName = search->cursorName;
  	search->cursorName = NULL;
  
  	// COPY search cursor info to cursor
          cursor->offset = cursor->tranIndex = search->offset;
          cursor->offsetLimit = search->offsetLimit;
  
  	// INSERT cursor into cursor list
          cursor->nextCursor = ctable->cursors;
          ctable->cursors = cursor;
  
  	// SAVE ctable link in cursor
          cursor->ownerTable = ctable;
  
  	// MARK cursor as valid
  	cursor->cursorState = CTABLE_CURSOR_OK;
  
  	// INIT remaining feilds
  	cursor->commandInfo = NULL;
  #ifdef WITH_SHARED_TABLES
  	cursor->lockCycle = LOST_HORIZON;
  #endif
  
  	return cursor;
  }
  
  CTABLE_INTERNAL void
  ctable_DeleteCursorCommand(ClientData clientData)
  {
  	struct cursor *cursor = (struct cursor *)clientData;
  
  	// Make sure we don't lead ourselves back here
  	cursor->commandInfo = NULL;
  
  	// Remove the cursor from the ctable and destroy it
  	ctable_DestroyCursor(NULL, cursor);
  }
  
  CTABLE_INTERNAL int
  ctable_CreateCursorCommand(Tcl_Interp *interp, struct cursor *cursor)
  {
  	if(cursor->commandInfo) return TCL_OK; // already exists
  
  	Tcl_ObjCmdProc *commandProc = cursor->ownerTable->creator->cursor_command;
  
  	cursor->commandInfo = Tcl_CreateObjCommand(interp, cursor->cursorName, commandProc, (ClientData)cursor, ctable_DeleteCursorCommand);
  
  	return cursor->commandInfo ? TCL_OK : TCL_ERROR;
  }
  
  CTABLE_INTERNAL Tcl_Obj *
  ctable_CursorToName(struct cursor *cursor)
  {
  	return Tcl_NewStringObj(cursor->cursorName, -1);
  }
  
  
  // vim: set ts=8 sw=4 sts=4 noet :
  