/* joinerCheck - Parse and check joiner file. */

#include "common.h"
#include "linefile.h"
#include "hash.h"
#include "options.h"
#include "dystring.h"
#include "obscure.h"
#include "jksql.h"
#include "joiner.h"


/* Variable that are set from command line. */
char *fieldListIn;
char *fieldListOut;
char *identifier;
char *database;
boolean foreignKeys;	/* "keys" command line variable. */
boolean checkTimes;	/* "times" command line variable. */
boolean checkFields;	/* "fields" command line variable. */
boolean tableCoverage;
boolean dbCoverage;

void usage()
/* Explain usage and exit. */
{
errAbort(
  "joinerCheck - Parse and check joiner file\n"
  "usage:\n"
  "   joinerCheck file.joiner\n"
  "options:\n"
  "   -fields - Check fields in joiner file exist, faster with -fieldListIn\n"
  "      -fieldListOut=file - List all fields in all databases to file.\n"
  "      -fieldListIn=file - Get list of fields from file rather than mysql.\n"
  "   -keys - Validate (foreign) keys.  Takes about an hour.\n"
  "   -tableCoverage - Check that all tables are mentioned in joiner file\n"
  "   -dbCoverage - Check that all databases are mentioned in joiner file\n"
  "   -times - Check update times of tables are after tables they depend on\n"
  "   -all - Do all tests: -fields -keys -tableCoverage -dbCoverage -times\n"
  "   -identifier=name - Just validate given identifier.\n"
  "                    Note only applies to keys and fields checks.\n"
  "   -database=name - Just validate given database.\n"
  "                    Note only applies to keys and times checks.\n"
  "   -verbose=N - use verbose to diagnose difficulties. N = 2, 3 or 4 to\n"
  "              - show increasing level of detail for some functions."
  );
}

static struct optionSpec options[] = {
   {"fieldListIn", OPTION_STRING},
   {"fieldListOut", OPTION_STRING},
   {"identifier", OPTION_STRING},
   {"database", OPTION_STRING},
   {"keys", OPTION_BOOLEAN},
   {"times", OPTION_BOOLEAN},
   {"fields", OPTION_BOOLEAN},
   {"tableCoverage", OPTION_BOOLEAN},
   {"dbCoverage", OPTION_BOOLEAN},
   {"all", OPTION_BOOLEAN},
   {NULL, 0},
};

/* Other globals */
struct hash *allDbHash;

static void printField(struct joinerField *jf, FILE *f)
/* Print out field info to f. */
{
struct slName *db;
for (db = jf->dbList; db != NULL; db = db->next)
    {
    if (db != jf->dbList)
        fprintf(f, ",");
    fprintf(f, "%s", db->name);
    }
fprintf(f, ".%s.%s", jf->table, jf->field);
}

void checkOneDependency(struct joiner *joiner,
	struct joinerDependency *dep, struct sqlConnection *conn, char *dbName)
/* Check out one dependency in one database. */
{
char *tableToCheck = dep->table->table;
if (sqlWildcardIn(tableToCheck))
    {
    errAbort("Can't handle wildCards in dependency tables line %d of %s",
    	dep->lineIx, joiner->fileName);
    }
if (slNameInList(dep->table->dbList, dbName) 
	&& sqlTableExists(conn, tableToCheck))
    {
    time_t tableTime = sqlTableUpdateTime(conn, tableToCheck);
    struct joinerTable *dependsOn;
    for (dependsOn = dep->dependsOnList; dependsOn != NULL; 
	dependsOn = dependsOn->next)
	{
	if (slNameInList(dependsOn->dbList, dbName))
	    {
	    if (!sqlTableExists(conn, dependsOn->table))
		{
		warn("Error: %s.%s doesn't exist line %d of %s",
		    dbName, dependsOn->table, 
		    dep->lineIx, joiner->fileName);
		}
	    else
		{
		time_t depTime = sqlTableUpdateTime(conn, dependsOn->table);
		if (depTime > tableTime)
		    {
		    char *depStr = sqlUnixTimeToDate(&depTime, FALSE);
		    char *tableStr = sqlUnixTimeToDate(&tableTime, FALSE);
		
		    warn("Error: %s.%s updated after %s.%s line %d of %s",
			dbName, dependsOn->table, dbName, tableToCheck,
			dep->lineIx, joiner->fileName);
		    warn("\t%s vs. %s", depStr, tableStr);
		    freeMem(depStr);
		    freeMem(tableStr);
		    }
		}
	    }
	}
    }
}

