path: root/storage/connect/tabpivot.cpp

/************ TabPivot C++ Program Source Code File (.CPP) *************/
/* PROGRAM NAME: TABPIVOT                                              */
/* -------------                                                       */
/*  Version 1.3                                                        */
/*                                                                     */
/* COPYRIGHT:                                                          */
/* ----------                                                          */
/*  (C) Copyright to the author Olivier BERTRAND          2005-2012    */
/*                                                                     */
/* WHAT THIS PROGRAM DOES:                                             */
/* -----------------------                                             */
/*  This program are the PIVOT classes DB execution routines.          */
/***********************************************************************/

/***********************************************************************/
/*  Include relevant sections of the operating system header file.     */
/***********************************************************************/
#include "my_global.h"
#if defined(WIN32)
#if defined(__BORLANDC__)
#define __MFC_COMPAT__                   // To define min/max as macro
#endif
//#include <windows.h>
#elif defined(UNIX)
#include <errno.h>
#include <unistd.h>
#include "osutil.h"
#else
#include <io.h>
#endif

/***********************************************************************/
/*  Include application header files:                                  */
/*  global.h    is header containing all global declarations.          */
/*  plgdbsem.h  is header containing the DB application declarations.  */
/***********************************************************************/
#define FRM_VER 6
#include "table.h"       // MySQL table definitions
#include "sql_const.h"
#include "field.h"
#include "global.h"
#include "plgdbsem.h"
#include "xtable.h"
#include "xindex.h"
#include "colblk.h"
#include "tabmysql.h"
#include "csort.h"
#include "tabpivot.h"
#include "valblk.h"
#include "ha_connect.h"
#include "mycat.h"       // For GetHandler

extern "C" int trace;

/* --------------- Implementation of the PIVOT classes --------------- */

/***********************************************************************/
/*  DefineAM: define specific AM block values from PIVOT table.        */
/***********************************************************************/
bool PIVOTDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
  {
  char *p1, *p2;
  PHC    hc = ((MYCAT*)Cat)->GetHandler();

  Host = Cat->GetStringCatInfo(g, Name, "Host", "localhost");
  User = Cat->GetStringCatInfo(g, Name, "User", "root");
  Pwd = Cat->GetStringCatInfo(g, Name, "Password", NULL);
  DB = Cat->GetStringCatInfo(g, Name, "Database", (PSZ)hc->GetDBName(NULL));
  Tabsrc = Cat->GetStringCatInfo(g, Name, "SrcDef", NULL);
  Tabname = Cat->GetStringCatInfo(g, Name, "Name", NULL);
  Picol = Cat->GetStringCatInfo(g, Name, "PivotCol", NULL);
  Fncol = Cat->GetStringCatInfo(g, Name, "FncCol", NULL);
  
  // If fncol is like avg(colname), separate Fncol and Function
  if (Fncol && (p1 = strchr(Fncol, '(')) && (p2 = strchr(p1, ')')) &&
      (*Fncol != '"') &&  (!*(p2+1))) {
    *p1++ = '\0'; *p2 = '\0';
    Function = Fncol;
    Fncol = p1;
  } else
    Function = Cat->GetStringCatInfo(g, Name, "Function", "SUM");

  GBdone = Cat->GetIntCatInfo(Name, "Groupby", 0) ? TRUE : FALSE;
  Port = Cat->GetIntCatInfo(Name, "Port", 3306);
  Desc = (*Tabname) ? Tabname : Tabsrc;
  return FALSE;
  } // end of DefineAM

/***********************************************************************/
/*  GetTable: makes a new TDB of the proper type.                      */
/***********************************************************************/
PTDB PIVOTDEF::GetTable(PGLOBAL g, MODE m)
  {
  return new(g) TDBPIVOT(this);
  } // end of GetTable

/* ------------------------------------------------------------------- */

/***********************************************************************/
/*  Implementation of the TDBPIVOT class.                              */
/***********************************************************************/
TDBPIVOT::TDBPIVOT(PPIVOTDEF tdp) : TDBASE(tdp), CSORT(FALSE)
  {
  Tqrp = NULL;
  Host = tdp->Host;
  Database = tdp->DB;
  User = tdp->User;
  Pwd = tdp->Pwd;
  Port = tdp->Port;
  Qryp = NULL;
  Tabname = tdp->Tabname;    // Name of source table
  Tabsrc = tdp->Tabsrc;      // SQL description of source table
  Picol = tdp->Picol;        // Pivot column name
  Fncol = tdp->Fncol;        // Function column name
  Function = tdp->Function;  // Aggregate function name
  Xcolp = NULL;              // To the FNCCOL column
//Xresp = NULL;              // To the pivot result column
//Rblkp = NULL;              // The value block of the pivot column
  Fcolp = NULL;              // To the function column
  GBdone = tdp->GBdone;
  Mult = -1;                // Estimated table size
  N = 0;                    // The current table index
  M = 0;                    // The occurence rank
  FileStatus = 0;           // Logical End-of-File
  RowFlag = 0;              // 0: Ok, 1: Same, 2: Skip
  } // end of TDBPIVOT constructor

