Implement calpont_discover_existence

dbcon/execplan/calpontsystemcatalog.cpp

@@ -368,6 +368,90 @@ CalpontSystemCatalog::CatalogMap CalpontSystemCatalog::fCatalogMap;
 /*static*/
 uint32_t CalpontSystemCatalog::fModuleID = numeric_limits<uint32_t>::max();
 
+const CalpontSystemCatalog::OID CalpontSystemCatalog::lookupTableOID(const TableName& tablename)
+{
+	TableName aTableName;
+	aTableName.schema = tablename.schema;
+	aTableName.table = tablename.table;	
+	transform( aTableName.schema.begin(), aTableName.schema.end(), aTableName.schema.begin(), to_lower() );
+	transform( aTableName.table.begin(), aTableName.table.end(), aTableName.table.begin(), to_lower() );
+
+	if (aTableName.schema.compare(CALPONT_SCHEMA) != 0)
+		DEBUG << "Enter tableInfo: " << tablename.schema << "|" << tablename.table << endl;
+
+	// select objectid from systable where schema = tableName.schema and tablename = tableName.table;
+	CalpontSelectExecutionPlan csep;
+	CalpontSelectExecutionPlan::ReturnedColumnList returnedColumnList;
+	CalpontSelectExecutionPlan::FilterTokenList filterTokenList;
+	CalpontSelectExecutionPlan::ColumnMap colMap;
+
+	SimpleColumn *c1 = new SimpleColumn(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+OBJECTID_COL, fSessionID);
+	SimpleColumn *c2 = new SimpleColumn(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+SCHEMA_COL, fSessionID);
+	SimpleColumn *c3 = new SimpleColumn(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+TABLENAME_COL, fSessionID);
+
+	SRCP srcp;
+	srcp.reset(c1);
+	colMap.insert(CMVT_(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+OBJECTID_COL, srcp));
+	srcp.reset(c2);
+	colMap.insert(CMVT_(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+SCHEMA_COL, srcp));
+	srcp.reset(c3);
+	colMap.insert(CMVT_(CALPONT_SCHEMA+"."+SYSTABLE_TABLE+"."+TABLENAME_COL, srcp));
+	csep.columnMapNonStatic(colMap);
+
+	srcp.reset(c1->clone());
+	returnedColumnList.push_back(srcp);
+	csep.returnedCols(returnedColumnList);
+	OID oid = c1->oid();
+
+	// Filters
+	SimpleFilter *f1 = new SimpleFilter (opeq,
+										 c2->clone(),
+										 new ConstantColumn(aTableName.schema, ConstantColumn::LITERAL));
+	filterTokenList.push_back(f1);
+	filterTokenList.push_back(new Operator("and"));
+
+	SimpleFilter *f2 = new SimpleFilter (opeq,
+										 c3->clone(),
+										 new ConstantColumn(aTableName.table, ConstantColumn::LITERAL));
+	filterTokenList.push_back(f2);
+	csep.filterTokenList(filterTokenList);
+
+	ostringstream oss;
+	oss << "select objectid from systable where schema='" << aTableName.schema << "' and tablename='" <<
+		aTableName.table << "' --tableRID/";
+
+	csep.data(oss.str()); //@bug 6078. Log the statement
+	if (fIdentity == EC) oss << "EC";
+	else oss << "FE";
+	NJLSysDataList sysDataList;
+	try {
+		getSysData (csep, sysDataList, SYSTABLE_TABLE);
+	}
+	catch ( IDBExcept& ){
+		throw;
+	}
+	catch ( runtime_error& e ) {
+		throw runtime_error ( e.what() );
+	}
+
+	vector<ColumnResult*>::const_iterator it;
+	for (it = sysDataList.begin(); it != sysDataList.end(); it++)
+	{
+		if ((*it)->dataCount() == 0)
+		{
+			return (OID)0;
+		}
+		if ((*it)->ColumnOID() == oid)
+		{
+			if (fIdentity == EC)
+				return (*it)->GetRid(0);
+			else
+			    return (OID)((*it)->GetData(0));
+		}	
+	}
+	return (OID)0;
+}
+
 const CalpontSystemCatalog::OID CalpontSystemCatalog::lookupOID(const TableColName& tableColName)
 {
     if (tableColName.schema.length() == 0 || tableColName.table.length() == 0 || tableColName.column.length() == 0)

dbcon/execplan/calpontsystemcatalog.h

@@ -496,6 +496,12 @@ public:
 	 */
 	typedef boost::shared_ptr<CalpontSystemCatalog> SPCSC;
 
+	/** looks up a table's OID in the System Catalog
+	 *
+	 * For a unique table_name return the internal OID
+	 */
+	const OID lookupTableOID(const TableName& tableName);
+
 	/** looks up a column's OID in the System Catalog
 	 *
 	 * For a unique table_name.column_name return the internal OID

dbcon/mysql/ha_calpont.cpp

@@ -137,6 +137,14 @@ static HASH calpont_open_tables;
 pthread_mutex_t calpont_mutex;
 #endif
 
+#ifdef DEBUG_ENTER
+#undef DEBUG_ENTER
+#endif
+#ifdef DEBUG_RETURN
+#undef DEBUG_ENTER
+#endif
+#define DEBUG_RETURN return
+
 /**
   @brief
   Function we use in the creation of our hash to get key.
@@ -154,14 +162,17 @@ int calpont_discover(handlerton *hton, THD* thd, TABLE_SHARE *share)
 #ifdef INFINIDB_DEBUG
 fprintf(stderr, "calpont_discover()\n");
 #endif
-DBUG_PRINT("info", ("calpont_discover."));
 	return 1;
 }
+int calpont_discover_existence(handlerton *hton, const char *db,
+                                const char *table_name)
+{
+	return ha_calpont_impl_discover_existence(db, table_name);
+}
 
 static int calpont_init_func(void *p)
 {
   DBUG_ENTER("calpont_init_func");
-DBUG_PRINT("info", ("calpont_init_func."));
 
   struct tm tm;
   time_t t;
@@ -184,7 +195,8 @@ DBUG_PRINT("info", ("calpont_init_func."));
   calpont_hton->state=   SHOW_OPTION_YES;
   calpont_hton->create=  calpont_create_handler;
   calpont_hton->flags=   HTON_CAN_RECREATE;
-  calpont_hton->discover_table=   calpont_discover;
+  calpont_hton->discover_table= calpont_discover;
+  calpont_hton->discover_table_existence= calpont_discover_existence;
   calpont_hton->commit= calpont_commit;
   calpont_hton->rollback= calpont_rollback;
   calpont_hton->close_connection = calpont_close_connection;
@@ -285,7 +297,6 @@ static handler* calpont_create_handler(handlerton *hton,
                                        TABLE_SHARE *table, 
                                        MEM_ROOT *mem_root)
 {
-DBUG_PRINT("info", ("calpont_create_handler."));
   return new (mem_root) ha_calpont(hton, table);
 }
 
@@ -303,7 +314,6 @@ static int calpont_rollback(handlerton *hton, THD* thd, bool all)
 
 static int calpont_close_connection ( handlerton *hton, THD* thd )
 {
-DBUG_PRINT("info", ("calpont_close_connection."));
 	int rc = ha_calpont_impl_close_connection( hton, thd);
 	return rc;
 }
@@ -316,7 +326,6 @@ static void calpont_set_error(THD* thd, uint64_t errCode, LEX_STRING* args, uint
 ha_calpont::ha_calpont(handlerton *hton, TABLE_SHARE *table_arg)
   :handler(hton, table_arg)
 {
-DBUG_PRINT("info", ("ha_calpont::ha_calpont."));
 }
 
 

dbcon/mysql/ha_calpont_impl.cpp

@@ -2305,6 +2305,21 @@ int ha_calpont_impl_close(void)
 	return 0;
 }
 
+int ha_calpont_impl_discover_existence(const char *schema, const char *name)
+{
+	boost::shared_ptr<CalpontSystemCatalog> csc = CalpontSystemCatalog::makeCalpontSystemCatalog();
+	try
+	{
+		const CalpontSystemCatalog::OID oid = csc->lookupTableOID(make_table(schema, name));
+		if (oid)
+			return 1;
+	}
+	catch ( ... )
+	{
+	}
+	return 0;
+}
+
 int ha_calpont_impl_rnd_init(TABLE* table)
 {
 #ifdef DEBUG_SETENV

dbcon/mysql/ha_calpont_impl.h

@@ -27,6 +27,7 @@
 #include "idb_mysql.h"
 
 #ifdef NEED_CALPONT_EXTERNS
+extern int ha_calpont_impl_discover_existence(const char *schema, const char *name);
 extern int ha_calpont_impl_create(const char *name, TABLE *table_arg, HA_CREATE_INFO *create_info);
 extern int ha_calpont_impl_delete_table(const char *name);
 extern int ha_calpont_impl_open(const char *name, int mode, uint32_t test_if_locked);