void joinerCheckDependencies(struct joiner *joiner, char *specificDb)
/* Check time stamps on dependent tables. */
{
struct hashEl *db, *dbList = hashElListHash(joiner->databasesChecked);
for (db = dbList; db != NULL; db = db->next)
    {
    if (specificDb == NULL || sameString(specificDb, db->name))
        {
	struct sqlConnection *conn = sqlMayConnect(db->name);
	if (conn != NULL)	/* We've already warned on this NULL */
	    {
	    struct joinerDependency *dep;
	    for (dep = joiner->dependencyList; dep != NULL; dep = dep->next)
	        {
		checkOneDependency(joiner, dep, conn, db->name);
		}
	    sqlDisconnect(&conn);
	    }
	}
    }
slFreeList(&dbList);
}

struct slName *getTablesForField(struct sqlConnection *conn, 
	char *splitPrefix, char *table, char *splitSuffix)
/* Get tables that match field. */
{
struct slName *list = NULL, *el;
if (splitPrefix != NULL || splitSuffix != NULL)
    {
    char query[256], **row;
    struct sqlResult *sr;
    safef(query, sizeof(query), "show tables like '%s%s%s'", 
    	emptyForNull(splitPrefix), table, emptyForNull(splitSuffix));
    sr = sqlGetResult(conn, query);
    while ((row = sqlNextRow(sr)) != NULL)
        {
	el = slNameNew(row[0]);
	slAddHead(&list, el);
	}
    sqlFreeResult(&sr);
    slReverse(&list);
    }
if (list == NULL)
    {
    if (sqlTableExists(conn, table))
	list = slNameNew(table);
    }
return list;
}

static boolean fieldExists(struct hash *fieldHash,
	struct joinerSet *js, struct joinerField *jf)
/* Make sure field exists in at least one database. */
{
struct slName *db;
boolean gotIt = FALSE;
for (db = jf->dbList; db != NULL && !gotIt; db = db->next)
    {
    if (hashLookup(allDbHash, db->name))
	{
	struct sqlConnection *conn = sqlConnect(db->name);
	struct slName *table, *tableList = getTablesForField(conn,
			    jf->splitPrefix, jf->table, jf->splitSuffix);
	char fieldName[512];
	sqlDisconnect(&conn);
	for (table = tableList; table != NULL; table = table->next)
	    {
	    safef(fieldName, sizeof(fieldName), "%s.%s.%s", 
		db->name, table->name, jf->field);
	    if (hashLookup(fieldHash, fieldName))
		{
		gotIt = TRUE;
		break;
		}
	    }
	slFreeList(&tableList);
	}
    else
        {
	/* Warn that database doesn't exist.  Just warn once per database. */
	static struct hash *uniqHash;
	if (uniqHash == NULL) 
	    uniqHash = hashNew(8);
	if (!hashLookup(uniqHash, db->name))
	    {
	    hashAdd(uniqHash, db->name, NULL);
	    warn("Database %s doesn't exist", db->name);
	    }
	}
    }
return gotIt;
}

static void joinerValidateFields(struct joiner *joiner, struct hash *fieldHash,
	char *oneIdentifier)
/* Make sure that joiner refers to fields that exist at
 * least somewhere. */
{
struct joinerSet *js;
struct joinerField *jf;

for (js=joiner->jsList; js != NULL; js = js->next)
    {
    if (oneIdentifier == NULL || wildMatch(oneIdentifier, js->name))
	{
	for (jf = js->fieldList; jf != NULL; jf = jf->next)
	    {
	    if (!fieldExists(fieldHash, js, jf))
		 {
		 if (!js->expanded)
		     {
		     fprintf(stderr, "Error: ");
		     printField(jf, stderr);
		     fprintf(stderr, " not found in %s line %d of %s\n",
			js->name, jf->lineIx, joiner->fileName);
		     }
		 }
	    }
	}
    }
}

struct slName *jsAllDb(struct joinerSet *js)
/* Get list of all databases referred to by set. */
{
struct slName *list = NULL, *db;
struct joinerField *jf;
for (jf = js->fieldList; jf != NULL; jf = jf->next)
    {
    for (db = jf->dbList; db != NULL; db = db->next)
	slNameStore(&list, db->name);
    }
return list;
}