#if 0
TDBPIVOT::TDBPIVOT(PTDBPIVOT tdbp) : TDBASE(tdbp), CSORT(FALSE)
  {
  Tdbp = tdbp->Tdbp;
  Sqlp = tdbp->Sqlp;
  Qryp = tdbp->Qryp;
  Tabname = tdbp->Tabname;
  Tabsrc = tdbp->Tabsrc;
  Picol = tdbp->Picol;
  Fncol = tdbp->Fncol;
  Function = tdbp->Function;
  Xcolp = tdbp->Xcolp;
  Xresp = tdbp->Xresp;
  Rblkp = tdbp->Rblkp;
  Fcolp = tdbp->Fcolp;
  Mult = tdbp->Mult;
  N = tdbp->N;
  M = tdbp->M;
  FileStatus = tdbp->FileStatus;
  RowFlag = tdbp->RowFlag;
  } // end of TDBPIVOT copy constructor

// Is this really useful ???
PTDB TDBPIVOT::CopyOne(PTABS t)
  {
  PTDB tp = new(t->G) TDBPIVOT(this);

  tp->SetColumns(Columns);
  return tp;
  } // end of CopyOne
#endif // 0

/***********************************************************************/
/*  Prepare the source table Query.                                    */
/***********************************************************************/
PQRYRES TDBPIVOT::GetSourceTable(PGLOBAL g)
  {
  if (Qryp)
    return Qryp;             // Already done

  if (!Tabsrc && Tabname) {
    char   *def, *colist;
    size_t  len = 0;
    PCOL    colp;
    PDBUSER dup = (PDBUSER)g->Activityp->Aptr;

    // Evaluate the length of the column list
    for (colp = Columns; colp; colp = colp->GetNext())
      len += (strlen(colp->GetName()) + 2);

    *(colist = (char*)PlugSubAlloc(g, NULL, len)) = 0;

    // Locate the suballocated string (size is not known yet)
    def = (char*)PlugSubAlloc(g, NULL, 0);
    strcpy(def, "SELECT ");
      
    if (!Fncol) {
      for (colp = Columns; colp; colp = colp->GetNext())
        if (!Picol || stricmp(Picol, colp->GetName()))
          Fncol = colp->GetName();

      if (!Fncol) {
        strcpy(g->Message, MSG(NO_DEF_FNCCOL));
        return NULL;
        } // endif Fncol

    }  else if (!(ColDB(g, Fncol, 0))) {
      // Function column not found in table                                       
      sprintf(g->Message, MSG(COL_ISNOT_TABLE), Fncol, Tabname);
      return NULL;
    } // endif Fcolp

    if (!Picol) {
      // Find default Picol as the last one not equal  to Fncol
      for (colp = Columns; colp; colp = colp->GetNext())
        if (!Fncol || stricmp(Fncol, colp->GetName()))
          Picol = colp->GetName();

      if (!Picol) {
        strcpy(g->Message, MSG(NO_DEF_PIVOTCOL));
        return NULL;
        } // endif Picol

    } else if (!(ColDB(g, Picol, 0))) {
      // Pivot column not found in table                                       
      sprintf(g->Message, MSG(COL_ISNOT_TABLE), Picol, Tabname);
      return NULL;
    } // endif Xcolp

    // Make the other column list
    for (colp = Columns; colp; colp = colp->GetNext())
      if (stricmp(Picol, colp->GetName()) &&
          stricmp(Fncol, colp->GetName()))
        strcat(strcat(colist, colp->GetName()), ", ");

    // Add the Pivot column at the end of the list
    strcat(strcat(def, strcat(colist, Picol)), ", ");

    // Continue making the definition
    if (!GBdone) {
      // Make it suitable for Pivot by doing the group by
      strcat(strcat(def, Function), "(");
      strcat(strcat(strcat(def, Fncol), ") "), Fncol);
      strcat(strcat(def, " FROM "), Tabname);
      strcat(strcat(def, " GROUP BY "), colist);
    } else   // Gbdone
      strcat(strcat(strcat(def, Fncol), " FROM "), Tabname);

    // Now we know how much was suballocated
    Tabsrc = (char*)PlugSubAlloc(g, NULL, strlen(def));
  } else {
    strcpy(g->Message, MSG(SRC_TABLE_UNDEF));
    return NULL;
  } // endif Tabsrc

  int w;

  // Open a MySQL connection for this table
  if (Myc.Open(g, Host, Database, User, Pwd, Port))
    return NULL;

  // Send the source command to MySQL
  if (Myc.ExecSQL(g, Tabsrc, &w) == RC_FX) {
    Myc.Close();
    return NULL;
    } // endif Exec

  // We must have a storage query to get pivot column values
  Qryp = Myc.GetResult(g);
  Myc.Close();
  Tqrp = new(g) TDBQRS(Qryp);
  Tqrp->OpenDB(g);

  if (MakePivotColumns(g) < 0)
    return NULL;

  return Qryp;
  } // end of GetSourceTable

