Skip to content

M
MariaDB
Commit c035bde3 authored by Olivier Bertrand's avatar Olivier Bertrand
Browse files
Options

- Allowing views and queries as parameters for PROXY base tables 

  NOTE: Checking for looping references cannot be done when using views
  as parameters. This should not be allowed on production servers and
  should be dependant on a system variable and/or on speciel grant.

modified:
  storage/connect/CMakeLists.txt
  storage/connect/connect.cc
  storage/connect/ha_connect.cc
  storage/connect/myconn.cpp
  storage/connect/myconn.h
  storage/connect/mysql-test/connect/r/fmt.result
  storage/connect/mysql-test/connect/r/pivot.result
  storage/connect/mysql-test/connect/t/fmt.test
  storage/connect/mysql-test/connect/t/pivot.test
  storage/connect/plgdbsem.h
  storage/connect/plugutil.c
  storage/connect/tabcol.cpp
  storage/connect/tabcol.h
  storage/connect/tabfmt.cpp
  storage/connect/tabmysql.cpp
  storage/connect/tabmysql.h
  storage/connect/taboccur.cpp
  storage/connect/taboccur.h
  storage/connect/tabpivot.cpp
  storage/connect/tabpivot.h
  storage/connect/tabtbl.cpp
  storage/connect/tabutil.cpp
  storage/connect/tabutil.h
  storage/connect/xtable.h