int sqlTableRows(struct sqlConnection *conn, char *table)
/* REturn number of rows in table. */
{
char query[256];
safef(query, sizeof(query), "select count(*) from %s", table);
return sqlQuickNum(conn, query);
}

int totalTableRows(struct sqlConnection *conn, struct slName *tableList)
/* Return total number of rows in all tables in list. */
{
int rowCount = 0;
struct slName *table;
for (table = tableList; table != NULL; table = table->next)
    {
    rowCount += sqlTableRows(conn, table->name);
    }
return rowCount;
}

struct sqlConnection *sqlWarnConnect(char *db)
/* Connect to database, or warn and return NULL. */
{
struct sqlConnection *conn = sqlMayConnect(db);
if (conn == NULL)
    warn("Error: Couldn't connect to database %s", db);
return conn;
}

static char *doChopsAndUpper(struct joinerField *jf, char *id)
/* Return chopped version of id.  (This may insert a zero into s).
 * Also upper case it. */
{
struct slName *chop;
for (chop = jf->chopBefore; chop != NULL; chop = chop->next)
    {
    char *s = stringIn(chop->name, id);
    if (s != NULL)
	 id = s + strlen(chop->name);
    }
for (chop = jf->chopAfter; chop != NULL; chop = chop->next)
    {
    char *s = rStringIn(chop->name, id);
    if (s != NULL)
	*s = 0;
    }
touppers(id);
return id;
}

struct keyHitInfo
/* Information on how many times a key is hit. */
    {
    struct keyHitInfo *next;
    char *name;	/* Name - not allocated here. */
    int count;	/* Hit count. */
    };

void keyHitInfoClear(struct keyHitInfo *khiList)
/* Clear out counts in list. */
{
struct keyHitInfo *khi;
for (khi = khiList; khi != NULL; khi = khi->next)
    khi->count = 0;
}

struct hash *readKeyHash(char *db, struct joiner *joiner, 
	struct joinerField *keyField, struct keyHitInfo **retList)
/* Read key-field into hash.  Check for dupes if need be. */
{
struct sqlConnection *conn = sqlWarnConnect(db);
struct hash *keyHash = NULL;
struct keyHitInfo *khiList = NULL, *khi;
if (conn == NULL)
    {
    return NULL;
    }
else
    {
    struct slName *table;
    struct slName *tableList = getTablesForField(conn,keyField->splitPrefix,
    						 keyField->table, 
						 keyField->splitSuffix);
    int rowCount = totalTableRows(conn, tableList);
    int hashSize = digitsBaseTwo(rowCount)+1;
    char query[256], **row;
    struct sqlResult *sr;
    int itemCount = 0;
    int dupeCount = 0;
    char *dupe = NULL;

    if (rowCount > 0)
	{
	if (hashSize > hashMaxSize)
	    hashSize = hashMaxSize;
	keyHash = hashNew(hashSize);
	for (table = tableList; table != NULL; table = table->next)
	    {
	    safef(query, sizeof(query), "select %s from %s", 
		keyField->field, table->name);
	    sr = sqlGetResult(conn, query);
	    while ((row = sqlNextRow(sr)) != NULL)
		{
		char *id = doChopsAndUpper(keyField, row[0]);
		if (hashLookup(keyHash, id))
		    {
		    if (keyField->unique)
			{
			if (keyField->exclude == NULL || 
				!slNameInList(keyField->exclude, id))
			    {
			    if (dupeCount == 0)
				dupe = cloneString(id);
			    ++dupeCount;
			    }
			}
		    }
		else
		    {
		    AllocVar(khi);
		    hashAddSaveName(keyHash, id, khi, &khi->name);
		    slAddHead(&khiList, khi);
		    ++itemCount;
		    }
		}
	    sqlFreeResult(&sr);
	    }
	if (dupe != NULL)
	    {
	    warn("Error: %d duplicates in %s.%s.%s including '%s'",
		    dupeCount, db, keyField->table, keyField->field, dupe);
	    freez(&dupe);
	    }
	verbose(2, " %s.%s.%s - %d unique identifiers\n", 
		db, keyField->table, keyField->field, itemCount);
	}
    slFreeList(&tableList);
    }
sqlDisconnect(&conn);
*retList = khiList;
return keyHash;
}