/***********************************************************************/
/*  Allocate PIVOT columns description block.                          */
/***********************************************************************/
int TDBPIVOT::MakePivotColumns(PGLOBAL g)
  {
  if (Mult < 0) {
    int     ndif, n = 0, nblin = Qryp->Nblin;
    PVAL    valp;
    PCOL    cp;
    PSRCCOL colp;
    PFNCCOL cfnp;

    // Allocate all the source columns
    Tqrp->ColDB(g, NULL, 0);
    Columns = NULL;        // Discard dummy columns blocks

    for (cp = Tqrp->GetColumns(); cp; cp = cp->GetNext()) {
      if (cp->InitValue(g))
        return -1;

      if (!stricmp(cp->GetName(), Picol)) {
        Xcolp = (PQRSCOL)cp;
        Xresp = Xcolp->GetCrp();
        Rblkp = Xresp->Kdata;
      } else if (!stricmp(cp->GetName(), Fncol)) {
        Fcolp = (PQRSCOL)cp;
      } else
        if ((colp = new(g) SRCCOL(cp, this, ++n))->Init(g, this))
          return -1;

      } // endfor cp

    if (!Xcolp) {
      sprintf(g->Message, MSG(COL_ISNOT_TABLE), 
              Picol, Tabname ? Tabname : "TabSrc");
      return -1;
    } else if (!Fcolp) {
      sprintf(g->Message, MSG(COL_ISNOT_TABLE), 
              Fncol, Tabname ? Tabname : "TabSrc");
      return -1;
    } // endif Fcolp

    // Before calling sort, initialize all
    Index.Size = nblin * sizeof(int);
    Index.Sub = TRUE;                   // Should be small enough

    if (!PlgDBalloc(g, NULL, Index))
      return -1;

    Offset.Size = (nblin + 1) * sizeof(int);
    Offset.Sub = TRUE;                 // Should be small enough

    if (!PlgDBalloc(g, NULL, Offset))
      return -2;

    ndif = Qsort(g, nblin);

    if (ndif < 0) {           // error
      return -3;
    } else
      Ncol = ndif;

    // Now make the functional columns
    for (int i = 0; i < Ncol; i++) {
      // Allocate the Value used to retieve column names
      if (!(valp = AllocateValue(g, Xcolp->GetResultType(),
          Xcolp->GetLengthEx(),  Xcolp->GetPrecision(),
          Xcolp->GetDomain(), Xcolp->GetTo_Tdb()->GetCat())))
        return -4;

      // Get the value that will be the generated column name
      valp->SetValue_pvblk(Rblkp, Pex[Pof[i]]);

      // Copy the functional column with new Name and new Value
      cfnp = (PFNCCOL)new(g) FNCCOL(Fcolp, this);

      // Initialize the generated column
      if (cfnp->InitColumn(g, valp))
        return -5;

      } // endfor i

    // Fields must be updated for ha_connect
//  if (UpdateTableFields(g, n + Ncol))
//    return -6;

    // This should be refined later
    Mult = nblin;
    } // endif Mult

  return Mult;
  } // end of MakePivotColumns