parent 3c76e0e2
Show whitespace changes
Inline Side-by-side
Showing with 900 additions and 1384 deletions
storage/connect/CMakeLists.txt
View file @ c035bde3
......@@ -178,7 +178,7 @@ OPTION(CONNECT_WITH_MYSQL
IF(CONNECT_WITH_MYSQL)
SET(CONNECT_SOURCES ${CONNECT_SOURCES}
myconn.cpp myconn.h tabmysql.cpp tabxml.h)
myconn.cpp myconn.h tabmysql.cpp tabmysql.h)
add_definitions(-DMYSQL_SUPPORT)
IF(NOT UNIX)
#
......
storage/connect/connect.cc
View file @ c035bde3
......@@ -71,10 +71,7 @@ int rename_file_ext(const char *from, const char *to,const char *ext);
PGLOBAL CntExit(PGLOBAL g)
{
if (g) {
PDBUSER dup= PlgGetUser(g);
CntEndDB(g);
free(dup);
if (g->Activityp)
delete g->Activityp;
......@@ -94,13 +91,10 @@ void CntEndDB(PGLOBAL g)
PDBUSER dbuserp= PlgGetUser(g);
if (dbuserp) {
if (dbuserp->Catalog) {
if (dbuserp->Catalog)
delete dbuserp->Catalog;
dbuserp->Catalog= NULL;
} // endif Catalog
*dbuserp->Name= '\0';
// *dbuserp->Work= '\0';
free(dbuserp);
} // endif dbuserp
} // end of CntEndDB
......@@ -258,10 +252,12 @@ bool CntOpenTable(PGLOBAL g, PTDB tdbp, MODE mode, char *c1, char *c2,
return true;
} // endif tdbp
if (!c1)
if (!c1) {
if (mode == MODE_INSERT)
// Allocate all column blocks for that table
tdbp->ColDB(g, NULL, 0);
else for (p= c1; *p; p+= n) {
} else for (p= c1; *p; p+= n) {
// Allocate only used column blocks
if (xtrace)
printf("Allocating column %s\n", p);
......
storage/connect/ha_connect.cc
View file @ c035bde3
......@@ -696,7 +696,9 @@ bool ha_connect::GetBooleanOption(char *opname, bool bdef)
char *pv;
PTOS options= GetTableOptionStruct(table);
if (!options)
if (!stricmp(opname, "View"))
opval= (tshp) ? tshp->is_view : table->s->is_view;
else if (!options)
;
else if (!stricmp(opname, "Mapped"))
opval= options->mapped;
......@@ -838,7 +840,7 @@ void *ha_connect::GetColumnOption(void *field, PCOLINFO pcf)
} else
fldp= (tshp) ? tshp->field : table->field;
if (!(fp= *fldp))
if (!fldp || !(fp= *fldp))
return NULL;
// Get the CONNECT field options structure
......@@ -1843,11 +1845,13 @@ int ha_connect::open(const char *name, int mode, uint test_if_locked)
// Try to get the user if possible
xp= GetUser(ha_thd(), xp);
PGLOBAL g= xp->g;
PGLOBAL g= (xp) ? xp->g : NULL;
// Try to set the database environment
if (g)
rc= (CntCheckDB(g, this, name)) ? (-2) : 0;
else
rc= HA_ERR_INTERNAL_ERROR;
DBUG_RETURN(rc);
} // end of open
......@@ -2809,6 +2813,10 @@ int ha_connect::external_lock(THD *thd, int lock_type)
bool oldsep= ((PCHK)g->Xchk)->oldsep;
bool newsep= ((PCHK)g->Xchk)->newsep;
PTDBDOS tdp= (PTDBDOS)(tdbp ? tdbp : GetTDB(g));
if (!tdp)
DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
PDOSDEF ddp= (PDOSDEF)tdp->GetDef();
PIXDEF xp, xp1, xp2, drp=NULL, adp= NULL;
PIXDEF oldpix= ((PCHK)g->Xchk)->oldpix;
......@@ -3300,7 +3308,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
{
char spc= ',', qch= 0;
const char *fncn= "?";
const char *user, *fn, *db, *host, *pwd, *prt, *sep, *tbl; // *csn;
const char *user, *fn, *db, *host, *pwd, *prt, *sep, *tbl, *src;
char *tab, *dsn;
#if defined(WIN32)
char *nsp= NULL, *cls= NULL;
......@@ -3324,12 +3332,13 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
sql.copy(STRING_WITH_LEN("CREATE TABLE whatever ("), system_charset_info);
user= host= pwd= prt= tbl= dsn= NULL;
user= host= pwd= prt= tbl= src= dsn= NULL;
// Get the useful create options
ttp= GetTypeID(topt->type);
fn= topt->filename;
tab= (char*)topt->tabname;
src= topt->srcdef;
db= topt->dbname;
fncn= topt->catfunc;
fnc= GetFuncID(fncn);
......@@ -3352,6 +3361,9 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
cls= GetListOption(g,"class", topt->oplist);
#endif // WIN32
mxr= atoi(GetListOption(g,"maxerr", topt->oplist, "0"));
} else {
host= "localhost";
user= "root";
} // endif option_list
if (!db)
......@@ -3359,18 +3371,10 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
// Check table type
if (ttp == TAB_UNDEF) {
if (!tab) {
strcpy(g->Message, "No table_type. Was set to DOS");
push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0, g->Message);
ttp= TAB_DOS;
topt->type= "DOS";
} else {
strcpy(g->Message, "No table_type. Was set to PROXY");
topt->type= (src) ? "MYSQL" : (tab) ? "PROXY" : "DOS";
ttp= GetTypeID(topt->type);
sprintf(g->Message, "No table_type. Was set to %s", topt->type);
push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0, g->Message);
ttp= TAB_PRX;
topt->type= "PROXY";
} // endif fnc
} else if (ttp == TAB_NIY) {
sprintf(g->Message, "Unsupported table type %s", topt->type);
my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
......@@ -3494,7 +3498,9 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
else
return HA_ERR_INTERNAL_ERROR; // Should never happen
switch (ttp) {
if (src && fnc == FNC_NO)
qrp= SrcColumns(g, host, db, user, pwd, src, port);
else switch (ttp) {
case TAB_DBF:
qrp= DBFColumns(g, fn, fnc == FNC_COL);
break;
......@@ -3523,7 +3529,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
#if defined(MYSQL_SUPPORT)
case TAB_MYSQL:
qrp= MyColumns(g, host, db, user, pwd, tab,
NULL, port, false, fnc == FNC_COL);
NULL, port, fnc == FNC_COL);
break;
#endif // MYSQL_SUPPORT
case TAB_CSV:
......@@ -3549,7 +3555,7 @@ static int connect_assisted_discovery(handlerton *hton, THD* thd,
return HA_ERR_INTERNAL_ERROR;
} // endif qrp
if (fnc != FNC_NO) {
if (fnc != FNC_NO || src) {
// Catalog table
for (crp=qrp->Colresp; !b && crp; crp= crp->Next) {
cnm= encode(g, crp->Name);
......@@ -3821,14 +3827,24 @@ int ha_connect::create(const char *name, TABLE *table_arg,
case TAB_PRX:
case TAB_XCL:
case TAB_OCCUR:
if (options->srcdef) {
strcpy(g->Message, "Cannot check looping reference");
push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0, g->Message);
} else if (options->tabname) {
if (!stricmp(options->tabname, create_info->alias) &&
(!options->dbname || !stricmp(options->dbname, thd->db))) {
sprintf(g->Message, "A %s table cannot refer to itself",
options->type);
my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
return HA_ERR_INTERNAL_ERROR;
DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
} // endif tab
} else {
strcpy(g->Message, "Missing object table name or definition");
my_message(ER_UNKNOWN_ERROR, g->Message, MYF(0));
DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
} // endif tabname
} // endswitch ttp
if (type == TAB_XML) {
......
storage/connect/myconn.cpp
View file @ c035bde3
......@@ -66,13 +66,13 @@ extern MYSQL_PLUGIN_IMPORT uint mysqld_port;
/************************************************************************/
/* MyColumns: constructs the result blocks containing all columns */
/* of a MySQL table that will be retrieved by GetData commands. */
/* key = TRUE when called from Create Table to get key informations. */
/* of a MySQL table or view. */
/* info = TRUE to get catalog column informations. */
/************************************************************************/
PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
const char *user, const char *pwd,
const char *table, const char *colpat,
int port, bool key, bool info)
int port, bool info)
{
static int buftyp[] = {TYPE_STRING, TYPE_SHORT, TYPE_STRING, TYPE_INT,
TYPE_STRING, TYPE_SHORT, TYPE_SHORT, TYPE_SHORT,
......@@ -124,9 +124,6 @@ PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
length[0] = 128;
} // endif info
//if (!key) // We are not called from Create table
// ncol--; // No date format column yet
/**********************************************************************/
/* Allocate the structures used to refer to the result set. */
/**********************************************************************/
......@@ -219,6 +216,7 @@ PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
crp->Kdata->SetValue(fld, i);
} // endfor i
#if 0
if (k > 1) {
// Multicolumn primary key
PVBLK vbp = qrp->Colresp->Next->Next->Next->Next->Kdata;
......@@ -228,6 +226,7 @@ PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
vbp->SetValue(k, i);
} // endif k
#endif // 0
/**********************************************************************/
/* Close MySQL connection. */
......@@ -240,6 +239,33 @@ PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
return qrp;
} // end of MyColumns
/************************************************************************/
/* SrcColumns: constructs the result blocks containing all columns */
/* resulting from an SQL source definition query execution. */
/************************************************************************/
PQRYRES SrcColumns(PGLOBAL g, const char *host, const char *db,
const char *user, const char *pwd,
const char *srcdef, int port)
{
int w;
MYSQLC myc;
PQRYRES qrp = NULL;
if (!port)
port = mysqld_port;
// Open a MySQL connection for this table
if (myc.Open(g, host, db, user, pwd, port))
return NULL;
// Send the source command to MySQL
if (myc.ExecSQL(g, srcdef, &w) == RC_OK)
qrp = myc.GetResult(g);
myc.Close();
return qrp;
} // end of SrcColumns
/* -------------------------- Class MYSQLC --------------------------- */
/***********************************************************************/
......
storage/connect/myconn.h
View file @ c035bde3
......@@ -38,7 +38,11 @@ typedef class MYSQLC *PMYC;
PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
const char *user, const char *pwd,
const char *table, const char *colpat,
int port, bool key, bool info);
int port, bool info);
PQRYRES SrcColumns(PGLOBAL g, const char *host, const char *db,
const char *user, const char *pwd,
const char *srcdef, int port);
/* -------------------------- MYCONN class --------------------------- */
......@@ -47,6 +51,7 @@ PQRYRES MyColumns(PGLOBAL g, const char *host, const char *db,
/***********************************************************************/
class DllItem MYSQLC {
friend class TDBMYSQL;
friend class MYSQLCOL;
// Construction
public:
MYSQLC(void);
......
storage/connect/mysql-test/connect/r/fmt.result
View file @ c035bde3
......@@ -18,7 +18,7 @@ CREATE TABLE t1
id INT NOT NULL field_format=' %n%d%n'
) ENGINE=CONNECT TABLE_TYPE=FMT FILE_NAME='t1.txt';
INSERT INTO t1 VALUES (10),(20);
ERROR HY000: Got error 122 'Writing FMT files is not implemented yet' from CONNECT
ERROR HY000: Got error 174 'Writing FMT files is not implemented yet' from CONNECT
DROP TABLE t1;
#
# Testing manual examples
......@@ -59,7 +59,7 @@ ID NAME DEPNO SALARY
56 POIROT-DELMOTTE 0 18009.00
345 67 19000.25
UPDATE t1 SET SALARY=1234;
ERROR HY000: Got error 122 'Writing FMT files is not implemented yet' from CONNECT
ERROR HY000: Got error 174 'Writing FMT files is not implemented yet' from CONNECT
DELETE FROM t1 WHERE ID=56;
SELECT * FROM t1;
ID NAME DEPNO SALARY
......
storage/connect/mysql-test/connect/r/pivot.result
View file @ c035bde3
......@@ -77,8 +77,8 @@ Beer DOUBLE(8,2) FLAG=1,
Car DOUBLE(8,2) FLAG=1,
Food DOUBLE(8,2) FLAG=1)
ENGINE=CONNECT TABLE_TYPE=PIVOT
SRCDEF='select who, week, what, sum(amount) from expenses where week in (4,5) group by who, week, what';
ALTER TABLE pivex OPTION_LIST='port=PORT';
SRCDEF='select who, week, what, sum(amount) as amount from expenses where week in (4,5) group by who, week, what';
ALTER TABLE pivex OPTION_LIST='PivotCol=what,FncCol=amount,port=PORT';
Warnings:
Warning 1105 The current version of CONNECT did not check what you changed in ALTER. Use at your own risk
SELECT * FROM pivex;
......@@ -125,7 +125,7 @@ Middle DOUBLE(8,2) FLAG=1,
Last DOUBLE(8,2) FLAG=1)
ENGINE=CONNECT TABLE_TYPE=PIVOT
SRCDEF='select who, what, case when week=3 then ''First'' when week=5 then ''Last'' else ''Middle'' end as wk, sum(amount) * 6.56 as amnt from expenses group by who, what, wk';
ALTER TABLE pivex OPTION_LIST='PivotCol=wk,port=PORT';
ALTER TABLE pivex OPTION_LIST='PivotCol=wk,FncCol=amnt,port=PORT';
Warnings:
Warning 1105 The current version of CONNECT did not check what you changed in ALTER. Use at your own risk
SELECT * FROM pivex;
......
storage/connect/mysql-test/connect/t/fmt.test
View file @ c035bde3
......@@ -32,7 +32,7 @@ INSERT INTO t1 VALUES (10),(20);
#DELETE FROM t1 WHERE id=10;
#SELECT * FROM t1;
DROP TABLE t1;
--remove_file $MYSQLD_DATADIR/test/t1.txt
#--remove_file $MYSQLD_DATADIR/test/t1.txt
--echo #
......
storage/connect/mysql-test/connect/t/pivot.test
View file @ c035bde3
let $MYSQLD_DATADIR= `select @@datadir`;
let $PORT= `select @@port`;
--copy_file $MTR_SUITE_DIR/std_data/expenses.txt $MYSQLD_DATADIR/test/expenses.txt
--copy_file $MTR_SUITE_DIR/std_data/connect.ini $MYSQLD_DATADIR/test/connect.ini
--echo #
--echo # Testing the PIVOT table type
......@@ -44,9 +45,9 @@ Beer DOUBLE(8,2) FLAG=1,
Car DOUBLE(8,2) FLAG=1,
Food DOUBLE(8,2) FLAG=1)
ENGINE=CONNECT TABLE_TYPE=PIVOT
SRCDEF='select who, week, what, sum(amount) from expenses where week in (4,5) group by who, week, what';
SRCDEF='select who, week, what, sum(amount) as amount from expenses where week in (4,5) group by who, week, what';
--replace_result $PORT PORT
--eval ALTER TABLE pivex OPTION_LIST='port=$PORT'
--eval ALTER TABLE pivex OPTION_LIST='PivotCol=what,FncCol=amount,port=$PORT'
SELECT * FROM pivex;
--echo #
......@@ -77,7 +78,7 @@ Last DOUBLE(8,2) FLAG=1)
ENGINE=CONNECT TABLE_TYPE=PIVOT
SRCDEF='select who, what, case when week=3 then ''First'' when week=5 then ''Last'' else ''Middle'' end as wk, sum(amount) * 6.56 as amnt from expenses group by who, what, wk';
--replace_result $PORT PORT
--eval ALTER TABLE pivex OPTION_LIST='PivotCol=wk,port=$PORT'
--eval ALTER TABLE pivex OPTION_LIST='PivotCol=wk,FncCol=amnt,port=$PORT'
SELECT * FROM pivex;
DROP TABLE pivex;
DROP TABLE expenses;
......
storage/connect/plgdbsem.h
View file @ c035bde3
......@@ -24,8 +24,8 @@
#define DOS_BUFF_LEN 100 /* Number of lines in binary file buffer */
#undef DOMAIN /* For Unix version */
enum BLKTYP {TYPE_TABLE = 50, /* Table Name/Correl Block */
TYPE_COLUMN = 51, /* Column Name/Correl Block */
enum BLKTYP {TYPE_TABLE = 50, /* Table Name/Srcdef/... Block */
TYPE_COLUMN = 51, /* Column Name/Qualifier Block */
// TYPE_OPVAL = 52, /* Operator value (OPVAL) */
TYPE_TDB = 53, /* Table Description Block */
TYPE_COLBLK = 54, /* Column Description Block */
......
storage/connect/plugutil.c
View file @ c035bde3
......@@ -147,6 +147,7 @@ PGLOBAL PlugInit(LPCSTR Language, uint worksize)
if (!(g = malloc(sizeof(GLOBAL)))) {
fprintf(stderr, MSG(GLOBAL_ERROR), (int)sizeof(GLOBAL));
return NULL;
} else {
g->Sarea_Size = worksize;
g->Trace = 0;
......@@ -180,7 +181,9 @@ int PlugExit(PGLOBAL g)
if (!g)
return rc;
if (g->Sarea)
free(g->Sarea);
free(g);
return rc;
} /* end of PlugExit */
......
storage/connect/tabcol.cpp
View file @ c035bde3
/************* TabCol C++ Functions Source Code File (.CPP) ************/
/* Name: TABCOL.CPP Version 2.6 */
/* Name: TABCOL.CPP Version 2.7 */
/* */
/* (C) Copyright to the author Olivier BERTRAND 1998-2012 */
/* (C) Copyright to the author Olivier BERTRAND 1998-2013 */
/* */
/* This file contains the PlugDB++ XTAB, COLUMN and XORDER methods. */
/***********************************************************************/
......@@ -23,18 +23,18 @@
#include "tabcol.h"
/***********************************************************************/
/* XTAB public constructor (in which Correl defaults to Name). */
/* XTAB public constructor. */
/***********************************************************************/
XTAB::XTAB(LPCSTR name, LPCSTR correl) : Name(name)
XTAB::XTAB(LPCSTR name, LPCSTR srcdef) : Name(name)
{
Next = NULL;
To_Tdb = NULL;
Correl = (correl) ? correl : name;
Srcdef = srcdef;
Creator = NULL;
Qualifier = NULL;
#ifdef DEBTRACE
htrc(" making new TABLE %s %s\n", Name, Correl);
htrc(" making new TABLE %s %s\n", Name, Srcdef);
#endif
} // end of XTAB constructor
......@@ -45,12 +45,12 @@ XTAB::XTAB(PTABLE tp) : Name(tp->Name)
{
Next = NULL;
To_Tdb = NULL;
Correl = tp->Correl;
Srcdef = tp->Srcdef;
Creator = tp->Creator;
Qualifier = tp->Qualifier;
#ifdef DEBTRACE
htrc(" making copy TABLE %s %s\n", Name, Correl);
htrc(" making copy TABLE %s %s\n", Name, Srcdef);
#endif
} // end of XTAB constructor
......@@ -83,7 +83,7 @@ void XTAB::Print(PGLOBAL g, FILE *f, uint n)
for (PTABLE tp = this; tp; tp = tp->Next) {
fprintf(f, "%sTABLE: %s.%s %s\n",
m, SVP(tp->Creator), tp->Name, SVP(tp->Correl));
m, SVP(tp->Creator), tp->Name, SVP(tp->Srcdef));
PlugPutOut(g, f, TYPE_TDB, tp->To_Tdb, n + 2);
} /* endfor tp */
......@@ -101,7 +101,7 @@ void XTAB::Print(PGLOBAL g, char *ps, uint z)
for (PTABLE tp = this; tp && n > 0; tp = tp->Next) {
i = sprintf(buf, "TABLE: %s.%s %s To_Tdb=%p ",
SVP(tp->Creator), tp->Name, SVP(tp->Correl), tp->To_Tdb);
SVP(tp->Creator), tp->Name, SVP(tp->Srcdef), tp->To_Tdb);
strncat(ps, buf, n);
n -= i;
} // endif tp
......
storage/connect/tabcol.h
View file @ c035bde3
/*************** TabCol H Declares Source Code File (.H) ***************/
/* Name: TABCOL.H Version 2.7 */
/* Name: TABCOL.H Version 2.8 */
/* */
/* (C) Copyright to the author Olivier BERTRAND 1998-2012 */
/* (C) Copyright to the author Olivier BERTRAND 1998-2013 */
/* */
/* This file contains the XTAB, COLUMN and XORDER class definitions. */
/***********************************************************************/
......@@ -15,23 +15,23 @@
/***********************************************************************/
/* Definition of class XTAB with all its method functions. */
/***********************************************************************/
class DllExport XTAB: public BLOCK { // Table Name-Owner-Correl block.
class DllExport XTAB: public BLOCK { // Table Name-Owner-Srcdef block.
friend class TDBPRX;
public:
// Constructors
XTAB(LPCSTR name, LPCSTR correl = NULL);
XTAB(LPCSTR name, LPCSTR srcdef = NULL);
XTAB(PTABLE tp);
// Implementation
PTABLE GetNext(void) {return Next;}
PTDB GetTo_Tdb(void) {return To_Tdb;}
LPCSTR GetName(void) {return Name;}
LPCSTR GetCorrel(void) {return Correl;}
LPCSTR GetSrc(void) {return Srcdef;}
LPCSTR GetCreator(void) {return Creator;}
LPCSTR GetQualifier(void) {return Qualifier;}
void SetTo_Tdb(PTDB tdbp) {To_Tdb = tdbp;}
void SetName(LPCSTR name) {Name = name;}
void SetCorrel(LPCSTR correl) {Correl = correl;}
void SetSrc(LPCSTR srcdef) {Srcdef = srcdef;}
void SetCreator(LPCSTR crname) {Creator = crname;}
void SetQualifier(LPCSTR qname) {Qualifier = qname;}
......@@ -44,8 +44,8 @@ class DllExport XTAB: public BLOCK { // Table Name-Owner-Correl block.
// Members
PTABLE Next; // Points to next table in chain
PTDB To_Tdb; // Points to Table description Block
LPCSTR Name; // Table name (can be changed by LNA and PLG)
LPCSTR Correl; // Correlation name
LPCSTR Name; // Table name
LPCSTR Srcdef; // Table Source definition
LPCSTR Creator; // Creator name
LPCSTR Qualifier; // Qualifier name
}; // end of class XTAB
......
storage/connect/tabfmt.cpp
View file @ c035bde3
......@@ -1088,7 +1088,12 @@ bool TDBFMT::OpenDB(PGLOBAL g)
{
Linenum = 0;
if (Use != USE_OPEN && (Columns || Mode == MODE_UPDATE)) {
if (Mode == MODE_INSERT || Mode == MODE_UPDATE) {
sprintf(g->Message, MSG(FMT_WRITE_NIY), "FMT");
return true; // NIY
} // endif Mode
if (Use != USE_OPEN && Columns) {
// Make the formats used to read records
PSZ pfm;
int i, n;
......@@ -1096,7 +1101,6 @@ bool TDBFMT::OpenDB(PGLOBAL g)
PCOLDEF cdp;
PDOSDEF tdp = (PDOSDEF)To_Def;
// if (Mode != MODE_UPDATE) {
for (colp = (PCSVCOL)Columns; colp; colp = (PCSVCOL)colp->Next)
if (!colp->IsSpecial()) // Not a pseudo column
Fields = max(Fields, (int)colp->Fldnum);
......@@ -1104,10 +1108,6 @@ bool TDBFMT::OpenDB(PGLOBAL g)
if (Columns)
Fields++; // Fldnum was 0 based
// } else
// for (cdp = tdp->GetCols(); cdp; cdp = cdp->GetNext())
// Fields++;
To_Fld = PlugSubAlloc(g, NULL, Lrecl + 1);
FldFormat = (PSZ*)PlugSubAlloc(g, NULL, sizeof(PSZ) * Fields);
memset(FldFormat, 0, sizeof(PSZ) * Fields);
......
storage/connect/tabmysql.cpp
View file @ c035bde3
......@@ -79,9 +79,11 @@ MYSQLDEF::MYSQLDEF(void)
Hostname = NULL;
Database = NULL;
Tabname = NULL;
Srcdef = NULL;
Username = NULL;
Password = NULL;
Portnumber = 0;
Isview = FALSE;
Bind = FALSE;
Delayed = FALSE;
} // end of MYSQLDEF constructor
......@@ -302,8 +304,8 @@ bool MYSQLDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
} else {
// MYSQL access from a PROXY table, not using URL
Database = Cat->GetStringCatInfo(g, "Database", "*");
Tabname = Cat->GetStringCatInfo(g, "Name", Name); // Deprecated
Tabname = Cat->GetStringCatInfo(g, "Tabname", Tabname);
Tabname = Name;
Isview = Cat->GetBoolCatInfo("View", FALSE);
// We must get connection parms from the calling table
Remove_tshp(Cat);
......@@ -313,6 +315,9 @@ bool MYSQLDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
Portnumber = Cat->GetIntCatInfo("Port", mysqld_port);
} // endif am
if ((Srcdef = Cat->GetStringCatInfo(g, "Srcdef", NULL)))
Isview = TRUE;
return FALSE;
} // end of DefineAM
......@@ -339,18 +344,22 @@ TDBMYSQL::TDBMYSQL(PMYDEF tdp) : TDBASE(tdp)
Host = tdp->GetHostname();
Database = tdp->GetDatabase();
Tabname = tdp->GetTabname();
Srcdef = tdp->GetSrcdef();
User = tdp->GetUsername();
Pwd = tdp->GetPassword();
Port = tdp->GetPortnumber();
Isview = tdp->Isview;
Prep = tdp->Bind;
Delayed = tdp->Delayed;
} else {
Host = NULL;
Database = NULL;
Tabname = NULL;
Srcdef = NULL;
User = NULL;
Pwd = NULL;
Port = 0;
Isview = FALSE;
Prep = FALSE;
Delayed = FALSE;
} // endif tdp
......@@ -370,9 +379,11 @@ TDBMYSQL::TDBMYSQL(PGLOBAL g, PTDBMY tdbp) : TDBASE(tdbp)
Host = tdbp->Host;
Database = tdbp->Database;
Tabname = tdbp->Tabname;
Srcdef = tdbp->Srcdef;
User = tdbp->User;
Pwd = tdbp->Pwd;
Port = tdbp->Port;
Isview = tdbp->Isview;
Prep = tdbp->Prep;
Delayed = tdbp->Delayed;
Bind = NULL;
......@@ -418,55 +429,54 @@ PCOL TDBMYSQL::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
/***********************************************************************/
bool TDBMYSQL::MakeSelect(PGLOBAL g)
{
char *colist;
char *tk = "`";
int len = 0, ncol = 0, rank = 0;
int rank = 0;
bool b = FALSE;
PCOL colp;
PDBUSER dup = PlgGetUser(g);
//PDBUSER dup = PlgGetUser(g);
if (Query)
return FALSE; // already done
for (colp = Columns; colp; colp = colp->GetNext())
ncol++;
if (Srcdef) {
Query = Srcdef;
return false;
} // endif Srcdef
if (ncol) {
colist = (char*)PlugSubAlloc(g, NULL, (NAM_LEN + 4) * ncol);
*colist = '\0';
//Find the address of the suballocated query
Query = (char*)PlugSubAlloc(g, NULL, 0);
strcpy(Query, "SELECT ");
if (Columns) {
for (colp = Columns; colp; colp = colp->GetNext())
if (colp->IsSpecial()) {
strcpy(g->Message, MSG(NO_SPEC_COL));
return TRUE;
} else {
if (b)
strcat(colist, ", ");
strcat(Query, ", ");
else
b = TRUE;
strcat(strcat(strcat(colist, tk), colp->GetName()), tk);
strcat(strcat(strcat(Query, tk), colp->GetName()), tk);
((PMYCOL)colp)->Rank = rank++;
} // endif colp
} else {
// ncol == 0 can occur for queries such as Query count(*) from...
// for which we will count the rows from Query '*' from...
// ncol == 0 can occur for views or queries such as
// Query count(*) from... for which we will count the rows from
// Query '*' from...
// (the use of a char constant minimize the result storage)
colist = (char*)PlugSubAlloc(g, NULL, 2);
strcpy(colist, "'*'");
strcat(Query, (Isview) ? "*" : "'*'");
} // endif ncol
// Below 32 is space to contain extra stuff
len += (strlen(colist) + strlen(Tabname) + 32);
len += (To_Filter ? strlen(To_Filter) + 7 : 0);
Query = (char*)PlugSubAlloc(g, NULL, len);
strcat(strcpy(Query, "SELECT "), colist);
strcat(strcat(strcat(strcat(Query, " FROM "), tk), Tabname), tk);
if (To_Filter)
strcat(strcat(Query, " WHERE "), To_Filter);
// Now we know how much to suballocate
PlugSubAlloc(g, NULL, strlen(Query) + 1);
return FALSE;
} // end of MakeSelect
......@@ -751,7 +761,7 @@ bool TDBMYSQL::OpenDB(PGLOBAL g)
/* Table already open, just replace it at its beginning. */
/*******************************************************************/
Myc.Rewind();
return FALSE;
return false;
} // endif use
/*********************************************************************/
......@@ -763,7 +773,7 @@ bool TDBMYSQL::OpenDB(PGLOBAL g)
/*********************************************************************/
if (!Myc.Connected()) {
if (Myc.Open(g, Host, Database, User, Pwd, Port))
return TRUE;
return true;
} // endif Connected
......@@ -774,7 +784,24 @@ bool TDBMYSQL::OpenDB(PGLOBAL g)
if (!MakeSelect(g))
m_Rc = Myc.ExecSQL(g, Query);
#if 0
if (!Myc.m_Res || !Myc.m_Fields) {
sprintf(g->Message, "%s result", (Myc.m_Res) ? "Void" : "No");
Myc.Close();
return true;
} // endif m_Res
#endif // 0
if (Srcdef)
if (SetColumnRanks(g))
return true;
} else if (Mode == MODE_INSERT) {
if (Srcdef) {
strcpy(g->Message, "No insert into anonym views");
return true;
} // endif Srcdef
if (!MakeInsert(g)) {
#if defined(MYSQL_PREPARED_STATEMENTS)
int n = (Prep) ? Myc.PrepareSQL(g, Query) : Nparm;
......@@ -826,6 +853,55 @@ bool TDBMYSQL::OpenDB(PGLOBAL g)
return FALSE;
} // end of OpenDB
/***********************************************************************/
/* Set the rank of columns in the result set. */
/***********************************************************************/
bool TDBMYSQL::SetColumnRanks(PGLOBAL g)
{
for (PCOL colp = Columns; colp; colp = colp->GetNext())
if (((PMYCOL)colp)->FindRank(g))
return TRUE;
return FALSE;
} // end of SetColumnRanks
/***********************************************************************/
/* Called by Parent table to make the columns of a View. */
/***********************************************************************/
PCOL TDBMYSQL::MakeFieldColumn(PGLOBAL g, char *name)
{
int n;
MYSQL_FIELD *fld;
PCOL cp, colp = NULL;
for (n = 0; n < Myc.m_Fields; n++) {
fld = &Myc.m_Res->fields[n];
if (!stricmp(name, fld->name)) {
colp = new(g) MYSQLCOL(fld, this, n);
if (colp->InitValue(g))
return NULL;
if (!Columns)
Columns = colp;
else for (cp = Columns; cp; cp = cp->GetNext())
if (!cp->GetNext()) {
cp->SetNext(colp);
break;
} // endif Next
break;
} // endif name
} // endfor n
if (!colp)
sprintf(g->Message, "Column %s is not in view", name);
return colp;
} // end of MakeFieldColumn
/***********************************************************************/
/* Data Base read routine for MYSQL access method. */
/***********************************************************************/
......@@ -939,7 +1015,7 @@ MYSQLCOL::MYSQLCOL(PCOLDEF cdp, PTDB tdbp, PCOL cprec, int i, PSZ am)
tdbp->SetColumns(this);
} // endif cprec
// Set additional Dos access method information for column.
// Set additional MySQL access method information for column.
Long = cdp->GetLong();
Bind = NULL;
To_Val = NULL;
......@@ -951,6 +1027,33 @@ MYSQLCOL::MYSQLCOL(PCOLDEF cdp, PTDB tdbp, PCOL cprec, int i, PSZ am)
} // end of MYSQLCOL constructor
/***********************************************************************/
/* MYSQLCOL public constructor. */
/***********************************************************************/
MYSQLCOL::MYSQLCOL(MYSQL_FIELD *fld, PTDB tdbp, int i, PSZ am)
: COLBLK(NULL, tdbp, i)
{
Name = fld->name;
Opt = 0;
Long = fld->length;
Buf_Type = MYSQLtoPLG(fld->type);
strcpy(Format.Type, GetFormatType(Buf_Type));
Format.Length = Long;
Format.Prec = fld->decimals;
ColUse = U_P;
Nullable = !IS_NOT_NULL(fld->flags);
// Set additional MySQL access method information for column.
Bind = NULL;
To_Val = NULL;
Slen = 0;
Rank = i;
if (trace)
htrc(" making new %sCOL C%d %s at %p\n", am, Index, Name, this);
} // end of MYSQLCOL constructor
/***********************************************************************/
/* MYSQLCOL constructor used for copying columns. */
/* tdbp is the pointer to the new table descriptor. */
......@@ -964,6 +1067,24 @@ MYSQLCOL::MYSQLCOL(MYSQLCOL *col1, PTDB tdbp) : COLBLK(col1, tdbp)
Rank = col1->Rank;
} // end of MYSQLCOL copy constructor
/***********************************************************************/
/* FindRank: Find the rank of this column in the result set. */
/***********************************************************************/
bool MYSQLCOL::FindRank(PGLOBAL g)
{
int n;
MYSQLC myc = ((PTDBMY)To_Tdb)->Myc;
for (n = 0; n < myc.m_Fields; n++)
if (!stricmp(Name, myc.m_Res->fields[n].name)) {
Rank = n;
return false;
} // endif Name
sprintf(g->Message, "Column %s not in result set", Name);
return true;
} // end of FindRank
/***********************************************************************/
/* SetBuffer: prepare a column block for write operation. */
/***********************************************************************/
......@@ -1049,11 +1170,6 @@ void MYSQLCOL::ReadColumn(PGLOBAL g)
int rc;
PTDBMY tdbp = (PTDBMY)To_Tdb;
if (trace)
htrc("MySQL ReadColumn: name=%s\n", Name);
assert (Rank >= 0);
/*********************************************************************/
/* If physical fetching of the line was deferred, do it now. */
/*********************************************************************/
......@@ -1066,14 +1182,17 @@ void MYSQLCOL::ReadColumn(PGLOBAL g)
} else
tdbp->Fetched = TRUE;
if ((buf = ((PTDBMY)To_Tdb)->Myc.GetCharField(Rank)))
if ((buf = ((PTDBMY)To_Tdb)->Myc.GetCharField(Rank))) {
if (trace)
htrc("MySQL ReadColumn: name=%s buf=%s\n", Name, buf);
Value->SetValue_char(buf, Long);
else {
} else {
if (Nullable)
Value->SetNull(true);
Value->Reset(); // Null value
} // endelse
} // endif buf
} // end of ReadColumn
......@@ -1121,5 +1240,5 @@ TDBMCL::TDBMCL(PMYDEF tdp) : TDBCAT(tdp)
/***********************************************************************/
PQRYRES TDBMCL::GetResult(PGLOBAL g)
{
return MyColumns(g, Host, Db, User, Pwd, Tab, NULL, Port, false, false);
return MyColumns(g, Host, Db, User, Pwd, Tab, NULL, Port, false);
} // end of GetResult
storage/connect/tabmysql.h
View file @ c035bde3
......@@ -30,6 +30,7 @@ class MYSQLDEF : public TABDEF {/* Logical table description */
inline PSZ GetHostname(void) {return Hostname;};
inline PSZ GetDatabase(void) {return Database;};
inline PSZ GetTabname(void) {return Tabname;}
inline PSZ GetSrcdef(void) {return Srcdef;}
inline PSZ GetUsername(void) {return Username;};
inline PSZ GetPassword(void) {return Password;};
inline int GetPortnumber(void) {return Portnumber;}
......@@ -44,9 +45,11 @@ class MYSQLDEF : public TABDEF {/* Logical table description */
PSZ Hostname; /* Host machine to use */
PSZ Database; /* Database to be used by server */
PSZ Tabname; /* External table name */
PSZ Srcdef; /* The source table SQL definition */
PSZ Username; /* User logon name */
PSZ Password; /* Password logon info */
int Portnumber; /* MySQL port number (0 = default) */
bool Isview; /* TRUE if this table is a MySQL view */
bool Bind; /* Use prepared statement on insert */
bool Delayed; /* Delayed insert */
}; // end of MYSQLDEF
......@@ -72,6 +75,7 @@ class TDBMYSQL : public TDBASE {
virtual int GetProgMax(PGLOBAL g);
virtual void ResetDB(void) {N = 0;}
virtual int RowNumber(PGLOBAL g, bool b = FALSE);
virtual bool IsView(void) {return Isview;}
void SetDatabase(LPCSTR db) {Database = (char*)db;}
// Database routines
......@@ -83,6 +87,10 @@ class TDBMYSQL : public TDBASE {
virtual int DeleteDB(PGLOBAL g, int irc);
virtual void CloseDB(PGLOBAL g);
// Specific routines
bool SetColumnRanks(PGLOBAL g);
PCOL MakeFieldColumn(PGLOBAL g, char *name);
protected:
// Internal functions
bool MakeSelect(PGLOBAL g);
......@@ -99,9 +107,11 @@ class TDBMYSQL : public TDBASE {
char *Pwd; // Password logon info
char *Database; // Database to be used by server
char *Tabname; // External table name
char *Srcdef; // The source table SQL definition
char *Query; // Points to SQL query
char *Qbuf; // Used for not prepared insert
bool Fetched; // True when fetch was done
bool Isview; // True if this table is a MySQL view
bool Prep; // Use prepared statement on insert
bool Delayed; // Use delayed insert
int m_Rc; // Return code from command
......@@ -119,6 +129,7 @@ class MYSQLCOL : public COLBLK {
public:
// Constructors
MYSQLCOL(PCOLDEF cdp, PTDB tdbp, PCOL cprec, int i, PSZ am = "MYSQL");
MYSQLCOL(MYSQL_FIELD *fld, PTDB tdbp, int i, PSZ am = "MYSQL");
MYSQLCOL(MYSQLCOL *colp, PTDB tdbp); // Constructor used in copy process
// Implementation
......@@ -129,6 +140,7 @@ class MYSQLCOL : public COLBLK {
virtual bool SetBuffer(PGLOBAL g, PVAL value, bool ok, bool check);
virtual void ReadColumn(PGLOBAL g);
virtual void WriteColumn(PGLOBAL g);
bool FindRank(PGLOBAL g);
protected:
// Default constructor not to be used
......
storage/connect/taboccur.cpp
View file @ c035bde3
......@@ -85,19 +85,43 @@ PTDB OCCURDEF::GetTable(PGLOBAL g, MODE m)
/***********************************************************************/
TDBOCCUR::TDBOCCUR(POCCURDEF tdp) : TDBPRX(tdp)
{
//Tdbp = NULL; // Source table
//Tdbp = NULL; // Source table (in TDBPRX)
Tabname = tdp->Tablep->GetName(); // Name of source table
Colist = tdp->Colist; // List of source columns
Xcolumn = tdp->Xcol; // Occur column name
Rcolumn = tdp->Rcol; // Rank column name
Xcolp = NULL; // To the OCCURCOL column
Col = NULL; // To source column blocks array
Mult = -1; // Multiplication factor
Mult = PrepareColist(); // Multiplication factor
N = 0; // The current table index
M = 0; // The occurence rank
RowFlag = 0; // 0: Ok, 1: Same, 2: Skip
} // 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. */
/***********************************************************************/
......@@ -111,14 +135,8 @@ PCOL TDBOCCUR::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
} else if (!stricmp(cdp->GetName(), Xcolumn)) {
// Allocate the OCCUR column
colp = Xcolp = new(g) OCCURCOL(cdp, this, n);
} else {
colp = new(g) PRXCOL(cdp, this, cprec, n);
if (((PPRXCOL)colp)->Init(g))
return NULL;
return colp;
} //endif name
} else
return new(g) PRXCOL(cdp, this, cprec, n);
if (cprec) {
colp->SetNext(cprec->GetNext());
......@@ -136,66 +154,91 @@ PCOL TDBOCCUR::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
/***********************************************************************/
bool TDBOCCUR::InitTable(PGLOBAL g)
{
if (!Tdbp) {
if (!Tdbp)
// Get the table description block of this table
if (!(Tdbp = (PTDBASE)GetSubTable(g, ((POCCURDEF)To_Def)->Tablep)))
if (!(Tdbp = GetSubTable(g, ((POCCURDEF)To_Def)->Tablep, TRUE)))
return TRUE;
if (MakeColumnList(g) < 0)
if (!Tdbp->IsView())
if (MakeColumnList(g))
return TRUE;
} // endif Tdbp
return FALSE;
} // end of InitTable
/***********************************************************************/
/* Allocate OCCUR column description block. */
/***********************************************************************/
int TDBOCCUR::MakeColumnList(PGLOBAL g)
bool TDBOCCUR::MakeColumnList(PGLOBAL g)
{
if (Mult < 0) {
char *p, *pn;
char *pn;
int i;
int n = 0;
PCOL colp;
// Count the number of columns and change separator into null char
for (pn = Colist; ; pn += (strlen(pn) + 1))
if ((p = strchr(pn, ',')) || (p = strchr(pn, ';'))) {
*p++ = '\0';
n++;
} else {
if (*pn)
n++;
break;
} // endif p
for (colp = Columns; colp; colp = colp->GetNext())
if (colp->GetAmType() == TYPE_AM_PRX)
if (((PPRXCOL)colp)->Init(g))
return true;
Col = (PCOL*)PlugSubAlloc(g, NULL, n * sizeof(PCOL));
Col = (PCOL*)PlugSubAlloc(g, NULL, Mult * sizeof(PCOL));
for (i = 0, pn = Colist; i < n; i++, pn += (strlen(pn) + 1)) {
for (i = 0, pn = Colist; i < Mult; i++, pn += (strlen(pn) + 1)) {
if (!(Col[i] = Tdbp->ColDB(g, pn, 0))) {
// Column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), pn, Tabname);
return -1;
return true;
} // endif Col
if (Col[i]->InitValue(g)) {
strcpy(g->Message, "OCCUR InitValue failed");
return -1;
return true;
} // endif InitValue
} // endfor i
// OCCUR column name defaults to the name of the list first column
if (!Xcolumn)
Xcolumn = Colist;
return false;
} // end of MakeColumnList
Mult = n;
} // endif Mult
/***********************************************************************/
/* Allocate OCCUR column description block for a view. */
/***********************************************************************/
bool TDBOCCUR::ViewColumnList(PGLOBAL g)
{
char *pn;
int i;
PCOL colp, cp;
PTDBMY tdbp;
return Mult;
} // end of MakeColumnList
if (!Tdbp->IsView())
return false;
if (Tdbp->GetAmType() != TYPE_AM_MYSQL) {
strcpy(g->Message, "View is not MySQL");
return true;
} else
tdbp = (PTDBMY)Tdbp;
for (cp = Columns; cp; cp = cp->GetNext())
if (cp->GetAmType() == TYPE_AM_PRX) {
if ((colp = tdbp->MakeFieldColumn(g, cp->GetName()))) {
((PPRXCOL)cp)->Colp = colp;
((PPRXCOL)cp)->To_Val = colp->GetValue();
} else
return true;
} // endif Type
Col = (PCOL*)PlugSubAlloc(g, NULL, Mult * sizeof(PCOL));
for (i = 0, pn = Colist; i < Mult; i++, pn += (strlen(pn) + 1))
if (!(Col[i] = tdbp->MakeFieldColumn(g, pn))) {
// Column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), pn, Tabname);
return true;
} // endif Col
return false;
} // end of ViewColumnList
/***********************************************************************/
/* OCCUR GetMaxSize: returns the maximum number of rows in the table. */
......@@ -203,8 +246,8 @@ int TDBOCCUR::MakeColumnList(PGLOBAL g)
int TDBOCCUR::GetMaxSize(PGLOBAL g)
{
if (MaxSize < 0) {
if (InitTable(g))
return NULL;
if (!(Tdbp = GetSubTable(g, ((POCCURDEF)To_Def)->Tablep, TRUE)))
return 0;
MaxSize = Mult * Tdbp->GetMaxSize(g);
} // endif MaxSize
......@@ -252,7 +295,7 @@ bool TDBOCCUR::OpenDB(PGLOBAL g)
/* Do it here if not done yet. */
/*********************************************************************/
if (InitTable(g))
return NULL;
return TRUE;
if (Xcolp)
// Lock this column so it is evaluated by its table only
......@@ -269,7 +312,10 @@ bool TDBOCCUR::OpenDB(PGLOBAL g)
/*********************************************************************/
/* Do open the source table. */
/*********************************************************************/
return Tdbp->OpenDB(g);
if (Tdbp->OpenDB(g))
return TRUE;
return ViewColumnList(g);
} // end of OpenDB
/***********************************************************************/
......
storage/connect/taboccur.h
View file @ c035bde3
......@@ -58,8 +58,9 @@ class TDBOCCUR : public TDBPRX {
// Methods
virtual void ResetDB(void) {N = 0; Tdbp->ResetDB();}
virtual int RowNumber(PGLOBAL g, bool b = FALSE);
PTDB GetSourceTable(PGLOBAL g);
int MakeColumnList(PGLOBAL g);
int PrepareColist(void);
bool MakeColumnList(PGLOBAL g);
bool ViewColumnList(PGLOBAL g);
// Database routines
virtual PCOL MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n);
......
storage/connect/tabpivot.cpp
View file @ c035bde3
/************ TabPivot C++ Program Source Code File (.CPP) *************/
/* PROGRAM NAME: TABPIVOT */
/* ------------- */
/* Version 1.4 */
/* Version 1.5 */
/* */
/* COPYRIGHT: */
/* ---------- */
......@@ -41,11 +41,9 @@
#include "global.h"
#include "plgdbsem.h"
#include "xtable.h"
//#include "xindex.h"
#include "tabcol.h"
#include "colblk.h"
//#include "tabmysql.h"
#include "myconn.h"
#include "tabmysql.h"
#include "csort.h"
#include "tabutil.h"
#include "tabpivot.h"
......@@ -55,381 +53,6 @@
extern "C" int trace;
#if 0
/***********************************************************************/
/* Prepare the source table Query. */
/***********************************************************************/
PQRYRES TDBPIVOT::GetSourceTable(PGLOBAL g)
{
if (Qryp)
return Qryp; // Already done
if (Tabname) {
char *def, *colist;
size_t len = 0;
PCOL colp;
PDBUSER dup = (PDBUSER)g->Activityp->Aptr;
if (InitTable(g))
return NULL;
// Evaluate the length of the column list
for (colp = Tdbp->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 = Tdbp->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 = Tdbp->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
/***********************************************************************/
/* 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
/* --------------- Implementation of the PIVOT classes --------------- */
/***********************************************************************/
......@@ -451,13 +74,12 @@ bool PIVOTDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
char *p1, *p2;
PHC hc = ((MYCAT*)Cat)->GetHandler();
if (!PRXDEF::DefineAM(g, am, poff)) {
if (PRXDEF::DefineAM(g, am, poff))
return TRUE;
Tabname = (char*)Tablep->GetName();
DB = (char*)Tablep->GetQualifier();
} else {
DB = Cat->GetStringCatInfo(g, "Database", "*");
Tabsrc = Cat->GetStringCatInfo(g, "SrcDef", NULL);
} // endif
Tabsrc = (char*)Tablep->GetSrc();
Host = Cat->GetStringCatInfo(g, "Host", "localhost");
User = Cat->GetStringCatInfo(g, "User", "*");
......@@ -477,7 +99,7 @@ bool PIVOTDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
GBdone = Cat->GetBoolCatInfo("Groupby", false);
Accept = Cat->GetBoolCatInfo("Accept", false);
Port = Cat->GetIntCatInfo("Port", 3306);
Desc = (Tabname) ? Tabname : Tabsrc;
Desc = (Tabsrc) ? Tabsrc : Tabname;
return FALSE;
} // end of DefineAM
......@@ -520,6 +142,25 @@ TDBPIVOT::TDBPIVOT(PPIVOTDEF tdp) : TDBPRX(tdp)
RowFlag = 0; // 0: Ok, 1: Same, 2: Skip
} // end of TDBPIVOT constructor
/***********************************************************************/
/* Allocate source column description block. */
/***********************************************************************/
PCOL TDBPIVOT::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
{
PCOL colp;
if (cdp->GetOffset()) {
colp = new(g) FNCCOL(cdp, this, cprec, n);
if (cdp->GetOffset() > 1)
Dcolp = colp;
} else
colp = new(g) SRCCOL(cdp, this, cprec, n);
return colp;
} // end of MakeCol
/***********************************************************************/
/* Prepare the source table Query. */
/***********************************************************************/
......@@ -528,14 +169,14 @@ bool TDBPIVOT::GetSourceTable(PGLOBAL g)
if (Tdbp)
return false; // Already done
if (Tabname) {
PTABDEF defp;
PCOLDEF cdp;
if (InitTable(g))
if (!Tabsrc && Tabname) {
// Get the table description block of this table
if (!(Tdbp = GetSubTable(g, ((PPIVOTDEF)To_Def)->Tablep, true)))
return true;
else
defp = Tdbp->GetDef();
if (!Tdbp->IsView()) {
PCOLDEF cdp;
PTABDEF defp = Tdbp->GetDef();
if (!Fncol) {
for (cdp = defp->GetCols(); cdp; cdp = cdp->GetNext())
......@@ -595,32 +236,84 @@ bool TDBPIVOT::GetSourceTable(PGLOBAL g)
PlugSubAlloc(g, NULL, strlen(Tabsrc));
} // endif !GBdone
} // endif IsView
} else if (!Tabsrc) {
strcpy(g->Message, MSG(SRC_TABLE_UNDEF));
return true;
} // endif
if (Tabsrc) {
MYSQLC myc; // MySQL connection class
PQRYRES qryp;
PCOLRES crp;
int w;
// Get the new table description block of this source table
PTABLE tablep = new(g) XTAB("whatever", Tabsrc);
tablep->SetQualifier(Database);
// Open a MySQL connection for this table
if (myc.Open(g, Host, Database, User, Pwd, Port))
if (!(Tdbp = GetSubTable(g, tablep, true)))
return true;
// Send the source command to MySQL
if (myc.ExecSQL(g, Tabsrc, &w) == RC_FX) {
myc.Close();
} // endif Tabsrc
return false;
} // end of GetSourceTable
/***********************************************************************/
/* Make the required pivot columns. */
/***********************************************************************/
bool TDBPIVOT::MakePivotColumns(PGLOBAL g)
{
if (!Tdbp->IsView()) {
// Now it is time to allocate the pivot and function columns
if (!(Fcolp = Tdbp->ColDB(g, Fncol, 0))) {
// Function column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), Fncol, Tabname);
return true;
} // endif Exec
} else if (Fcolp->InitValue(g))
return true;
if (!(Xcolp = Tdbp->ColDB(g, Picol, 0))) {
// Pivot column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), Picol, Tabname);
return true;
} else if (Xcolp->InitValue(g))
return true;
// Check and initialize the subtable columns
for (PCOL cp = Columns; cp; cp = cp->GetNext())
if (cp->GetAmType() == TYPE_AM_SRC) {
if (((PSRCCOL)cp)->Init(g))
return TRUE;
} else if (cp->GetAmType() == TYPE_AM_FNC)
if (((PFNCCOL)cp)->InitColumn(g))
return TRUE;
} // endif isview
return false;
} // end of MakePivotColumns
/***********************************************************************/
/* Make the required pivot columns for an object view. */
/***********************************************************************/
bool TDBPIVOT::MakeViewColumns(PGLOBAL g)
{
if (Tdbp->IsView()) {
// Tdbp is a view ColDB cannot be used
PCOL colp, cp;
PTDBMY tdbp;
// We must have a storage query to get pivot column values
qryp = myc.GetResult(g);
myc.Close();
Tdbp = new(g) TDBQRS(qryp);
if (Tdbp->GetAmType() != TYPE_AM_MYSQL) {
strcpy(g->Message, "View is not MySQL");
return true;
} else
tdbp = (PTDBMY)Tdbp;
if (!Fncol || !Picol) {
strcpy(g->Message, "Missing Function or Pivot column");
return true;
} // endif
#if 0
if (!Fncol) {
for (crp = qryp->Colresp; crp; crp = crp->Next)
if (!Picol || stricmp(Picol, crp->Name))
......@@ -645,45 +338,33 @@ bool TDBPIVOT::GetSourceTable(PGLOBAL g)
} // endif Picol
} // endif Picol
} // endif Tabsrc
#endif // 0
// Now it is time to allocate the pivot and function columns
if (!(Fcolp = Tdbp->ColDB(g, Fncol, 0))) {
// Function column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), Fncol, Tabname);
return true;
} else if (Fcolp->InitValue(g))
if (!(Fcolp = tdbp->MakeFieldColumn(g, Fncol)))
return true;
if (!(Xcolp = Tdbp->ColDB(g, Picol, 0))) {
// Pivot column not found in table
sprintf(g->Message, MSG(COL_ISNOT_TABLE), Picol, Tabname);
return true;
} else if (Xcolp->InitValue(g))
if (!(Xcolp = tdbp->MakeFieldColumn(g, Picol)))
return true;
return false;
} // end of GetSourceTable
/***********************************************************************/
/* Allocate source column description block. */
/***********************************************************************/
PCOL TDBPIVOT::MakeCol(PGLOBAL g, PCOLDEF cdp, PCOL cprec, int n)
{
PCOL colp;
if (cdp->GetOffset()) {
colp = new(g) FNCCOL(cdp, this, cprec, n);
// Check and initialize the subtable columns
for (cp = Columns; cp; cp = cp->GetNext())
if (cp->GetAmType() == TYPE_AM_SRC) {
if ((colp = tdbp->MakeFieldColumn(g, cp->GetName()))) {
((PSRCCOL)cp)->Colp = colp;
((PSRCCOL)cp)->To_Val = colp->GetValue();
cp->AddStatus(BUF_READ); // All is done here
} else
return true;
if (cdp->GetOffset() > 1)
Dcolp = colp;
} else if (cp->GetAmType() == TYPE_AM_FNC)
if (((PFNCCOL)cp)->InitColumn(g))
return TRUE;
} else
colp = new(g) SRCCOL(cdp, this, cprec, n);
} // endif isview
return colp;
} // end of MakeCol
return false;
} // end of MakeViewColumns
/***********************************************************************/
/* PIVOT GetMaxSize: returns the maximum number of rows in the table. */
......@@ -696,7 +377,7 @@ int TDBPIVOT::GetMaxSize(PGLOBAL g)
return MaxSize;
#endif // 0
return 0;
return 10;
} // end of GetMaxSize
/***********************************************************************/
......@@ -713,8 +394,6 @@ int TDBPIVOT::RowNumber(PGLOBAL g, bool b)
/***********************************************************************/
bool TDBPIVOT::OpenDB(PGLOBAL g)
{
//PDBUSER dup = (PDBUSER)g->Activityp->Aptr;
if (Use == USE_OPEN) {
/*******************************************************************/
/* Table already open, just replace it at its beginning. */
......@@ -747,16 +426,8 @@ bool TDBPIVOT::OpenDB(PGLOBAL g)
if (GetSourceTable(g))
return TRUE;
/*********************************************************************/
/* Check and initialize the subtable columns. */
/*********************************************************************/
for (PCOL cp = Columns; cp; cp = cp->GetNext())
if (cp->GetAmType() == TYPE_AM_SRC) {
if (((PPRXCOL)cp)->Init(g))
return TRUE;
} else if (cp->GetAmType() == TYPE_AM_FNC)
if (((PFNCCOL)cp)->InitColumn(g))
// For tables, columns must be allocated before opening
if (MakePivotColumns(g))
return TRUE;
/*********************************************************************/
......@@ -765,7 +436,10 @@ bool TDBPIVOT::OpenDB(PGLOBAL g)
if (Tdbp->OpenDB(g))
return TRUE;
return FALSE;
/*********************************************************************/
/* Make all required pivot columns for object views. */
/*********************************************************************/
return MakeViewColumns(g);
} // end of OpenDB
/***********************************************************************/
......@@ -881,17 +555,6 @@ void TDBPIVOT::CloseDB(PGLOBAL g)
} // end of CloseDB
#if 0
/***********************************************************************/
/* 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
#endif // 0
// ------------------------ FNCCOL functions ----------------------------
/***********************************************************************/
......@@ -948,8 +611,6 @@ bool FNCCOL::CompareColumn(void)
SRCCOL::SRCCOL(PCOLDEF cdp, PTDB tdbp, PCOL cprec, int n)
: PRXCOL(cdp, tdbp, cprec, n)
{
// Set additional SRC access method information for column.
//Cnval = NULL;
} // end of SRCCOL constructor
/***********************************************************************/
......@@ -960,9 +621,6 @@ bool SRCCOL::Init(PGLOBAL g)
if (PRXCOL::Init(g))
return true;
// Will contain the last value
//Cnval = AllocateValue(g, Value, TYPE_VOID);
AddStatus(BUF_READ); // All is done here
return false;
} // end of SRCCOL constructor
......@@ -972,7 +630,6 @@ bool SRCCOL::Init(PGLOBAL g)
/***********************************************************************/
void SRCCOL::SetColumn(void)
{
//Cnval->SetValue_pval(Value);
Value->SetValue_pval(To_Val);
} // end of SetColumn
......@@ -982,315 +639,7 @@ void SRCCOL::SetColumn(void)
bool SRCCOL::CompareLast(void)
{
// Compare the unconverted values
//return !Cnval->IsEqual(Colp->GetValue(), true);
return !Value->IsEqual(To_Val, true);
} // 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)
return FALSE;
/*********************************************************************/
/* 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 0
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
#endif // 0
/*******************************************************************/
/* !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 ---------------------- */
storage/connect/tabpivot.h
View file @ c035bde3
/************** TabPivot H Declares Source Code File (.H) **************/
/* Name: TABPIVOT.H Version 1.3 */
/* Name: TABPIVOT.H Version 1.4 */
/* */
/* (C) Copyright to the author Olivier BERTRAND 2005-2012 */
/* (C) Copyright to the author Olivier BERTRAND 2005-2013 */
/* */
/* This file contains the PIVOT classes declares. */
/***********************************************************************/
typedef class PIVOTDEF *PPIVOTDEF;
typedef class TDBPIVOT *PTDBPIVOT;
typedef class FNCCOL *PFNCCOL;
typedef class SRCCOL *PSRCCOL;
typedef class TDBQRS *PTDBQRS;
typedef class QRSCOL *PQRSCOL;
/* -------------------------- PIVOT classes -------------------------- */
......@@ -22,7 +21,6 @@ typedef class QRSCOL *PQRSCOL;
/***********************************************************************/
/* PIVOT table. */
/***********************************************************************/
//ass DllExport PIVOTDEF : public PRXDEF {/* Logical table description */
class PIVOTDEF : public PRXDEF { /* Logical table description */
friend class TDBPIVOT;
public:
......@@ -55,22 +53,16 @@ class PIVOTDEF : public PRXDEF { /* Logical table description */
/***********************************************************************/
/* This is the class declaration for the PIVOT table. */
/***********************************************************************/
//ass DllExport TDBPIVOT : public TDBASE, public CSORT {
class TDBPIVOT : public TDBPRX {
friend class FNCCOL;
//friend class SRCCOL;
public:
// Constructor
TDBPIVOT(PPIVOTDEF tdp);
//TDBPIVOT(PTDBPIVOT tdbp);
// Implementation
virtual AMT GetAmType(void) {return TYPE_AM_PIVOT;}
//virtual PTDB Duplicate(PGLOBAL g) {return (PTDB)new(g) TDBPIVOT(this);}
// void SetTdbp(PTDB tdbp) {Tdbp = tdbp;}
// Methods
//virtual PTDB CopyOne(PTABS t);
virtual int GetRecpos(void) {return N;}
virtual void ResetDB(void) {N = 0;}
virtual int RowNumber(PGLOBAL g, bool b = FALSE);
......@@ -84,22 +76,17 @@ class TDBPIVOT : public TDBPRX {
virtual int DeleteDB(PGLOBAL g, int irc);
virtual void CloseDB(PGLOBAL g);
// The sorting function
//virtual int Qcompare(int *, int *);
protected:
// Internal routines
bool GetSourceTable(PGLOBAL g);
//int MakePivotColumns(PGLOBAL g);
//bool UpdateTableFields(PGLOBAL g, int n);
bool MakePivotColumns(PGLOBAL g);
bool MakeViewColumns(PGLOBAL g);
// Members
//MYSQLC Myc; // MySQL connection class
//PTDBQRS Tqrp; // To the source table result
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
......@@ -108,9 +95,6 @@ class TDBPIVOT : public TDBPRX {
PCOL Fcolp; // To the function column in source
PCOL Xcolp; // To the pivot column in source
PCOL Dcolp; // To the dump column
//PCOLRES Xresp; // To the pivot result column
//PCOLRES To_Sort; // Saved Qryp To_Sort pointer
//PVBLK Rblkp; // The value block of the pivot column
bool GBdone; // True when subtable is "Group by"
bool Accept; // TRUE if no match is accepted
int Mult; // Multiplication factor
......@@ -168,75 +152,4 @@ class SRCCOL : public PRXCOL {
SRCCOL(void) {}
// Members
//PVAL Cnval;
}; // end of class SRCCOL
/***********************************************************************/
/* TDBQRS: This is the Access Method class declaration for the Query */
/* Result stored in memory in the current work area (volatil). */
/***********************************************************************/
class DllExport TDBQRS : public TDBASE {
friend class QRSCOL;
public:
// Constructor
TDBQRS(PQRYRES qrp) : TDBASE() {Qrp = qrp; CurPos = 0;}
TDBQRS(PTDBQRS tdbp);
// Implementation
virtual AMT GetAmType(void) {return TYPE_AM_QRS;}
virtual PTDB Duplicate(PGLOBAL g)
{return (PTDB)new(g) TDBQRS(this);}
PQRYRES GetQrp(void) {return Qrp;}
// Methods
virtual PTDB CopyOne(PTABS t);
virtual int RowNumber(PGLOBAL g, BOOL b = FALSE);
virtual int GetRecpos(void);
//virtual PCATLG GetCat(void);
//virtual PSZ GetPath(void);
virtual int GetBadLines(void) {return Qrp->BadLines;}
// Database routines
virtual PCOL ColDB(PGLOBAL g, PSZ name, int num);
virtual int GetMaxSize(PGLOBAL g);
virtual bool OpenDB(PGLOBAL g);
virtual int ReadDB(PGLOBAL g);
virtual int WriteDB(PGLOBAL g);
virtual int DeleteDB(PGLOBAL g, int irc);
virtual void CloseDB(PGLOBAL g);
private:
TDBQRS(void) : TDBASE() {} // Standard constructor not to be used
protected:
// Members
PQRYRES Qrp; // Points to Query Result block
int CurPos; // Current line position
}; // end of class TDBQRS
/***********************************************************************/
/* Class QRSCOL: QRS access method column descriptor. */
/***********************************************************************/
class DllExport QRSCOL : public COLBLK {
friend class TDBQRS;
public:
// Constructors
QRSCOL(PGLOBAL g, PCOLRES crp, PTDB tdbp, PCOL cprec, int i);
QRSCOL(QRSCOL *colp, PTDB tdbp); // Constructor used in copy process
// Implementation
virtual int GetAmType(void) {return TYPE_AM_QRS;}
PCOLRES GetCrp(void) {return Crp;}
void *GetQrsData(void) {return Crp->Kdata;}
// Methods
virtual void ReadColumn(PGLOBAL g);
virtual void Print(PGLOBAL g, FILE *, uint);
protected:
QRSCOL(void) {} // Default constructor not to be used
// Members
PCOLRES Crp;
}; // end of class QRSCOL
storage/connect/tabtbl.cpp
View file @ c035bde3
......@@ -216,7 +216,7 @@ bool TDBTBL::InitTableList(PGLOBAL g)
{
int n;
PTABLE tp, tabp;
PTDB tdbp;
PTDBASE tdbp;
PCOL colp;
PTBLDEF tdp = (PTBLDEF)To_Def;
......@@ -252,7 +252,7 @@ bool TDBTBL::InitTableList(PGLOBAL g)
n++;
} // endif filp
} // endfor tblp
} // endfor tp
//NumTables = n;
To_Filter = NULL; // To avoid doing it several times
......
storage/connect/tabutil.cpp
View file @ c035bde3
......@@ -90,12 +90,9 @@ TABLE_SHARE *GetTableShare(PGLOBAL g, THD *thd, const char *db,
// 1 2 4 8
//flags = GTS_TABLE | GTS_VIEW | GTS_NOLOCK | GTS_FORCE_DISCOVERY;
if (!open_table_def(thd, s, GTS_TABLE)) {
#ifdef DBUG_OFF
if (stricmp(s->db_plugin->name.str, "connect")) {
#else
if (stricmp((*s->db_plugin)->name.str, "connect")) {
#endif
if (!open_table_def(thd, s, GTS_TABLE | GTS_VIEW)) {
if (!s->is_view) {
if (stricmp(plugin_name(s->db_plugin)->str, "connect")) {
#if defined(MYSQL_SUPPORT)
mysql = true;
#else // !MYSQL_SUPPORT
......@@ -105,6 +102,10 @@ TABLE_SHARE *GetTableShare(PGLOBAL g, THD *thd, const char *db,
} else
mysql = false;
} else {
mysql = true;
} // endif is_view
} else {
sprintf(g->Message, "Error %d opening share\n", s->error);
free_table_share(s);
......@@ -139,9 +140,12 @@ PQRYRES TabColumns(PGLOBAL g, THD *thd, const char *db,
PCOLRES crp;
if (!info) {
if (!(s = GetTableShare(g, thd, db, name, mysql)))
if (!(s = GetTableShare(g, thd, db, name, mysql))) {
return NULL;
else
} else if (s->is_view) {
strcpy(g->Message, "Cannot retreive Proxy columns from a view");
return NULL;
} else
n = s->fieldnames.count;
} else {
......@@ -257,15 +261,19 @@ PRXDEF::PRXDEF(void)
/***********************************************************************/
bool PRXDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
{
char *pn, *db, *tab;
char *pn, *db, *tab, *def = NULL;
db = Cat->GetStringCatInfo(g, "Dbname", "*");
def = Cat->GetStringCatInfo(g, "Srcdef", NULL);
if (!(tab = Cat->GetStringCatInfo(g, "Tabname", NULL))) {
strcpy(g->Message, "Missing object table name");
if (!def) {
strcpy(g->Message, "Missing object table definition");
return TRUE;
} // endif tab
} else
tab = "Noname";
} else
// Analyze the table name, it may have the format: [dbname.]tabname
if ((pn = strchr(tab, '.'))) {
*pn++ = 0;
......@@ -273,7 +281,7 @@ bool PRXDEF::DefineAM(PGLOBAL g, LPCSTR am, int poff)
tab = pn;
} // endif pn
Tablep = new(g) XTAB(tab);
Tablep = new(g) XTAB(tab, def);
Tablep->SetQualifier(db);
return FALSE;
} // end of DefineAM
......@@ -303,12 +311,13 @@ TDBPRX::TDBPRX(PPRXDEF tdp) : TDBASE(tdp)
/***********************************************************************/
/* Get the PTDB of the sub-table. */
/***********************************************************************/
PTDB TDBPRX::GetSubTable(PGLOBAL g, PTABLE tabp)
PTDBASE TDBPRX::GetSubTable(PGLOBAL g, PTABLE tabp, bool b)
{
char *db, *name;
bool mysql;
bool mysql = true;
PTDB tdbp = NULL;
TABLE_SHARE *s;
TABLE_SHARE *s = NULL;
Field* *fp;
PCATLG cat = To_Def->GetCat();
PHC hc = ((MYCAT*)cat)->GetHandler();
LPCSTR cdb, curdb = hc->GetDBName(NULL);
......@@ -328,10 +337,20 @@ PTDB TDBPRX::GetSubTable(PGLOBAL g, PTABLE tabp)
} // endfor tp
if (!tabp->GetSrc()) {
if (!(s = GetTableShare(g, thd, db, name, mysql)))
return NULL;
if (s->is_view && !b)
s->field = hc->get_table()->s->field;
hc->tshp = s;
} else if (b) {
// Don't use caller's columns
fp = hc->get_table()->field;
hc->get_table()->field = NULL;
hc->get_table()->s->option_struct->srcdef = tabp->GetSrc();
} // endif srcdef
if (mysql) {
#if defined(MYSQL_SUPPORT)
......@@ -355,15 +374,23 @@ PTDB TDBPRX::GetSubTable(PGLOBAL g, PTABLE tabp)
tdbp = cat->GetTable(g, tabp);
} // endif mysql
if (s) {
if (s->is_view && !b)
s->field = NULL;
hc->tshp = NULL;
} else if (b)
hc->get_table()->field = fp;
if (trace && tdbp)
htrc("Subtable %s in %s\n",
name, SVP(((PTDBASE)tdbp)->GetDef()->GetDB()));
err:
if (s)
free_table_share(s);
return tdbp;
return (PTDBASE)tdbp;
} // end of GetSubTable
/***********************************************************************/
......@@ -373,7 +400,7 @@ bool TDBPRX::InitTable(PGLOBAL g)
{
if (!Tdbp) {
// Get the table description block of this table
if (!(Tdbp = (PTDBASE)GetSubTable(g, ((PPRXDEF)To_Def)->Tablep)))
if (!(Tdbp = GetSubTable(g, ((PPRXDEF)To_Def)->Tablep)))
return TRUE;
} // endif Tdbp
......
storage/connect/tabutil.h
View file @ c035bde3
......@@ -78,7 +78,7 @@ class DllExport TDBPRX : public TDBASE {
virtual int WriteDB(PGLOBAL g);
virtual int DeleteDB(PGLOBAL g, int irc);
virtual void CloseDB(PGLOBAL g) {if (Tdbp) Tdbp->CloseDB(g);}
PTDB GetSubTable(PGLOBAL g, PTABLE tabp);
PTDBASE GetSubTable(PGLOBAL g, PTABLE tabp, bool b = false);
void RemoveNext(PTABLE tp);
protected:
......@@ -93,6 +93,7 @@ class DllExport TDBPRX : public TDBASE {
class DllExport PRXCOL : public COLBLK {
friend class TDBPRX;
friend class TDBTBL;
friend class TDBOCCUR;
public:
// Constructors
PRXCOL(PCOLDEF cdp, PTDB tdbp, PCOL cprec, int i, PSZ am = "PRX");
......
storage/connect/xtable.h
View file @ c035bde3
......@@ -173,6 +173,7 @@ class DllExport TDBASE : public TDB {
virtual int GetRecpos(void) = 0;
virtual bool SetRecpos(PGLOBAL g, int recpos);
virtual bool IsReadOnly(void) {return Read_Only;}
virtual bool IsView(void) {return FALSE;}
virtual CHARSET_INFO *data_charset()
{
/*
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7