static void addHitMiss(struct hash *keyHash, char *id, struct joinerField *jf,
	int *pHits, char **pMiss, int *pTotal)
/* Record info about one hit or miss to keyHash */
{
id = doChopsAndUpper(jf, id);
if (jf->exclude == NULL || !slNameInList(jf->exclude, id))
    {
    struct keyHitInfo *khi;
    if ((khi = hashFindVal(keyHash, id)) != NULL)
	{
	*pHits += 1;
	khi->count += 1;
	}
    else
	{
	if (*pMiss == NULL)
	    *pMiss = cloneString(id);
	}
    *pTotal += 1;
    }
}

void checkUniqueAndFull(struct joiner *joiner, struct joinerSet *js, 
	char *db, struct joinerField *jf, struct joinerField *keyField,
	struct keyHitInfo *khiList)
/* Make sure that unique and full fields really are so. */
{
struct keyHitInfo *khi;
int keyCount = 0, missCount = 0, doubleCount = 0;
char *missExample = NULL, *doubleExample = NULL;

for (khi = khiList; khi != NULL; khi = khi->next)
    {
    ++keyCount;
    if (khi->count == 0)
	 {
         ++missCount;
	 missExample = khi->name;
	 }
    else if (khi->count > 1)
	{
        ++doubleCount;
	doubleExample = khi->name;
	}
    }
if (jf->full && missCount != 0)
    warn("Error: %d of %d elements of key %s.%s are not in 'full' %s.%s.%s line %d of %s\n"
	 "    Example miss: %s"
	, missCount, keyCount
	, keyField->table, keyField->field
	, db, jf->table, jf->field
	, jf->lineIx, joiner->fileName, missExample);
if (jf->unique && doubleCount != 0)
    warn("Error: %d of %d elements of %s.%s.%s are not unique line %d of %s\n"
	 "    Example: %s"
	, doubleCount, keyCount
	, db, jf->table, jf->field
	, jf->lineIx, joiner->fileName, doubleExample);
}

void doKeyChecks(char *db, struct joiner *joiner, struct joinerSet *js, 
	struct hash *keyHash, struct keyHitInfo *khiList,
	struct joinerField *keyField, struct joinerField *jf)
/* Check that at least minimum is covered, and that full and
 * unique tags are respected. */
{
struct sqlConnection *conn = sqlMayConnect(db);
if (conn != NULL)
    {
    boolean okFlag = FALSE;
    int total = 0, hits = 0, hitsNeeded;
    char *miss = NULL;
    struct slName *table;
    struct slName *tableList = getTablesForField(conn,jf->splitPrefix,
    						 jf->table, jf->splitSuffix);
    keyHitInfoClear(khiList);
    for (table = tableList; table != NULL; table = table->next)
	{
	char query[256], **row;
	struct sqlResult *sr;
	safef(query, sizeof(query), "select %s from %s", 
		jf->field, table->name);
	sr = sqlGetResult(conn, query);
	while ((row = sqlNextRow(sr)) != NULL)
	    {
	    if (jf->separator == NULL)
		{
		addHitMiss(keyHash, row[0], jf, &hits, &miss, &total);
		}
	    else 
	        {
		/* Do list. */
		struct slName *el, *list;
		int ix;
		list = slNameListFromString(row[0], jf->separator[0]);
		for (el = list, ix=0; el != NULL; el = el->next, ++ix)
		    {
		    char *id = el->name;
		    char buf[16];
		    if (jf->indexOf)
			{
		        safef(buf, sizeof(buf), "%d", ix);
			id = buf;
			}
		    addHitMiss(keyHash, id, jf, &hits, &miss, &total);
		    }
		slFreeList(&list);
		}
	    }
	sqlFreeResult(&sr);
	}
    if (tableList != NULL)
	{
	okFlag = FALSE;
	if (hits==total) okFlag = TRUE;
	if (jf->minCheck < 0.999999)	/* Control for rounding error */
	    hitsNeeded = round(total * jf->minCheck);
	else
	    hitsNeeded = total;
	if (jf->minCheck < 1.0 && hits >= hitsNeeded) okFlag = TRUE;
	verbose(1, " %s.%s.%s - hits %d of %d%s\n", db, jf->table, jf->field, hits, total,
		okFlag ? " ok" : js->isFuzzy ? " fuzzy" : "");
	if (hits < hitsNeeded && !js->isFuzzy)
	    {
	    warn("Error: %d of %d elements of %s.%s.%s are not in key %s.%s line %d of %s\n"
		 "Example miss: %s"
		, total - hits, total, db, jf->table, jf->field
		, keyField->table, keyField->field
		, jf->lineIx, joiner->fileName, miss);
	    }

	if (jf->unique || jf->full)
	    checkUniqueAndFull(joiner, js, db, jf, keyField, khiList);
	}
    freez(&miss);
    slFreeList(&tableList);
    sqlDisconnect(&conn);
    }
}