#if 0
/***********************************************************************/
/*  Update fields in the MySQL table structure                         */
/*  Note: this does not work. Indeed the new rows are correctly made   */
/*  but the final result still specify the unmodified table and the    */
/*  returned table only contains the original column values.           */
/*  In addition, a new query on the table, when it is into the cache,  */
/*  specifies all the new columns and fails because they do not belong */
/*  to the original table.                                             */
/***********************************************************************/
bool TDBPIVOT::UpdateTableFields(PGLOBAL g, int n)
  {
  uchar  *trec, *srec, *tptr, *sptr;
  int     i = 0, k = 0;
  uint    len;
  uint32  nmp, lwm;
  size_t  buffsize;
  PCOL    colp;
  PHC      hc = ((MYCAT*)((PIVOTDEF*)To_Def)->Cat)->GetHandler();
  TABLE   *table = hc->GetTable();
  st_mem_root *tmr = &table->mem_root;
  st_mem_root *smr = &table->s->mem_root;
  Field* *field;
  Field  *fp, *tfncp, *sfncp;
  Field* *ntf;
  Field* *nsf;
//my_bitmap_map   *org_bitmap;
  const MY_BITMAP *map;

  // When sorting read_set selects all columns, so we use def_read_set
  map= (const MY_BITMAP *)&table->def_read_set;

  // Find the function field 
  for (field= table->field; *field; field++) {
    fp= *field;

    if (bitmap_is_set(map, fp->field_index))
      if (!stricmp(fp->field_name, Fncol)) {
        tfncp = fp;
        break;
        } // endif Name

    } // endfor field

  for (field= table->s->field; *field; field++) {
    fp= *field;

    if (bitmap_is_set(map, fp->field_index))
      if (!stricmp(fp->field_name, Fncol)) {
        sfncp = fp;
        break;
        } // endif Name

    } // endfor field

  // Calculate the new buffer size
  len = tfncp->max_data_length();
   buffsize = table->s->rec_buff_length + len * Ncol;

  // Allocate the new record space
  if (!(tptr = trec = (uchar*)alloc_root(tmr, 2 * buffsize)))
    return TRUE;

  if (!(sptr = srec = (uchar*)alloc_root(smr, 2 * buffsize)))
    return TRUE;


  // Allocate the array of all new table field pointers
  if (!(ntf = (Field**)alloc_root(tmr, (uint)((n+1) * sizeof(Field*)))))
    return TRUE;

  // Allocate the array of all new table share field pointers
  if (!(nsf = (Field**)alloc_root(smr, (uint)((n+1) * sizeof(Field*)))))
    return TRUE;

  // First fields are the the ones of the source columns
  for (colp = Columns; colp; colp = colp->GetNext())
    if (colp->GetAmType() == TYPE_AM_SRC) {
      for (field= table->field; *field; field++) {
        fp= *field;

        if (bitmap_is_set(map, fp->field_index))
          if (!stricmp(colp->GetName(),  fp->field_name)) {
            ntf[i] = fp;
            fp->field_index =  i++;
            fp->ptr = tptr;
            tptr += fp->max_data_length();
            break;
            } // endif Name

        } // endfor field

      for (field= table->s->field; *field; field++) {
        fp= *field;

        if (bitmap_is_set(map, fp->field_index))
          if (!stricmp(colp->GetName(),  fp->field_name)) {
            nsf[k] = fp;
            fp->field_index =  k++;
            fp->ptr = srec;
            srec += fp->max_data_length();
            break;
            } // endif Name

        } // endfor field

      } // endif  AmType

  // Now add the pivot generated columns
  for (colp = Columns; colp; colp = colp->GetNext())
    if (colp->GetAmType() == TYPE_AM_FNC) {
      if ((fp = (Field*)memdup_root(tmr, (char*)tfncp, tfncp->size_of()))) {
        ntf[i] = fp;
        fp->ptr = tptr;
        fp->field_name = colp->GetName();
        fp->field_index = i++;
        fp->vcol_info = NULL;
        fp->stored_in_db = TRUE;
        tptr += len;
      } else
        return TRUE;

      if ((fp = (Field*)memdup_root(smr, (char*)sfncp, sfncp->size_of()))) {
        nsf[i] = fp;
        fp->ptr = sptr;
        fp->field_name = colp->GetName();
        fp->field_index = k++;
        fp->vcol_info = NULL;
        fp->stored_in_db = TRUE;
        sptr += len;
      } else
        return TRUE;

      } // endif AM_FNC

  // Mark end of the list
  ntf[i] = NULL;
  nsf[k] = NULL;

  // Update the table fields
  nmp = (uint32)((1<<i) - 1);
  lwm = (uint32)((-1)<<i);
  table->field = ntf;
  table->used_fields = i;
  table->record[0] = trec;
  table->record[1] = trec + buffsize;
  *table->def_read_set.bitmap = nmp;
  *table->def_read_set.last_word_ptr = nmp;
  table->def_read_set.last_word_mask = lwm;
  table->def_read_set.n_bits = i;
  *table->read_set->bitmap = nmp;
  *table->read_set->last_word_ptr = nmp;
  table->read_set->last_word_mask = lwm;
  table->read_set->n_bits = i;
  table->write_set->n_bits = i;
  *table->vcol_set->bitmap = 0;
  table->vcol_set->n_bits = i;

  // and the share fields
  table->s->field = nsf;
  table->s->reclength = sptr - srec;
  table->s->stored_rec_length = sptr - srec;
  table->s->fields = k;
  table->s->stored_fields = k;
  table->s->rec_buff_length = buffsize;
//table->s->varchar_fields = ???;
//table->s->db_record_offset = ???;
//table->s->null_field_first = ???;
  return FALSE;
  } // end of UpdateTableFields
#endif // 0

/***********************************************************************/
/*  Allocate source column description block.                          */
/***********************************************************************/
PCOL TDBPIVOT::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
  {
  PCOL colp = NULL;

//if (stricmp(cdp->GetName(), Picol) && stricmp(cdp->GetName(), Fncol)) {
    colp = new(g) COLBLK(cdp, this, n);

//  if (((PSRCCOL)colp)->Init(g, this))
//    return NULL;

//} else {
//  sprintf(g->Message, MSG(NO_MORE_COL), cdp->GetName());
//  return NULL;
//} // endif Name

  if (cprec) {
    colp->SetNext(cprec->GetNext());
    cprec->SetNext(colp);
  } else {
    colp->SetNext(Columns);
    Columns = colp;
  } // endif cprec

  return colp;
  } // end of MakeCol

/***********************************************************************/
/*  PIVOT GetMaxSize: returns the maximum number of rows in the table. */
/***********************************************************************/
int TDBPIVOT::GetMaxSize(PGLOBAL g)
  {     
#if  0
  if (MaxSize < 0)
    MaxSize = MakePivotColumns(g);

  return MaxSize;
#endif // 0
  return 0;
  } // end of GetMaxSize

