Skip to content

Commit

Permalink
Merge pull request #25 from RussWhitehead/hpccSchema
Browse files Browse the repository at this point in the history 
ODBC-32 Create output columns using HPCC Metadata

Reviewed-By: Rodrigo Pastrana <[email protected]>
Reviewed-By: Richard Chapman <[email protected]>
  • Loading branch information
@richardkchapman
richardkchapman committed Jun 4, 2015
2 parents adac65a + 6383fb6 commit 50527dd
Show file tree
Hide file tree
Showing 5 changed files with 164 additions and 57 deletions.
52 changes: 1 addition & 51 deletions src/hpcc_drv.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,6 @@ int OAIP_execute(IP_HDBC hdbc,
HPCC_CONN_DA * pConnDA = (HPCC_CONN_DA *)hdbc;
HPCC_STMT_DA * pStmtDA;
XM_Tree * pMemTree;
DAM_HCOL hcol;

tm_trace(hpcc_tm_Handle, UL_TM_F_TRACE, "HPCC_Conn:OAIP_execute() has been called\n", (0));

Expand All @@ -471,57 +470,8 @@ int OAIP_execute(IP_HDBC hdbc,
pStmtDA->dam_hstmt = hstmt;
pStmtDA->iType = iStmtType;

/* get the table information */
IArrayOf<CTable> _tables;
// get the table name
dam_describeTable(pStmtDA->dam_hstmt, NULL, NULL, pStmtDA->sTableName, NULL, (char*)NULL);
pConnDA->pHPCCdb->getTableSchema(pStmtDA->sTableName, _tables);
pConnDA->pHPCCdb->clearOutputColumnDescriptors();
if (_tables.empty())
{
StringBuffer err;
err.setf("HPCC_Conn:OAIP_execute : Table '%s' not found", pStmtDA->sTableName);
dam_addError(hdbc, NULL, DAM_IP_ERROR, 0, (char*)err.str());
return DAM_FAILURE;
}
assertex(!_tables.empty());
CTable &table = _tables.item(0);

//-----------------------------------------------------------
//The following calls retrieve the output table column format.
//Look up these columns in our hpcc CColumn cache, and save them
//in an ordered array of CColumn descriptors. Once data is retrieved (below)
//we can iterate over these columns to add the columns to OpenAcces rows
//-----------------------------------------------------------
hcol = dam_getFirstCol(pStmtDA->dam_hstmt, DAM_COL_IN_USE);
while (hcol)
{
char sColName[DAM_MAX_ID_LEN+1];
int iColNum;
int iXOType;
int iColType;

dam_describeCol(hcol, //DAM_HCOL hcol,
&iColNum, //int *piColNum,
sColName, //char *pColName,
&iXOType, //int *piXOType,
&iColType); //int *piColType);

CColumn *pCol = table.queryColumn(iColNum);//find this hpcc CColumn descriptor
if (!pCol)
{
StringBuffer err;
err.setf("HPCC_Conn:OAIP_execute : Table '%s', Column '%s' not found", table.queryName(), sColName);
dam_addError(hdbc, NULL, DAM_IP_ERROR, 0, (char*)err.str());
return DAM_FAILURE;
}

assert(0 == strcmp(sColName, pCol->m_name));
// pCol->m_iColNum = iColNum;
assert(pCol->m_iXOType = iXOType);
pCol->m_hcol = hcol;
pConnDA->pHPCCdb->addOutputColumnDescriptor(pCol);//save in ordered array of column descriptors
hcol = dam_getNextCol(pStmtDA->dam_hstmt);
}

//-----------------------------------
//Build the SQL SELECT query string
Expand Down
37 changes: 35 additions & 2 deletions src/hpcc_util.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,12 @@ int hpcc_add_row(HPCC_STMT_DA *pStmtDA, DAM_HROW hrow, IPropertyTree * pRow, CCo
int64 i64Val;
double dVal;

const char * value = pRow->queryProp(pCol->m_name);//string representation of the row data
if (pCol->m_hcol == (DAM_HCOL)UNINITIALIZED)//Progress column doesnt exist. This happens on COUNT(*) and similar, so lets ignore it for now
return DAM_SUCCESS;

const char * value = pRow->queryProp(pCol->m_alias);//Alias takes precedence over name
if (NULL == value)
value = pRow->queryProp(pCol->m_name);
if (NULL == value)
{
//No value provided. Inject empty string or "XO_NULL_DATA"
Expand Down Expand Up @@ -97,7 +102,7 @@ int hpcc_exec(const char * sqlQuery, const char * targetQuerySet, HP

//call ws_sql to get row(s)
sbErrors.set("HPCC_Conn:hpcc_exec : ");
if (!pHPCCdb->executeSQL(sqlQuery, targetQuerySet, sbErrors))
if (!pHPCCdb->executeSQL(sqlQuery, targetQuerySet, sbErrors, pStmtDA))
{
dam_addError(pStmtDA->pConnDA->dam_hdbc, pStmtDA->dam_hstmt, DAM_IP_ERROR, 0, (char *)sbErrors.str());
return DAM_FAILURE;
Expand Down Expand Up @@ -138,3 +143,31 @@ int hpcc_exec(const char * sqlQuery, const char * targetQuerySet, HP
tm_trace(hpcc_tm_Handle, UL_TM_F_TRACE, "HPCC_Conn:Read '%ld' rows\n", (*piNumRows));
return DAM_SUCCESS;
}


/************************************************************************
Function: queryColumnDetails
Description: Get the table name
************************************************************************/
bool queryColumnDetails(/*input*/void *pStmtDA, /*input*/aindex_t colIdx,
StringAttr &tblName, int * piColNum, int * piXOType, DAM_HCOL * phcol)
{
aindex_t idx = 0;
DAM_HCOL hcol = dam_getFirstCol(((HPCC_STMT_DA*)pStmtDA)->dam_hstmt, DAM_COL_IN_USE);
while (hcol && idx < colIdx)
{
hcol = dam_getNextCol(((HPCC_STMT_DA*)pStmtDA)->dam_hstmt);
++idx;
}

if (hcol && idx == colIdx)
{
char sColName[DAM_MAX_ID_LEN+1];
int colType;
dam_describeCol(hcol, piColNum, sColName, piXOType, &colType);
tblName.set(((HPCC_STMT_DA*)pStmtDA)->sTableName);
*phcol = hcol;
return true;
}
return false;
}
2 changes: 2 additions & 0 deletions src/hpcc_util.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -55,5 +55,7 @@ typedef struct hpcc_statement_struct
extern TM_ModuleCB hpcc_tm_Handle; /* declared in hpcc_drv.c */

int hpcc_exec(const char * sqlQuery, const char * targetQuerySet, HPCC_STMT_DA *pStmtDA, int *piNumResRows);
bool queryColumnDetails(/*input*/void *pStmtDA, /*input*/aindex_t ColIdx,
StringAttr &tblName, int * piColNum, int * piXOType, DAM_HCOL * phcol);

#endif /* __HPCCUTIL_HPP */
116 changes: 113 additions & 3 deletions src/hpccdb.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,9 @@
#include "common.esp"

static TM_ModuleCB driver_tm_Hdle;//for logging

extern void populateOAtypes(CColumn * pColumn);
extern bool queryColumnDetails(/*input*/void *pStmtDA, /*input*/aindex_t ColIdx,
StringAttr &tblName, int * piColNum, int * piXOType, DAM_HCOL * phcol);

//*******************************************************************
//Shared schema cache singleton, persists even when connections close
Expand Down Expand Up @@ -560,10 +562,11 @@ Description: Call esp "ws_sql" service to execute the given SQL "SELECT" or "
Return: true on Success
false unsuccessful
************************************************************************/
bool HPCCdb::executeSQL(const char * sql, const char * targetQuerySet, StringBuffer & sbErrors)
bool HPCCdb::executeSQL(const char * sql, const char * targetQuerySet, StringBuffer & sbErrors, void *pStmtDA)
{
tm_trace(driver_tm_Hdle, UL_TM_INFO, "HPCC_Conn:call to HPCCdb::executeSQL with query '%s'\n", (sql));
killResultsDatasets();
clearOutputColumnDescriptors();

Owned<IClientExecuteSQLRequest> req;
Owned<IClientExecuteSQLResponse> resp;
Expand Down Expand Up @@ -618,7 +621,7 @@ bool HPCCdb::executeSQL(const char * sql, const char * targetQuerySet, StringBuf
#endif
Owned<IPropertyTree> resultsTree;
{
Owned<IPTreeMaker> maker = createRootLessPTreeMaker();
Owned<IPTreeMaker> maker = createRootLessPTreeMaker(ipt_ordered);
resultsTree.setown(createPTreeFromXMLString(resultXML, ipt_none, (PTreeReaderOptions)(ptr_noRoot | ptr_ignoreWhiteSpace), maker));
}
#ifdef _DEBUG
Expand All @@ -643,6 +646,113 @@ bool HPCCdb::executeSQL(const char * sql, const char * targetQuerySet, StringBuf
}
}
}

//Iterate over the returned schema looking for column aliases (sumout1, etc)
StringBuffer sb;
Owned<IPropertyTreeIterator> it1;
it1.setown(resultsTree->getElements("XmlSchema"));
ForEach(*it1)
{
IPropertyTree & pt1 = it1->query();
Owned<IPropertyTreeIterator> it2;
it2.setown(pt1.getElements("xs:schema"));
ForEach(*it2)
{
IPropertyTree & pt2 = it2->query();
Owned<IPropertyTreeIterator> it3;
it3.setown(pt2.getElements("xs:element"));
ForEach(*it3)
{
IPropertyTree & pt3 = it3->query();
if (strieq(pt3.queryProp("@name"),"Dataset"))
{
Owned<IPropertyTreeIterator> it4;
it4.setown(pt3.getElements("xs:complexType"));
ForEach(*it4)
{
IPropertyTree & pt4 = it4->query();
Owned<IPropertyTreeIterator> it5;
it5.setown(pt4.getElements("xs:sequence"));
ForEach(*it5)
{
IPropertyTree & pt5 = it5->query();
Owned<IPropertyTreeIterator> it6;
it6.setown(pt5.getElements("xs:element"));
ForEach(*it6)
{
IPropertyTree & pt6 = it6->query();
if (strieq(pt6.queryProp("@name"),"Row"))
{
Owned<IPropertyTreeIterator> it7;
it7.setown(pt6.getElements("xs:complexType"));
ForEach(*it7)
{
IPropertyTree & pt7 = it7->query();
Owned<IPropertyTreeIterator> it8;
it8.setown(pt7.getElements("xs:sequence"));
ForEach(*it8)
{
IPropertyTree & pt8 = it8->query();
Owned<IPropertyTreeIterator> it9;
aindex_t iColIdx = 0;
it9.setown(pt8.getElements("xs:element"));
ForEach(*it9)
{
IPropertyTree & pt9 = it9->query();
DUMPPTREE(&pt9);

//allocate and append new output CColumn
Owned<CColumn> outputColumn;
const char * pColName = pt9.queryProp("@name");
outputColumn.setown(new CColumn(pt9.queryProp(pColName)));
addOutputColumnDescriptor(LINK(outputColumn));

//Populate the output CColumn
outputColumn->m_alias.set(pColName);
int colNum;
StringAttr tblName;
DAM_HCOL hcol;

//Ask Progress to tell us about this output column
if (queryColumnDetails(pStmtDA, iColIdx, tblName, &colNum, (int*)&outputColumn->m_iXOType, &hcol))
{
//Lets find this output column in the schema cache so we can reuse the populated data type contents
IArrayOf<CTable> _tables;
getTableSchema(tblName.str(), _tables);
if (!_tables.empty())
{
CTable &table = _tables.item(0);
CColumn * pCol = table.queryColumn(colNum);
if (pCol)
{
outputColumn->copyFrom(pCol);
outputColumn->m_hcol = hcol;
outputColumn->m_alias.set(pColName);
}
}
}
else
{
//This must be a scalar like the result of a SUM(1) or COUNT(*) or something like that
const char * pHPCCType = pt9.queryProp("@type");
const char * pp = strchr(pHPCCType, ':');
outputColumn->m_hpccType.set(pp ? pp + 1 : pHPCCType);
outputColumn->m_iXOType = UNINITIALIZED;
outputColumn->m_hcol = (DAM_HCOL)UNINITIALIZED;
}
populateOAtypes(outputColumn.get());//map HPCC data types to OpenAccess types
++iColIdx;
}
}
}
}
}
}
}
}
}
}
}
return true;
}

Expand Down
14 changes: 13 additions & 1 deletion src/hpccdb.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,18 @@ class CColumn : public CInterface, public IInterface
CColumn(const char * _name) : m_name(_name), m_iXOType(UNINITIALIZED) {}
virtual ~CColumn() {}

void copyFrom(CColumn * pFrom)
{
m_name.set(pFrom->m_name);
m_alias.set(pFrom->m_alias);
m_hpccType.set(pFrom->m_hpccType);

m_iXOType = pFrom->m_iXOType;
m_type_name.set(pFrom->m_type_name);
m_char_max_length = pFrom->m_char_max_length;
m_numeric_precision = pFrom->m_numeric_precision;
m_hcol = pFrom->m_hcol;
}
//HPCC attributes, queried from wssql
StringAttr m_name; // Column/dataElement name
StringAttr m_alias; // sumout1, etc
Expand Down Expand Up @@ -264,7 +276,7 @@ class HPCCdb
//ws_sql calls
bool getHPCCDBSystemInfo();
bool getTableSchema(const char * _tableFilter, IArrayOf<CTable> &_tables);
bool executeSQL(const char * sql, const char * targetQuerySet, StringBuffer & sbErrors);
bool executeSQL(const char * sql, const char * targetQuerySet, StringBuffer & sbErrors, void *pStmtDA);
bool getMoreResults(const char * _wuid, const char * dsName, aindex_t _start, aindex_t _count, IPropertyTree ** _ppResultsTree, StringBuffer & _sbErrors);
bool executeStoredProcedure(const char * procName, const char * querySet);

Expand Down

0 comments on commit 50527dd

Please sign in to comment.