boolean dbInAnyField(struct joinerField *fieldList, char *db)
/* Return TRUE if database is used in any field. */
{
struct joinerField *field;
for (field = fieldList; field != NULL; field = field->next)
    {
    if (slNameInList(field->dbList, db))
        return TRUE;
    }
return FALSE;
}

void jsValidateKeysOnDb(struct joiner *joiner, struct joinerSet *js,
	char *db, struct hash *keyHash, struct keyHitInfo *khiList)
/* Validate keys pertaining to this database. */
{
struct joinerField *jf;
struct hash *localKeyHash = NULL;
struct keyHitInfo *localKhiList = NULL;
struct joinerField *keyField;

if ((keyField = js->fieldList) == NULL)
    return;
if (keyHash == NULL)
    {
    char *keyDb;
    if (slNameInList(keyField->dbList, db))
	{
	keyDb = db;
	}
    else if (!dbInAnyField(js->fieldList->next, db))
        {
	return;
	}
    else
	{
	if (slCount(keyField->dbList) == 1)
	    keyDb = keyField->dbList->name;
	else
	    {
	    struct slName *lastDb = slLastEl(keyField->dbList);
	    keyDb = lastDb->name;
	    verbose(1, " note - using %s database as reference for identifier %s\n",
	    	keyDb, js->name);
	    }
	}
    keyHash = localKeyHash = readKeyHash(keyDb, joiner, keyField, 
    	&localKhiList);
    khiList = localKhiList;
    }
if (keyHash != NULL)
    {
    for (jf = js->fieldList; jf != NULL; jf = jf->next)
	{
	if (jf != keyField && slNameInList(jf->dbList, db))
	    {
	    doKeyChecks(db, joiner, js, keyHash, khiList, keyField, jf);
	    }
	}
    }
hashFree(&localKeyHash);
slFreeList(&localKhiList);
}

void jsValidateKeys(struct joiner *joiner, struct joinerSet *js, 
	char *preferredDb)
/* Validate keys on js.  If preferredDb is non-NULL then do it on
 * that database.  Otherwise do it on all databases. */
{
struct joinerField *keyField;
struct hash *keyHash = NULL;
struct keyHitInfo *khiList = NULL;

/* If key is found in a single database then make hash here
 * rather than separately for each database. */
if ((keyField = js->fieldList) == NULL)
    return;
if (slCount(keyField->dbList) == 1)
    {
    if ((preferredDb == NULL) || (sameString(keyField->dbList->name, preferredDb)))
	keyHash = readKeyHash(keyField->dbList->name, joiner, keyField, &khiList);
    else
        return;
    }

/* Check key for database(s) */
if (preferredDb)
    jsValidateKeysOnDb(joiner, js, preferredDb, keyHash, khiList);
else
    {
    struct slName *db, *dbList = jsAllDb(js);
    for (db = dbList; db != NULL; db = db->next)
        jsValidateKeysOnDb(joiner, js, db->name, keyHash, khiList);
    slFreeList(&dbList);
    }
hashFree(&keyHash);
}

void joinerValidateKeys(struct joiner *joiner, 
	char *oneIdentifier, char *oneDatabase)
/* Validate all keys in joiner.  If oneDatabase is non-NULL then do it on
 * that database.  Otherwise do it on all databases. */
{
struct joinerSet *js;
int validations = 0;
verbose(1, "Checking keys on database %s\n", oneDatabase);
for (js = joiner->jsList; js != NULL; js = js->next)
    {
    verbose(2, "identifier %s\n", js->name);
    if (oneIdentifier == NULL || wildMatch(oneIdentifier, js->name))
	{
        jsValidateKeys(joiner, js, oneDatabase);
	++validations;
	}
    }
if (validations < 1 && oneIdentifier)
    errAbort("Identifier %s not found in %s", oneIdentifier, joiner->fileName);
}