/***********************************************************************/
/*  In this sample, ROWID will be the (virtual) row number,            */
/*  while ROWNUM will be the occurence rank in the multiple column.    */
/***********************************************************************/
int TDBPIVOT::RowNumber(PGLOBAL g, bool b)
  {
  return (b) ? M : N;
  } // end of RowNumber
 
/***********************************************************************/
/*  PIVOT Access Method opening routine.                               */
/***********************************************************************/
bool TDBPIVOT::OpenDB(PGLOBAL g)
  {
//PDBUSER dup = (PDBUSER)g->Activityp->Aptr;

  if (Use == USE_OPEN) {
    /*******************************************************************/
    /*  Table already open, just replace it at its beginning.          */
    /*******************************************************************/
    N = M = 0;
    RowFlag = 0;
    FileStatus = 0;
    return FALSE;
    } // endif use

  /*********************************************************************/
  /*  Do it here if not done yet (should not be the case).             */
  /*********************************************************************/
//if (MakePivotColumns(g) < 0)
  if (!(Qryp = GetSourceTable(g)))
    return TRUE;

  if (Mode != MODE_READ) {
    /*******************************************************************/
    /* Currently PIVOT tables cannot be modified.                      */
    /*******************************************************************/
    sprintf(g->Message, MSG(TABLE_READ_ONLY), "PIVOT");
    return TRUE;
    } // endif Mode

  if (To_Key_Col || To_Kindex) {
    /*******************************************************************/
    /* Direct access of PIVOT tables is not implemented yet.           */
    /*******************************************************************/
    strcpy(g->Message, MSG(NO_PIV_DIR_ACC));
    return TRUE;
    } // endif To_Key_Col

  return FALSE;
  } // end of OpenDB

/***********************************************************************/
/*  Data Base read routine for PIVOT access method.                    */
/***********************************************************************/
int TDBPIVOT::ReadDB(PGLOBAL g)
  {
  int  rc = RC_OK;
  bool newrow = FALSE;
  PCOL colp;
  PVAL vp1, vp2;

  if (FileStatus == 2)
    return RC_EF;

  if (FileStatus)
    for (colp = Columns; colp; colp = colp->GetNext())
      if (colp->GetAmType() == TYPE_AM_SRC)
        ((PSRCCOL)colp)->SetColumn();

  // New row, reset all function column values
  for (colp = Columns; colp; colp = colp->GetNext())
    if (colp->GetAmType() == TYPE_AM_FNC)
      colp->GetValue()->Reset();

  /*********************************************************************/
  /*  Now start the multi reading process.                             */
  /*********************************************************************/
  do {
    if (RowFlag != 1) {
      if ((rc = Tqrp->ReadDB(g)) != RC_OK) {
        if (FileStatus && rc == RC_EF) {
          // A prepared row remains to be sent
          FileStatus = 2;
          rc = RC_OK;
          } // endif FileStatus

        break;
        } // endif rc

      for (colp = Tqrp->GetColumns(); colp; colp = colp->GetNext())
        colp->ReadColumn(g);

      for (colp = Columns; colp; colp = colp->GetNext())
        if (colp->GetAmType() == TYPE_AM_SRC)
          if (FileStatus) {
            if (((PSRCCOL)colp)->CompareColumn())
              newrow = (RowFlag) ? TRUE : FALSE;

          } else
            ((PSRCCOL)colp)->SetColumn();

      FileStatus = 1;
      } // endif RowFlag

    if (newrow) {
      RowFlag = 1;
      break;
    } else
      RowFlag = 2;

    vp1 = Xcolp->GetValue();

    // Look for the column having this header
    for (colp = Columns; colp; colp = colp->GetNext())
      if (colp->GetAmType() == TYPE_AM_FNC) {
        vp2 = ((PFNCCOL)colp)->Hval;

        if (!vp1->CompareValue(vp2))
          break;

        } // endif AmType

    if (!colp) {
      strcpy(g->Message, MSG(NO_MATCH_COL));
      return RC_FX;
      } // endif colp

    // Set the value of the matching column from the fonction value
    colp->GetValue()->SetValue_pval(Fcolp->GetValue());
    } while (RowFlag == 2);

  N++;
  return rc;
  } // end of ReadDB

/***********************************************************************/
/*  WriteDB: Data Base write routine for PIVOT access methods.         */
/***********************************************************************/
int TDBPIVOT::WriteDB(PGLOBAL g)
  {
  sprintf(g->Message, MSG(TABLE_READ_ONLY), "PIVOT");
  return RC_FX;
  } // end of WriteDB

/***********************************************************************/
/*  Data Base delete line routine for PIVOT access methods.            */
/***********************************************************************/
int TDBPIVOT::DeleteDB(PGLOBAL g, int irc)
  {
  sprintf(g->Message, MSG(NO_TABLE_DEL), "PIVOT");
  return RC_FX;
  } // end of DeleteDB

