1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

- Extending connect_assisted_discovery column automatic definition

Browse Source 
to OCCUR and PIVOT table types.

modified:
  storage/connect/ha_connect.cc
  storage/connect/myconn.cpp
  storage/connect/myconn.h
  storage/connect/plgdbsem.h
  storage/connect/plgdbutl.cpp
  storage/connect/taboccur.cpp
  storage/connect/taboccur.h
  storage/connect/tabpivot.cpp
  storage/connect/tabpivot.h

- Fix wrong definition of GetVlen for TYPE template

modified:
  storage/connect/valblk.h
This commit is contained in:
Olivier Bertrand 2013-05-28 17:22:38 +02:00
parent cc8174db48
commit 9f7c3fedfa
10 changed files with 494 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
storage/connect/ha_connect.cc
View File

@ -27,11 +27,12 @@
ha_connect will let you create/open/delete tables, the created table can be ha_connect will let you create/open/delete tables, the created table can be
done specifying an already existing file, the drop table command will just done specifying an already existing file, the drop table command will just
suppress the table definition but not the eventual data file. suppress the table definition but not the eventual data file.
Indexes are not yet supported but data can be inserted, updated or deleted. Indexes are not supported for all table types but data can be inserted,
updated or deleted.
You can enable the CONNECT storage engine in your build by doing the You can enable the CONNECT storage engine in your build by doing the
following during your build process:<br> ./configure following during your build process:<br> ./configure
--with-connect-storage-engine (not implemented yet) --with-connect-storage-engine
You can install the CONNECT handler as all other storage handlers. You can install the CONNECT handler as all other storage handlers.
@ -166,6 +167,16 @@ extern "C" {
int trace= 0; // The general trace value int trace= 0; // The general trace value
} // extern "C" } // extern "C"
bool OcrColumns(PGLOBAL g, PQRYRES qrp, const char *col,
const char *ocr, const char *rank);
bool OcrSrcCols(PGLOBAL g, PQRYRES qrp, const char *col,
const char *ocr, const char *rank);
PQRYRES PivotColumns(PGLOBAL g, const char *tab, const char *src,
const char *picol, const char *fncol,
const char *host, const char *db,
const char *user, const char *pwd,
int port);
/****************************************************************************/ /****************************************************************************/
/* Initialize the ha_connect static members. */ /* Initialize the ha_connect static members. */
/****************************************************************************/ /****************************************************************************/
@ -3333,7 +3344,7 @@ static char *encode(PGLOBAL g, char *cnm)
*/ */
static bool add_field(String *sql, const char *field_name, const char *type, static bool add_field(String *sql, const char *field_name, const char *type,
int len, int dec, uint tm, const char *rem) int len, int dec, uint tm, const char *rem, int flag)
{ {
bool error= false; bool error= false;
@ -3341,15 +3352,18 @@ static bool add_field(String *sql, const char *field_name, const char *type,
error|= sql->append(field_name); error|= sql->append(field_name);
error|= sql->append("` "); error|= sql->append("` ");
error|= sql->append(type); error|= sql->append(type);
if (len) { if (len) {
error|= sql->append('('); error|= sql->append('(');
error|= sql->append_ulonglong(len); error|= sql->append_ulonglong(len);
if (dec || !strcmp(type, "DOUBLE")) { if (dec || !strcmp(type, "DOUBLE")) {
error|= sql->append(','); error|= sql->append(',');
error|= sql->append_ulonglong(dec); error|= sql->append_ulonglong(dec);
} } // endif dec
error|= sql->append(')'); error|= sql->append(')');
} } // endif len
if (tm) if (tm)
error|= sql->append(STRING_WITH_LEN(" NOT NULL"), system_charset_info); error|= sql->append(STRING_WITH_LEN(" NOT NULL"), system_charset_info);
@ -3358,10 +3372,14 @@ static bool add_field(String *sql, const char *field_name, const char *type,
error|= sql->append(" COMMENT '"); error|= sql->append(" COMMENT '");
error|= sql->append_for_single_quote(rem, strlen(rem)); error|= sql->append_for_single_quote(rem, strlen(rem));
error|= sql->append("'"); error|= sql->append("'");
} } // endif rem
if (flag) {
error|= sql->append(" FLAG=");
error|= sql->append_ulonglong(flag);
} // endif flag
sql->append(','); sql->append(',');
return error; return error;
} // end of add_field } // end of add_field
@ -3381,6 +3399,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
char spc= ',', qch= 0; char spc= ',', qch= 0;
const char *fncn= "?"; const char *fncn= "?";
const char *user, *fn, *db, *host, *pwd, *prt, *sep, *tbl, *src; const char *user, *fn, *db, *host, *pwd, *prt, *sep, *tbl, *src;
const char *col, *ocl, *rnk, *pic, *fcl;
char *tab, *dsn; char *tab, *dsn;
#if defined(WIN32) #if defined(WIN32)
char *nsp= NULL, *cls= NULL; char *nsp= NULL, *cls= NULL;
@ -3402,7 +3421,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
sql.copy(STRING_WITH_LEN("CREATE TABLE whatever ("), system_charset_info); sql.copy(STRING_WITH_LEN("CREATE TABLE whatever ("), system_charset_info);
user= host= pwd= prt= tbl= src= dsn= NULL; user= host= pwd= prt= tbl= src= col= ocl= pic= fcl= rnk= dsn= NULL;
// Get the useful create options // Get the useful create options
ttp= GetTypeID(topt->type); ttp= GetTypeID(topt->type);
@ -3417,12 +3436,18 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
qch= topt->qchar ? *topt->qchar : topt->quoted >= 0 ? '"' : 0; qch= topt->qchar ? *topt->qchar : topt->quoted >= 0 ? '"' : 0;
hdr= (int)topt->header; hdr= (int)topt->header;
tbl= topt->tablist; tbl= topt->tablist;
col= topt->colist;
if (topt->oplist) { if (topt->oplist) {
host= GetListOption(g,"host", topt->oplist, "localhost"); host= GetListOption(g,"host", topt->oplist, "localhost");
user= GetListOption(g,"user", topt->oplist, "root"); user= GetListOption(g,"user", topt->oplist, "root");
// Default value db can come from the DBNAME=xxx option. // Default value db can come from the DBNAME=xxx option.
db= GetListOption(g,"database", topt->oplist, db); db= GetListOption(g,"database", topt->oplist, db);
col= GetListOption(g,"colist", topt->oplist, col);
ocl= GetListOption(g,"occurcol", topt->oplist, NULL);
pic= GetListOption(g,"pivotcol", topt->oplist, NULL);
fcl= GetListOption(g,"fnccol", topt->oplist, NULL);
rnk= GetListOption(g,"rankcol", topt->oplist, NULL);
pwd= GetListOption(g,"password", topt->oplist); pwd= GetListOption(g,"password", topt->oplist);
prt= GetListOption(g,"port", topt->oplist); prt= GetListOption(g,"port", topt->oplist);
port= (prt) ? atoi(prt) : 0; port= (prt) ? atoi(prt) : 0;
@ -3537,9 +3562,12 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
ok= true; ok= true;
break; break;
#endif // WIN32 #endif // WIN32
case TAB_PIVOT:
supfnc = FNC_NO;
case TAB_PRX: case TAB_PRX:
case TAB_TBL: case TAB_TBL:
case TAB_XCL: case TAB_XCL:
case TAB_OCCUR:
ok= true; ok= true;
break; break;
default: default:
@ -3563,7 +3591,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
if (ok) { if (ok) {
char *cnm, *rem; char *cnm, *rem;
int i, len, dec, typ; int i, len, dec, typ, flg;
const char *type; const char *type;
PDBUSER dup= PlgGetUser(g); PDBUSER dup= PlgGetUser(g);
PCATLG cat= (dup) ? dup->Catalog : NULL; PCATLG cat= (dup) ? dup->Catalog : NULL;
@ -3573,9 +3601,16 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
else else
return HA_ERR_INTERNAL_ERROR; // Should never happen return HA_ERR_INTERNAL_ERROR; // Should never happen
if (src) if (src && ttp != TAB_PIVOT) {
qrp= SrcColumns(g, host, db, user, pwd, src, port); qrp= SrcColumns(g, host, db, user, pwd, src, port);
else switch (ttp) {
if (ttp == TAB_OCCUR)
if (OcrSrcCols(g, qrp, col, ocl, rnk)) {
my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
return HA_ERR_INTERNAL_ERROR;
} // endif OcrSrcCols
} else switch (ttp) {
case TAB_DBF: case TAB_DBF:
qrp= DBFColumns(g, fn, fnc == FNC_COL); qrp= DBFColumns(g, fn, fnc == FNC_COL);
break; break;
@ -3618,12 +3653,22 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
case TAB_PRX: case TAB_PRX:
case TAB_TBL: case TAB_TBL:
case TAB_XCL: case TAB_XCL:
case TAB_OCCUR:
bif= fnc == FNC_COL; bif= fnc == FNC_COL;
qrp= TabColumns(g, thd, db, tab, bif); qrp= TabColumns(g, thd, db, tab, bif);
if (!qrp && bif && fnc != FNC_COL) // tab is a view if (!qrp && bif && fnc != FNC_COL) // tab is a view
qrp= MyColumns(g, host, db, user, pwd, tab, NULL, port, false); qrp= MyColumns(g, host, db, user, pwd, tab, NULL, port, false);
if (ttp == TAB_OCCUR && fnc != FNC_COL)
if (OcrColumns(g, qrp, col, ocl, rnk)) {
my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
return HA_ERR_INTERNAL_ERROR;
} // endif OcrColumns
break;
case TAB_PIVOT:
qrp= PivotColumns(g, tab, src, pic, fcl, host, db, user, pwd, port);
break; break;
default: default:
strcpy(g->Message, "System error during assisted discovery"); strcpy(g->Message, "System error during assisted discovery");
@ -3635,16 +3680,17 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
return HA_ERR_INTERNAL_ERROR; return HA_ERR_INTERNAL_ERROR;
} // endif qrp } // endif qrp
if (fnc != FNC_NO || src) { if (fnc != FNC_NO || src || ttp == TAB_PIVOT) {
// Catalog table // Catalog like table
for (crp=qrp->Colresp; !b && crp; crp= crp->Next) { for (crp=qrp->Colresp; !b && crp; crp= crp->Next) {
cnm= encode(g, crp->Name); cnm= encode(g, crp->Name);
type= PLGtoMYSQLtype(crp->Type, dbf); type= PLGtoMYSQLtype(crp->Type, dbf);
len= crp->Length; len= crp->Length;
dec= crp->Prec; dec= crp->Prec;
flg= crp->Flag;
// Now add the field // Now add the field
if (add_field(&sql, cnm, type, len, dec, NOT_NULL_FLAG, 0)) if (add_field(&sql, cnm, type, len, dec, NOT_NULL_FLAG, 0, flg))
b= HA_ERR_OUT_OF_MEM; b= HA_ERR_OUT_OF_MEM;
} // endfor crp } // endfor crp
@ -3714,7 +3760,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
len= 0; len= 0;
// Now add the field // Now add the field
if (add_field(&sql, cnm, type, len, dec, tm, rem)) if (add_field(&sql, cnm, type, len, dec, tm, rem, 0))
b= HA_ERR_OUT_OF_MEM; b= HA_ERR_OUT_OF_MEM;
} // endfor i } // endfor i

9
storage/connect/myconn.cpp
View File

@ -64,6 +64,12 @@ static char *server_groups[] = {
extern "C" int trace; extern "C" int trace;
extern MYSQL_PLUGIN_IMPORT uint mysqld_port; extern MYSQL_PLUGIN_IMPORT uint mysqld_port;
// Returns the current used port
uint GetDefaultPort(void)
{
return mysqld_port;
} // end of GetDefaultPort
/************************************************************************/ /************************************************************************/
/* MyColumns: constructs the result blocks containing all columns */ /* MyColumns: constructs the result blocks containing all columns */
/* of a MySQL table or view. */ /* of a MySQL table or view. */
@ -673,6 +679,7 @@ PQRYRES MYSQLC::GetResult(PGLOBAL g, bool pdb)
*pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES)); *pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
crp = *pcrp; crp = *pcrp;
pcrp = &crp->Next; pcrp = &crp->Next;
memset(crp, 0, sizeof(COLRES));
crp->Ncol = ++qrp->Nbcol; crp->Ncol = ++qrp->Nbcol;
crp->Name = (char*)PlugSubAlloc(g, NULL, fld->name_length + 1); crp->Name = (char*)PlugSubAlloc(g, NULL, fld->name_length + 1);
@ -686,7 +693,7 @@ PQRYRES MYSQLC::GetResult(PGLOBAL g, bool pdb)
// For direct MySQL connection, display the MySQL date string // For direct MySQL connection, display the MySQL date string
crp->Type = TYPE_STRING; crp->Type = TYPE_STRING;
crp->Prec = fld->decimals; crp->Prec = (crp->Type == TYPE_FLOAT) ? fld->decimals : 0;
crp->Length = fld->max_length; crp->Length = fld->max_length;
crp->Clen = GetTypeSize(crp->Type, crp->Length); crp->Clen = GetTypeSize(crp->Type, crp->Length);

2
storage/connect/myconn.h
View File

@ -44,6 +44,8 @@ PQRYRES SrcColumns(PGLOBAL g, const char *host, const char *db,
const char *user, const char *pwd, const char *user, const char *pwd,
const char *srcdef, int port); const char *srcdef, int port);
uint GetDefaultPort(void);
/* -------------------------- MYCONN class --------------------------- */ /* -------------------------- MYCONN class --------------------------- */
/***********************************************************************/ /***********************************************************************/

1
storage/connect/plgdbsem.h
View File

@ -518,6 +518,7 @@ typedef struct _colres {
int Clen; /* Data individual internal size */ int Clen; /* Data individual internal size */
int Length; /* Data individual print length */ int Length; /* Data individual print length */
int Prec; /* Precision */ int Prec; /* Precision */
int Flag; /* Flag option value */
XFLD Fld; /* Type of field info */ XFLD Fld; /* Type of field info */
} COLRES; } COLRES;

1
storage/connect/plgdbutl.cpp
View File

@ -298,6 +298,7 @@ PQRYRES PlgAllocResult(PGLOBAL g, int ncol, int maxres, int ids,
*pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES)); *pcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
crp = *pcrp; crp = *pcrp;
pcrp = &crp->Next; pcrp = &crp->Next;
memset(crp, 0, sizeof(COLRES));
crp->Colp = NULL; crp->Colp = NULL;
crp->Ncol = ++qrp->Nbcol; crp->Ncol = ++qrp->Nbcol;
crp->Type = buftyp[i]; crp->Type = buftyp[i];

234
storage/connect/taboccur.cpp
View File

@ -1,5 +1,5 @@
/************ TabOccur CPP Declares Source Code File (.CPP) ************/ /************ TabOccur CPP Declares Source Code File (.CPP) ************/
/* Name: TABOCCUR.CPP Version 1.0 */ /* Name: TABOCCUR.CPP Version 1.1 */
/* */ /* */
/* (C) Copyright to the author Olivier BERTRAND 2013 */ /* (C) Copyright to the author Olivier BERTRAND 2013 */
/* */ /* */
@ -53,6 +53,210 @@
extern "C" int trace; extern "C" int trace;
/***********************************************************************/
/* Prepare and count columns in the column list. */
/***********************************************************************/
int PrepareColist(char *colist)
{
char *p, *pn;
int n = 0;
// Count the number of columns and change separator into null char
for (pn = colist; ; pn += (strlen(pn) + 1))
// Separator can be ; if colist was specified in the option_list
if ((p = strchr(pn, ',')) || (p = strchr(pn, ';'))) {
*p++ = '\0';
n++;
} else {
if (*pn)
n++;
break;
} // endif p
return n;
} // end of PrepareColist
/************************************************************************/
/* OcrColumns: constructs the result blocks containing all the columns */
/* of the object table that will be retrieved by GetData commands. */
/************************************************************************/
bool OcrColumns(PGLOBAL g, PQRYRES qrp, const char *col,
const char *ocr, const char *rank)
{
char *pn, *colist;
int i, k, m, n = 0, c = 0, j = qrp->Nblin;
bool rk, b = false;
PCOLRES crp;
if (!col || !*col) {
strcpy(g->Message, "Missing colist");
return true;
} // endif col
// Prepare the column list
colist = (char*)PlugSubAlloc(g, NULL, strlen(col) + 1);
strcpy(colist, col);
m = PrepareColist(colist);
if ((rk = (rank && *rank))) {
if (m == 1) {
strcpy(g->Message, "Cannot handle one column colist and rank");
return true;
} // endif m
for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
n = max(n, (signed)strlen(pn));
} // endif k
// Default occur column name is the 1st colist column name
if (!ocr || !*ocr)
ocr = colist;
/**********************************************************************/
/* Replace the columns of the colist by the rank and occur columns. */
/**********************************************************************/
for (i = 0; i < qrp->Nblin; i++) {
for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
if (!stricmp(pn, qrp->Colresp->Kdata->GetCharValue(i)))
break;
if (k < m) {
// This column belongs to colist
if (rk) {
// Place the rank column here
for (crp = qrp->Colresp; crp; crp = crp->Next)
switch (crp->Fld) {
case FLD_NAME: crp->Kdata->SetValue((char*)rank, i); break;
case FLD_TYPE: crp->Kdata->SetValue(TYPE_STRING, i); break;
case FLD_PREC: crp->Kdata->SetValue(n, i); break;
case FLD_SCALE: crp->Kdata->SetValue(0, i); break;
case FLD_NULL: crp->Kdata->SetValue(0, i); break;
case FLD_REM: crp->Kdata->Reset(i); break;
default: ; // Ignored by CONNECT
} // endswich Fld
rk = false;
} else if (!b) {
// First remaining listed column, will be the occur column
for (crp = qrp->Colresp; crp; crp = crp->Next)
switch (crp->Fld) {
case FLD_NAME: crp->Kdata->SetValue((char*)ocr, i); break;
case FLD_REM: crp->Kdata->Reset(i); break;
default: ; // Nothing to do
} // endswich Fld
b = true;
} else if (j == qrp->Nblin)
j = i; // Column to remove
c++;
} else if (j < i) {
// Move this column in empty spot
for (crp = qrp->Colresp; crp; crp = crp->Next)
crp->Kdata->Move(i, j);
j++;
} // endif k
} // endfor i
// Check whether all columns of the list where found
if (c < m) {
strcpy(g->Message, "Some colist columns are not in the source table");
return true;
} // endif crp
/**********************************************************************/
/* Set the number of columns of the table. */
/**********************************************************************/
qrp->Nblin = j;
return false;
} // end of OcrColumns
/************************************************************************/
/* OcrSrcCols: constructs the result blocks containing all the columns */
/* of the object table that will be retrieved by GetData commands. */
/************************************************************************/
bool OcrSrcCols(PGLOBAL g, PQRYRES qrp, const char *col,
const char *ocr, const char *rank)
{
char *pn, *colist;
int i, k, m, n = 0, c = 0;
bool rk, b = false;
PCOLRES crp, rcrp, *pcrp;
if (!col || !*col) {
strcpy(g->Message, "Missing colist");
return true;
} // endif col
// Prepare the column list
colist = (char*)PlugSubAlloc(g, NULL, strlen(col) + 1);
strcpy(colist, col);
m = PrepareColist(colist);
if ((rk = (rank && *rank)))
for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
n = max(n, (signed)strlen(pn));
// Default occur column name is the 1st colist column name
if (!ocr || !*ocr)
ocr = colist;
/**********************************************************************/
/* Replace the columns of the colist by the rank and occur columns. */
/**********************************************************************/
for (i = 0, pcrp = &qrp->Colresp; crp = *pcrp; ) {
for (k = 0, pn = colist; k < m; k++, pn += (strlen(pn) + 1))
if (!stricmp(pn, crp->Name))
break;
if (k < m) {
// This column belongs to colist
c++;
if (!b) {
if (rk) {
// Add the rank column here
rcrp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
memset(rcrp, 0, sizeof(COLRES));
rcrp->Next = crp;
rcrp->Name = (char*)rank;
rcrp->Type = TYPE_STRING;
rcrp->Length = n;
rcrp->Ncol = ++i;
*pcrp = rcrp;
} // endif rk
// First remaining listed column, will be the occur column
crp->Name = (char*)ocr;
b = true;
} else {
*pcrp = crp->Next; // Remove this column
continue;
} // endif b
} // endif k
crp->Ncol = ++i;
pcrp = &crp->Next;
} // endfor pcrp
// Check whether all columns of the list where found
if (c < m) {
strcpy(g->Message, "Some colist columns are not in the source table");
return true;
} // endif crp
/**********************************************************************/
/* Set the number of columns of the table. */
/**********************************************************************/
qrp->Nblin = i;
return false;
} // end of OcrSrcCols
/* -------------- Implementation of the OCCUR classes ---------------- */ /* -------------- Implementation of the OCCUR classes ---------------- */
/***********************************************************************/ /***********************************************************************/
@ -60,9 +264,9 @@ extern "C" int trace;
/***********************************************************************/ /***********************************************************************/
bool OCCURDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff) bool OCCURDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
{ {
Xcol = Cat->GetStringCatInfo(g, "OccurCol", "");
Rcol = Cat->GetStringCatInfo(g, "RankCol", ""); Rcol = Cat->GetStringCatInfo(g, "RankCol", "");
Colist = Cat->GetStringCatInfo(g, "Colist", ""); Colist = Cat->GetStringCatInfo(g, "Colist", "");
Xcol = Cat->GetStringCatInfo(g, "OccurCol", Colist);
return PRXDEF::DefineAM(g, am, poff); return PRXDEF::DefineAM(g, am, poff);
} // end of DefineAM } // end of DefineAM
@ -92,36 +296,12 @@ TDBOCCUR::TDBOCCUR(POCCURDEF tdp) : TDBPRX(tdp)
Rcolumn = tdp->Rcol; // Rank column name Rcolumn = tdp->Rcol; // Rank column name
Xcolp = NULL; // To the OCCURCOL column Xcolp = NULL; // To the OCCURCOL column
Col = NULL; // To source column blocks array Col = NULL; // To source column blocks array
Mult = PrepareColist(); // Multiplication factor Mult = PrepareColist(Colist); // Multiplication factor
N = 0; // The current table index N = 0; // The current table index
M = 0; // The occurence rank M = 0; // The occurence rank
RowFlag = 0; // 0: Ok, 1: Same, 2: Skip RowFlag = 0; // 0: Ok, 1: Same, 2: Skip
} // end of TDBOCCUR constructor } // end of TDBOCCUR constructor
/***********************************************************************/
/* Prepare and count columns in the column list. */
/***********************************************************************/
int TDBOCCUR::PrepareColist(void)
{
char *p, *pn;
int n = 0;
// Count the number of columns and change separator into null char
for (pn = Colist; ; pn += (strlen(pn) + 1))
// Separator can be ; if colist was specified in the option_list
if ((p = strchr(pn, ',')) || (p = strchr(pn, ';'))) {
*p++ = '\0';
n++;
} else {
if (*pn)
n++;
break;
} // endif p
return n;
} // end of PrepareColist
/***********************************************************************/ /***********************************************************************/
/* Allocate OCCUR/SRC column description block. */ /* Allocate OCCUR/SRC column description block. */
/***********************************************************************/ /***********************************************************************/

1
storage/connect/taboccur.h
View File

@ -58,7 +58,6 @@ class TDBOCCUR : public TDBPRX {
// Methods // Methods
virtual void ResetDB(void) {N = 0; Tdbp->ResetDB();} virtual void ResetDB(void) {N = 0; Tdbp->ResetDB();}
virtual int RowNumber(PGLOBAL g, bool b = FALSE); virtual int RowNumber(PGLOBAL g, bool b = FALSE);
int PrepareColist(void);
bool MakeColumnList(PGLOBAL g); bool MakeColumnList(PGLOBAL g);
bool ViewColumnList(PGLOBAL g); bool ViewColumnList(PGLOBAL g);

178
storage/connect/tabpivot.cpp
View File

@ -1,7 +1,7 @@
/************ TabPivot C++ Program Source Code File (.CPP) *************/ /************ TabPivot C++ Program Source Code File (.CPP) *************/
/* PROGRAM NAME: TABPIVOT */ /* PROGRAM NAME: TABPIVOT */
/* ------------- */ /* ------------- */
/* Version 1.5 */ /* Version 1.6 */
/* */ /* */
/* COPYRIGHT: */ /* COPYRIGHT: */
/* ---------- */ /* ---------- */
@ -53,6 +53,182 @@
extern "C" int trace; extern "C" int trace;
/***********************************************************************/
/* Make the Pivot table column list. */
/***********************************************************************/
PQRYRES PivotColumns(PGLOBAL g, const char *tab, const char *src,
const char *picol, const char *fncol,
const char *host, const char *db,
const char *user, const char *pwd,
int port)
{
PIVAID pvd(tab, src, picol, fncol, host, db, user, pwd, port);
return pvd.MakePivotColumns(g);
} // end of PivotColumns
/* --------------- Implementation of the PIVAID classe --------------- */
/***********************************************************************/
/* PIVAID constructor. */
/***********************************************************************/
PIVAID::PIVAID(const char *tab, const char *src, const char *picol,
const char *fncol, const char *host, const char *db,
const char *user, const char *pwd, int port)
: CSORT(false)
{
Host = (char*)host;
User = (char*)user;
Pwd = (char*)pwd;
Qryp = NULL;
Database = (char*)db;
Tabname = (char*)tab;
Tabsrc = (char*)src;
Picol = (char*)picol;
Fncol = (char*)fncol;
Rblkp = NULL;
Port = (port) ? port : GetDefaultPort();
} // end of PIVAID constructor
/***********************************************************************/
/* Make the Pivot table column list. */
/***********************************************************************/
PQRYRES PIVAID::MakePivotColumns(PGLOBAL g)
{
char *query, *colname, buf[32];
int ndif, nblin, w = 0;
PVAL valp;
PCOLRES *pcrp, crp, fncrp = NULL;
if (!Tabsrc && Tabname) {
// Locate the query
query = (char*)PlugSubAlloc(g, NULL, strlen(Tabname) + 16);
sprintf(query, "SELECT * FROM %s", Tabname);
} else if (!Tabsrc) {
strcpy(g->Message, MSG(SRC_TABLE_UNDEF));
return NULL;
} else
query = Tabsrc;
// 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, query, &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();
if (!Fncol) {
for (crp = Qryp->Colresp; crp; crp = crp->Next)
if (!Picol || stricmp(Picol, crp->Name))
Fncol = crp->Name;
if (!Fncol) {
strcpy(g->Message, MSG(NO_DEF_FNCCOL));
return NULL;
} // endif Fncol
} // endif Fncol
if (!Picol) {
// Find default Picol as the last one not equal to Fncol
for (crp = Qryp->Colresp; crp; crp = crp->Next)
if (stricmp(Fncol, crp->Name))
Picol = crp->Name;
if (!Picol) {
strcpy(g->Message, MSG(NO_DEF_PIVOTCOL));
return NULL;
} // endif Picol
} // endif picol
// Prepare the column list
for (pcrp = &Qryp->Colresp; crp = *pcrp; )
if (!stricmp(Picol, crp->Name)) {
Rblkp = crp->Kdata;
*pcrp = crp->Next;
} else if (!stricmp(Fncol, crp->Name)) {
fncrp = crp;
*pcrp = crp->Next;
} else
pcrp = &crp->Next;
if (!Rblkp) {
strcpy(g->Message, MSG(NO_DEF_PIVOTCOL));
return NULL;
} else if (!fncrp) {
strcpy(g->Message, MSG(NO_DEF_FNCCOL));
return NULL;
} // endif
// Before calling sort, initialize all
nblin = Qryp->Nblin;
Index.Size = nblin * sizeof(int);
Index.Sub = TRUE; // Should be small enough
if (!PlgDBalloc(g, NULL, Index))
return NULL;
Offset.Size = (nblin + 1) * sizeof(int);
Offset.Sub = TRUE; // Should be small enough
if (!PlgDBalloc(g, NULL, Offset))
return NULL;
ndif = Qsort(g, nblin);
if (ndif < 0) // error
return NULL;
// Allocate the Value used to retieve column names
if (!(valp = AllocateValue(g, Rblkp->GetType(),
Rblkp->GetVlen(),
Rblkp->GetPrec())))
return NULL;
// Now make the functional columns
for (int i = 0; i < ndif; i++) {
if (i) {
crp = (PCOLRES)PlugSubAlloc(g, NULL, sizeof(COLRES));
memcpy(crp, fncrp, sizeof(COLRES));
} else
crp = fncrp;
// Get the value that will be the generated column name
valp->SetValue_pvblk(Rblkp, Pex[Pof[i]]);
colname = valp->GetCharString(buf);
crp->Name = (char*)PlugSubAlloc(g, NULL, strlen(colname) + 1);
strcpy(crp->Name, colname);
crp->Flag = 1;
// Add this column
*pcrp = crp;
crp->Next = NULL;
pcrp = &crp->Next;
} // endfor i
// We added ndif columns and removed 2 (picol and fncol)
Qryp->Nbcol += (ndif - 2);
return Qryp;
} // end of MakePivotColumns
/***********************************************************************/
/* PIVAID: Compare routine for sorting pivot column values. */
/***********************************************************************/
int PIVAID::Qcompare(int *i1, int *i2)
{
// TODO: the actual comparison between pivot column result values.
return Rblkp->CompVal(*i1, *i2);
} // end of Qcompare
/* --------------- Implementation of the PIVOT classes --------------- */ /* --------------- Implementation of the PIVOT classes --------------- */
/***********************************************************************/ /***********************************************************************/

36
storage/connect/tabpivot.h
View File

@ -1,5 +1,5 @@
/************** TabPivot H Declares Source Code File (.H) **************/ /************** TabPivot H Declares Source Code File (.H) **************/
/* Name: TABPIVOT.H Version 1.4 */ /* Name: TABPIVOT.H Version 1.5 */
/* */ /* */
/* (C) Copyright to the author Olivier BERTRAND 2005-2013 */ /* (C) Copyright to the author Olivier BERTRAND 2005-2013 */
/* */ /* */
@ -10,6 +10,40 @@ typedef class TDBPIVOT *PTDBPIVOT;
typedef class FNCCOL *PFNCCOL; typedef class FNCCOL *PFNCCOL;
typedef class SRCCOL *PSRCCOL; typedef class SRCCOL *PSRCCOL;
/***********************************************************************/
/* This class is used to generate PIVOT table column definitions. */
/***********************************************************************/
class PIVAID : public CSORT {
friend class FNCCOL;
friend class SRCCOL;
public:
// Constructor
PIVAID(const char *tab, const char *src, const char *picol,
const char *fncol, const char *host, const char *db,
const char *user, const char *pwd, int port);
// Methods
PQRYRES MakePivotColumns(PGLOBAL g);
// The sorting function
virtual int Qcompare(int *, int *);
protected:
// Members
MYSQLC Myc; // MySQL connection class
char *Host; // Host machine to use
char *User; // User logon info
char *Pwd; // Password logon info
char *Database; // Database to be used by server
PQRYRES Qryp; // Points to Query result block
char *Tabname; // Name of source table
char *Tabsrc; // SQL of source table
char *Picol; // Pivot column name
char *Fncol; // Function column name
PVBLK Rblkp; // The value block of the pivot column
int Port; // MySQL port number
}; // end of class PIVAID
/* -------------------------- PIVOT classes -------------------------- */ /* -------------------------- PIVOT classes -------------------------- */
/***********************************************************************/ /***********************************************************************/

3
storage/connect/valblk.h
View File

@ -36,6 +36,7 @@ class VALBLK : public BLOCK {
void *GetValPointer(void) {return Blkp;} void *GetValPointer(void) {return Blkp;}
void SetValPointer(void *mp) {Blkp = mp;} void SetValPointer(void *mp) {Blkp = mp;}
int GetType(void) {return Type;} int GetType(void) {return Type;}
int GetPrec(void) {return Prec;}
void SetCheck(bool b) {Check = b;} void SetCheck(bool b) {Check = b;}
void MoveNull(int i, int j) void MoveNull(int i, int j)
{if (To_Nulls) To_Nulls[j] = To_Nulls[j];} {if (To_Nulls) To_Nulls[j] = To_Nulls[j];}
@ -110,7 +111,7 @@ class TYPBLK : public VALBLK {
// Implementation // Implementation
virtual void Init(PGLOBAL g, bool check); virtual void Init(PGLOBAL g, bool check);
virtual int GetVlen(void) {return sizeof(int);} virtual int GetVlen(void) {return sizeof(TYPE);}
//virtual PSZ GetCharValue(int n); //virtual PSZ GetCharValue(int n);
virtual short GetShortValue(int n) {return (short)Typp[n];} virtual short GetShortValue(int n) {return (short)Typp[n];}
virtual int GetIntValue(int n) {return (int)Typp[n];} virtual int GetIntValue(int n) {return (int)Typp[n];}