static struct hash *joinerAllFields(struct joiner *joiner)
/* Get hash of all fields in database.table.field format.
 * Similar to sqlAllFields(void) function in jksql.c, but different
 *	method of creating the list of databases to check.
 */
{
struct hash *fullHash = hashNew(18);
struct hashEl *db, *dbList = hashElListHash(joiner->databasesChecked);
for (db = dbList; db != NULL; db = db->next)
    {
    verbose(3, "joinerAllFields: extracting fields from database: '%s'\n",
	db->name);
    sqlAddDatabaseFields(db->name, fullHash);
    }
slFreeList(&dbList);
return fullHash;
}

static struct hash *processFieldHash(struct joiner *joiner, char *inName,
    char *outName)
/* Read in field hash from file if inName is non-NULL, 
 * else read from database.  If outName is non-NULL, 
 * save it to file.  */
{
struct hash *fieldHash;

if (inName != NULL)
    fieldHash = hashWordsInFile(inName, 18);
else
    fieldHash = joinerAllFields(joiner);
if (outName != NULL)
    {
    struct hashEl *el, *list = hashElListHash(fieldHash);
    FILE *f = mustOpen(outName, "w");
    slSort(&list, hashElCmp);
    for (el = list; el != NULL; el = el->next)
	fprintf(f, "%s\n", el->name);
    slFreeList(&list);
    carefulClose(&f);
    }
return fieldHash;
}

void reportErrorList(struct slName **pList, char *message)
/* Report error on list of string references. */
/* Convert a list of string references to a comma-separated
 * slName.  Free up refList. */
{
if (*pList != NULL)
    {
    struct dyString *dy = dyStringNew(0);
    struct slName *name, *list;
    slReverse(pList);
    list = *pList;
    dyStringAppend(dy, list->name);
    for (name = list->next; name != NULL; name = name->next)
	{
        dyStringAppendC(dy, ',');
	dyStringAppend(dy, name->name);
	}
    warn("Error: %s %s", dy->string, message);
    slFreeList(pList);
    dyStringFree(&dy);
    }
}


static void joinerCheckDbCoverage(struct joiner *joiner)
/* Complain about databases that aren't in databasesChecked or ignored. */
{
struct slName *missList = NULL, *miss;
struct slName *db, *dbList = sqlListOfDatabases();

/* Keep a list of databases that aren't in either hash. */
for (db = dbList; db != NULL; db = db->next)
    {
    if (!hashLookup(joiner->databasesChecked, db->name) 
        && !hashLookup(joiner->databasesIgnored, db->name))
	{
	miss = slNameNew(db->name);
	slAddHead(&missList, miss);
	}
    }

/* If necessary report (in a single message) database not in joiner file. */
reportErrorList(&missList, "not in databasesChecked or databasesIgnored");
}

static void addTablesLike(struct hash *hash, struct sqlConnection *conn,
	char *spec)
/* Add tables like spec to hash. */
{
if (sqlWildcardIn(spec))
    {
    struct sqlResult *sr;
    char query[512], **row;
    safef(query, sizeof(query), 
	    "show tables like '%s'", spec);
    sr = sqlGetResult(conn, query);
    while ((row = sqlNextRow(sr)) != NULL)
	hashAdd(hash, row[0], NULL);
    sqlFreeResult(&sr);
    }
else
    {
    hashAdd(hash, spec, NULL);
    }
}


static struct hash *getCoveredTables(struct joiner *joiner, char *db, 
	struct sqlConnection *conn)