/***********************************************************************/
/*  Data Base close routine for PIVOT access method.                   */
/***********************************************************************/
void TDBPIVOT::CloseDB(PGLOBAL g)
  {
//Tdbp->CloseDB(g);
  } // end of CloseDB

/***********************************************************************/
/*  TDBPIVOT: Compare routine for sorting pivot column values.         */
/***********************************************************************/
int TDBPIVOT::Qcompare(int *i1, int *i2)
  {
  // TODO: the actual comparison between pivot column result values.
  return Rblkp->CompVal(*i1, *i2);
  } // end of Qcompare

// ------------------------ FNCCOL functions ----------------------------

/***********************************************************************/
/*  FNCCOL public constructor.                                       */
/***********************************************************************/
FNCCOL::FNCCOL(PCOL col1, PTDBPIVOT tdbp)
      : COLBLK(col1, tdbp)
  {
  Value = NULL;     // We'll get a new one later
  Hval = NULL;     // The unconverted header value
  } // end of FNCCOL constructor

/***********************************************************************/
/*  FNCCOL initialization function.                                  */
/***********************************************************************/
bool FNCCOL::InitColumn(PGLOBAL g, PVAL valp)
{
  char *p, buf[128];
  int   len;

  // Must have its own value block
  if (InitValue(g))
    return TRUE;

  // Convert header value to a null terminated character string
  Hval = valp;
  p = Hval->GetCharString(buf);
  len = strlen(p) + 1;

  if (len > (signed)sizeof(buf)) {
    strcpy(g->Message, MSG(COLNAM_TOO_LONG));
    return TRUE;
    } // endif buf

  // Set the name of the functional pivot column
  Name = (PSZ)PlugSubAlloc(g, NULL, len);
  strcpy(Name, p);
  AddStatus(BUF_READ);      // All is done here
  return FALSE;
} // end of InitColumn

// ------------------------ SRCCOL functions ----------------------------

#if 0
/***********************************************************************/
/*  SRCCOL public constructor.                                         */
/***********************************************************************/
SRCCOL::SRCCOL(PCOLDEF cdp, PTDBPIVOT tdbp, int n)
      : COLBLK(cdp, tdbp, n)
  {
  // Set additional SRC access method information for column.
  Colp = NULL;
  Cnval = NULL;
  } // end of SRCCOL constructor
#endif // 0

/***********************************************************************/
/*  SRCCOL public constructor.                                         */
/***********************************************************************/
SRCCOL::SRCCOL(PCOL cp, PTDBPIVOT tdbp, int n)
      : COLBLK(cp, tdbp)
  {
  Index = n;

  // Set additional SRC access method information for column.
  Colp = (PQRSCOL)cp;

  // Don't share Value with the source column  so we can compare
  Cnval = Value = NULL;
  } // end of SRCCOL constructor

#if 0
/***********************************************************************/
/*  SRCCOL constructor used for copying columns.                       */
/*  tdbp is the pointer to the new table descriptor.                   */
/***********************************************************************/
SRCCOL::SRCCOL(SRCCOL *col1, PTDB tdbp) : COLBLK(col1, tdbp)
  {
  Colp = col1->Colp;
  } // end of SRCCOL copy constructor
#endif // 0

/***********************************************************************/
/*  Initialize the column as pointing to the source column.            */
/***********************************************************************/
bool SRCCOL::Init(PGLOBAL g, PTDBPIVOT tdbp)
  {
  // Currently we ignore the type of the create table column
  bool conv = FALSE;

#if 0
  if (!Colp) {
    // Column was defined in the Create Table statement
    if (!(Colp = tdbp->ColDB(g, Name, 0)))
      return TRUE;

    // We can have a conversion problem converting from numeric to
    // character because GetCharValue does not exist for numeric types.
    conv = (IsTypeChar(Buf_Type) && IsTypeNum(Colp->GetResultType()));
  } else
#endif // 0
    conv = FALSE;

  if (InitValue(g))
    return TRUE;

//if (conv)
//  Cnval = AllocateValue(g, Colp->GetValue());
//else
    Cnval = Value;

  AddStatus(BUF_READ);     // All is done here
  return FALSE;
  } // end of SRCCOL constructor

/***********************************************************************/
/*  SetColumn: have the column value set from the source column.       */
/***********************************************************************/
void SRCCOL::SetColumn(void)
  {
#if defined(_DEBUG)
  Cnval->SetValue_pval(Colp->GetValue(), TRUE);
#else
  Cnval->SetValue_pval(Colp->GetValue());
#endif   // _DEBUG

  if (Value != Cnval)
    // Convert value
    Value->SetValue_pval(Cnval);

  } // end of SetColumn

/***********************************************************************/
/*  SetColumn: Compare column value with source column value.          */
/***********************************************************************/
bool SRCCOL::CompareColumn(void)
  {
  // Compare the unconverted values
  return Cnval->CompareValue(Colp->GetValue());
  } // end of CompareColumn


/* ------------------------------------------------------------------- */

/***********************************************************************/
/*  Implementation of the TDBQRS class.                                */
/***********************************************************************/
TDBQRS::TDBQRS(PTDBQRS tdbp) : TDBASE(tdbp)
  {
  Qrp = tdbp->Qrp;
  CurPos = tdbp->CurPos;
  } // end of TDBQRS copy constructor

// Method
PTDB TDBQRS::CopyOne(PTABS t)
  {
  PTDB    tp;
  PQRSCOL cp1, cp2;
  PGLOBAL g = t->G;        // Is this really useful ???

  tp = new(g) TDBQRS(this);

  for (cp1 = (PQRSCOL)Columns; cp1; cp1 = (PQRSCOL)cp1->GetNext()) {
    cp2 = new(g) QRSCOL(cp1, tp);  // Make a copy
    NewPointer(t, cp1, cp2);
    } // endfor cp1

  return tp;
  } // end of CopyOne

#if 0    // The TDBASE functions return NULL when To_Def is NULL
/***********************************************************************/
/*  Return the pointer on the DB catalog this table belongs to.        */
/***********************************************************************/
PCATLG TDBQRS::GetCat(void)
  {
  // To_Def is null for QRYRES tables
  return NULL;
  }  // end of GetCat

/***********************************************************************/
/*  Return the datapath of the DB this table belongs to.               */
/***********************************************************************/
PSZ TDBQRS::GetPath(void)
  {
  // To_Def is null for QRYRES tables
  return NULL;
  }  // end of GetPath
#endif // 0

/***********************************************************************/
/*  Initialize QRS column description block construction.              */
/*        name is used to call columns by name.                        */
/*        num is used by LNA to construct columns by index number.     */
/*  Note: name=Null and num=0 for constructing all columns (select *)  */
/***********************************************************************/
PCOL TDBQRS::ColDB(PGLOBAL g, PSZ name, int num)
  {
  int     i;
  PCOLRES crp;
  PCOL    cp, colp = NULL, cprec = NULL;

  if (trace)
    htrc("QRS ColDB: colname=%s tabname=%s num=%d\n",
                    SVP(name), Name, num);

  for (crp = Qrp->Colresp, i = 1; crp; crp = crp->Next, i++)
    if ((!name && !num) ||
         (name && !stricmp(crp->Name, name)) || num == i) {
      // Check for existence of desired column
      // Also find where to insert the new block
      for (cp = Columns; cp; cp = cp->GetNext())
        if (cp->GetIndex() < i)
          cprec = cp;
        else if (cp->GetIndex() == i)
          break;

      if (trace) {
        if (cp)
          htrc("cp(%d).Name=%s cp=%p\n", i, cp->GetName(), cp);
        else
          htrc("cp(%d) cp=%p\n", i, cp);
        } // endif trace

      // Now take care of Column Description Block
      if (cp)
        colp = cp;
      else
        colp = new(g) QRSCOL(g, crp, this, cprec, i);

      if (name || num)
        break;
      else
        cprec = colp;

      } // endif Name

  return (colp);
  } // end of ColDB

/***********************************************************************/
/*  QRS GetMaxSize: returns maximum table size in number of lines.     */
/***********************************************************************/
int TDBQRS::GetMaxSize(PGLOBAL g)
  {
  MaxSize = Qrp->Maxsize;
  return MaxSize;
  } // end of GetMaxSize

/***********************************************************************/
/*  RowNumber: returns the current row ordinal number.                 */
/***********************************************************************/
int TDBQRS::RowNumber(PGLOBAL g, BOOL b)
  {
  return (CurPos + 1);
  } // end of RowNumber

/***********************************************************************/
/*  QRS Access Method opening routine.                                 */
/*  New method now that this routine is called recursively (last table */
/*  first in reverse order): index blocks are immediately linked to    */
/*  join block of next table if it exists or else are discarted.       */
/***********************************************************************/
bool TDBQRS::OpenDB(PGLOBAL g)
  {
  if (trace)
    htrc("QRS OpenDB: tdbp=%p tdb=R%d use=%d key=%p mode=%d\n",
                      this, Tdb_No, Use, To_Key_Col, Mode);

  if (Mode != MODE_READ) {
    sprintf(g->Message, MSG(BAD_QUERY_OPEN), Mode);
    return TRUE;
    } // endif Mode

  CurPos = -1;

  if (Use == USE_OPEN) {
    /*******************************************************************/
    /*  Table already open, just replace it at its beginning.          */
    /*******************************************************************/
    if (To_Kindex)
      /*****************************************************************/
      /*  Table is to be accessed through a sorted index table.        */
      /*****************************************************************/
      To_Kindex->Reset();

    return FALSE;
    } // endif use

  /*********************************************************************/
  /*  Open (retrieve data from) the query if not already open.         */
  /*********************************************************************/
  Use = USE_OPEN;       // Do it now in case we are recursively called

  return FALSE;
  } // end of OpenDB

/***********************************************************************/
/*  GetRecpos: returns current position of next sequential read.       */
/***********************************************************************/
int TDBQRS::GetRecpos(void)
  {
  return (CurPos);
  } // end of GetRecpos