/* Get list of tables covered in database. */
{
struct hash *hash = hashNew(0);
struct joinerIgnore *ig;
struct slName *spec;
struct joinerSet *js;
struct joinerField *jf;

/* First put in all the ignored tables. */
for (ig = joiner->tablesIgnored; ig != NULL; ig = ig->next)
    {
    if (slNameInList(ig->dbList, db))
        {
	for (spec = ig->tableList; spec != NULL; spec = spec->next)
	    {
	    verbose(3,"ignoreTable: '%s'\n", spec->name);
	    addTablesLike(hash, conn, spec->name);
	    }
	}
    }

/* Now put in tables that are in one of the identifiers. */
for (js = joiner->jsList; js != NULL; js = js->next)
    {
    for (jf = js->fieldList; jf != NULL; jf = jf->next)
        {
	if (slNameInList(jf->dbList, db))
	    {
	    char spec[512];
	    safef(spec, sizeof(spec), "%s%s%s",
	        emptyForNull(jf->splitPrefix), jf->table,
		emptyForNull(jf->splitSuffix));
	    addTablesLike(hash, conn, spec);
	    verbose(4,"ident: '%s', table: '%s'\n", js->name, spec);
	    }
	}
    }
return hash;
}

void joinerCheckTableCoverage(struct joiner *joiner, char *specificDb)
/* Check that all tables either are part of an identifier or
 * are in the tablesIgnored statements. */
{
struct slName *miss, *missList = NULL;
struct hashEl *dbList, *db;

dbList = hashElListHash(joiner->databasesChecked);
for (db = dbList; db != NULL; db = db->next)
    {
    if (specificDb == NULL || sameString(db->name, specificDb))
	{
	struct sqlConnection *conn = sqlMayConnect(db->name);
	if (conn == NULL)
	    warn("Error: database %s doesn't exist", db->name);
	else
	    {
	    struct slName *table;
	    struct slName *tableList = sqlListTables(conn);
	    struct hash *hash = getCoveredTables(joiner, db->name, conn);
	    for (table = tableList; table != NULL; table = table->next)
		{
		if (!hashLookup(hash, table->name))
		    {
		    char fullName[256];
		    safef(fullName, sizeof(fullName), "%s.%s", 
			    db->name, table->name);
		    miss = slNameNew(fullName);
		    slAddHead(&missList, miss);
		    }
		else
		    verbose(2,"tableCovered: '%s'\n", table->name);
		}
	    slFreeList(&tableList);
	    freeHash(&hash);
	    reportErrorList(&missList, "tables not in .joiner file");
	    }
	sqlDisconnect(&conn);
	}
    }
slFreeList(&dbList);
}

void joinerCheck(char *fileName)
/* joinerCheck - Parse and check joiner file. */
{
struct joiner *joiner = joinerRead(fileName);
/* verify specified database is in all.joiner */
if (database)
    {
    if (hashLookup(joiner->databasesIgnored, database))
	errAbort("specified database '%s' is on list of databasesIgnored", database);
    if (!hashLookup(joiner->databasesChecked, database))
	errAbort("specified database '%s' is not listed in all.joiner", database);
    }
if (dbCoverage)
    joinerCheckDbCoverage(joiner);
if (tableCoverage)
    joinerCheckTableCoverage(joiner, database);
if (checkTimes)
    joinerCheckDependencies(joiner, database);
if (checkFields)
    {
    struct hash *fieldHash;
    fieldHash = processFieldHash(joiner, fieldListIn, fieldListOut);
    joinerValidateFields(joiner, fieldHash, identifier);
    }
if (foreignKeys)
    joinerValidateKeys(joiner, identifier, database);
}

int main(int argc, char *argv[])
/* Process command line. */
{
optionInit(&argc, argv, options);
if (argc != 2)
    usage();
fieldListIn = optionVal("fieldListIn", NULL);
fieldListOut = optionVal("fieldListOut", NULL);
identifier = optionVal("identifier", NULL);
database = optionVal("database", NULL);
checkFields = optionExists("fields");
foreignKeys = optionExists("keys");
dbCoverage = optionExists("dbCoverage");
tableCoverage = optionExists("tableCoverage");
checkTimes = optionExists("times");
if (optionExists("all"))
    {
    checkFields = foreignKeys = dbCoverage = tableCoverage = checkTimes = TRUE;
    }
allDbHash = sqlHashOfDatabases();
if (database)
    {
    if (! hashLookup(allDbHash, database))
	errAbort("specified database '%s' not available", database);
    }
if (! (checkFields|foreignKeys|dbCoverage|tableCoverage|checkTimes))
    {
    warn("*** Error: need to specify one of the options:\n"
	"\tfields, keys, tableCoverage, dbCoverage, times or all\n");
    usage();
    }
joinerCheck(argv[1]);
return 0;
}