/***********************************************************************/
/*  ReadDB: Data Base read routine for QRS access method.              */
/***********************************************************************/
int TDBQRS::ReadDB(PGLOBAL g)
  {
  int rc = RC_OK;

  if (trace)
    htrc("QRS ReadDB: R%d CurPos=%d key=%p link=%p Kindex=%p\n",
              GetTdb_No(), CurPos, To_Key_Col, To_Link, To_Kindex);

  if (To_Kindex) {
    /*******************************************************************/
    /*  Reading is by an index table.                                  */
    /*******************************************************************/
    int recpos = To_Kindex->Fetch(g);

    switch (recpos) {
      case -1:           // End of file reached
        rc = RC_EF;
        break;
      case -2:           // No match for join
        rc = RC_NF;
        break;
      case -3:           // Same record as last non null one
        rc = RC_OK;
        break;
      default:
        /***************************************************************/
        /*  Set the file position according to record to read.         */
        /***************************************************************/
        CurPos = recpos;
      } // endswitch recpos

    if (trace)
      htrc("Position is now %d\n", CurPos);

  } else
    /*******************************************************************/
    /*  !To_Kindex ---> sequential reading                             */
    /*******************************************************************/
    rc = (++CurPos < Qrp->Nblin) ? RC_OK : RC_EF;

  return rc;
  } // end of ReadDB

/***********************************************************************/
/*  Dummy WriteDB: just send back an error return.                     */
/***********************************************************************/
int TDBQRS::WriteDB(PGLOBAL g)
  {
  strcpy(g->Message, MSG(QRY_READ_ONLY));
  return RC_FX;
  } // end of WriteDB

/***********************************************************************/
/*  Dummy DeleteDB routine, just returns an error code.                */
/***********************************************************************/
int TDBQRS::DeleteDB(PGLOBAL g, int irc)
  {
  strcpy(g->Message, MSG(NO_QRY_DELETE));
  return RC_FX;
  } // end of DeleteDB

/***********************************************************************/
/*  Data Base close routine for QRS access method.                     */
/***********************************************************************/
void TDBQRS::CloseDB(PGLOBAL g)
  {
  if (To_Kindex) {
    To_Kindex->Close();
    To_Kindex = NULL;
    } // endif

  if (trace)
    htrc("Qryres CloseDB");

//Qryp->Sqlp->CloseDB();
  } // end of CloseDB

// ------------------------ QRSCOL functions ----------------------------

/***********************************************************************/
/*  QRSCOL public constructor.                                         */
/***********************************************************************/
QRSCOL::QRSCOL(PGLOBAL g, PCOLRES crp, PTDB tdbp, PCOL cprec, int i)
      : COLBLK(NULL, tdbp, i) 
  {
  if (cprec) {
    Next = cprec->GetNext();
    cprec->SetNext(this);
  } else {
    Next = tdbp->GetColumns();
    tdbp->SetColumns(this);
  } // endif cprec

  // Set additional QRS access method information for column.
  Crp = crp;
  Name = Crp->Name;
  Long = Crp->Clen;
  Buf_Type = crp->Type;
  strcpy(Format.Type, GetFormatType(Buf_Type));
  Format.Length = (SHORT)Long;
  Format.Prec = (SHORT)Crp->Prec;

  if (trace) {
    htrc("Making new QRSCOL C%d %s at %p\n", Index, Name, this);
    htrc(" BufType=%d Long=%d length=%d clen=%d\n",
          Buf_Type, Long, Format.Length, Crp->Clen);
    } // endif trace

  } // end of QRSCOL constructor

/***********************************************************************/
/*  QRSCOL constructor used for copying columns.                       */
/*  tdbp is the pointer to the new table descriptor.                   */
/***********************************************************************/
QRSCOL::QRSCOL(QRSCOL *col1, PTDB tdbp) : COLBLK(col1, tdbp)
  {
  Crp = col1->Crp;
  } // end of QRSCOL copy constructor

/***********************************************************************/
/*  ReadColumn: what this routine does is to extract the RESCOL block  */
/*  current value and convert it to the column buffer type.            */
/***********************************************************************/
void QRSCOL::ReadColumn(PGLOBAL g)
  {
  PTDBQRS tdbp = (PTDBQRS)To_Tdb;

  if (trace)
    htrc("QRS RC: col %s R%d type=%d CurPos=%d Len=%d\n",
          Name, tdbp->GetTdb_No(), Buf_Type, tdbp->CurPos, Crp->Clen);

  if (Crp->Kdata)
    Value->SetValue_pvblk(Crp->Kdata, tdbp->CurPos);
  else
    Value->Reset();

  } // end of ReadColumn

/***********************************************************************/
/*  Make file output of a Dos column descriptor block.                 */
/***********************************************************************/
void QRSCOL::Print(PGLOBAL g, FILE *f, UINT n)
  {
  COLBLK::Print(g, f, n);

  fprintf(f, " Crp=%p\n", Crp);
  } // end of Print

/* --------------------- End of TabPivot/TabQrs ---------------------- */