sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/dispatch/FdcDispatcher.java
private void executeFdc(AutoGeneratedQueryTypes qt, SparqlResultTypes rt, String targetSparqlID) {
		String errorHeader = "";

		try {
	
			long startMsec = System.currentTimeMillis();
			
			SparqlEndpointInterface tempDataSei = this.createTempSei();
			SparqlConnection ingestConn = this.createIngestConn(tempDataSei);
			SparqlConnection expandedConn = this.createExpandedConn(tempDataSei);
			
			int statusIncrement = 99 / (4 * this.fdcServiceManager.getFdcNodeCount() + 1);
			int percentComplete = 1;
			
			// loop through FDC nodes
			// TODO: this should be smarter and manage dependencies/order 
			HashMap<String, Table> paramSetTables = new HashMap<String, Table>();
			while (this.fdcServiceManager.nextFdcNode()) {
				Node node = this.fdcServiceManager.getCurrentFdcNode();
				String fdcName = node.getSparqlID();
				errorHeader = "setting up FDC node " + fdcName;
				
				// get and execute each param nodegroup
				// TODO this could break queries into LIMIT/OFFSET chunks
				// TODO this might be async
				
				this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - getting inputs " );
				
				HashMap<String, NodeGroup> paramSetNodeGroups = this.fdcServiceManager.getParamNodeGroups(node);
				for (String paramSet : paramSetNodeGroups.keySet()) {
					errorHeader = "querying " + fdcName + " param set " + paramSet;
					NodeGroup ng = paramSetNodeGroups.get(paramSet);
					
					if (ng == null) {
						// if nodegroup doesn't have necessary subgraph to generate param
						paramSetTables.put(paramSet, null);
						break;
						
					} else {
						// if param can be calculated with nodegroup, do so
						ng.setSparqlConnection(expandedConn);
						String query = ng.generateSparqlSelect();
						paramSetTables.put(paramSet, this.querySei.executeQueryToTable(query));
					}
				}
				
				percentComplete += statusIncrement;   // 1 of FDC - finished retrieval
				this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - running" );
				
				// log inputs
				LocalLogger.logToStdErr("FDC call: " + this.fdcServiceManager.getCurrentServiceUrl());
				//for (Table t : paramSetTables.values()) {
				//	LocalLogger.logToStdErr("Input table: \n" + (t == null ? "null" : t.toCSVString()));
				//}
				
				// if nodegroup had subgraphs available to generate all parameter tables
				// run the FDC service
				if (! paramSetTables.containsValue(null)) {
					errorHeader = "running fdc service " + this.fdcServiceManager.getCurrentServiceUrl();
					
					// execute FDC call
					FdcClient fdcClient = new FdcClient(FdcClientConfig.fromFullEndpoint(this.fdcServiceManager.getCurrentServiceUrl(), paramSetTables));
					TableResultSet fdcResultSet = fdcClient.executeWithTableResultReturn();
					fdcResultSet.throwExceptionIfUnsuccessful("Error making FDC call");
					Table fdcResults = fdcResultSet.getTable();
					String fdcCsv = fdcResults.toCSVString();
					
					// log outputs
					LocalLogger.logToStdOut("FDC results returned " + String.valueOf(fdcResults.getNumRows()) + " rows");
					
					// ingest
					percentComplete += statusIncrement; // 2 of FDC - finished running FDC client
					this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - ingesting " + String.valueOf(fdcResults.getNumRows()) + " rows of results" );
					
					String ingestId = this.fdcServiceManager.getCurrentIngestNodegroupId();
					errorHeader = "ingesting fdc results using " + ingestId;
					
					// try retrieving nodegroup from fdcClient first, failures will be silent
					FdcClient fdcGetNgClient = new FdcClient(FdcClientConfig.buildGetNodegroup(this.fdcServiceManager.getCurrentServiceUrl(), ingestId));
					SparqlGraphJson sgJson = fdcGetNgClient.executeGetNodegroup();
					if (sgJson == null) {
						// try nodegroup store if not found yet.  This one will throw an exception on failure.
						sgJson = this.ngStoreClient.executeGetNodeGroupByIdToSGJson(ingestId);
					}
					
					sgJson.setSparqlConn(ingestConn);
					boolean precheck = false;
					DataLoader.loadFromCsvString(sgJson.getJson(), fdcCsv, this.tmpGraphUser, this.tmpGraphPassword, precheck);	
				}
				percentComplete += 2 * statusIncrement; // 3 and 4 of FDC - finished ingestion
			}
			
			errorHeader = "running final sparql query";

			// run regular query
			this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: running final query" );
			
			this.queryNodeGroup.setSparqlConnection(expandedConn);
			
			String query = this.getSparqlQuery(qt, targetSparqlID);
			this.executePlainSparqlQuery(query, rt);
			
			//--- At this point results have been sent ---
						
			// log milliseconds
			long endMsec = System.currentTimeMillis();
			LocalLogger.logToStdErr("FDC query milliseconds: " + String.valueOf(endMsec - startMsec));
			
			// delete temp graph
			tempDataSei.dropGraph();
			
			
		} catch (Exception e) {
			LocalLogger.printStackTrace(e);
			try {
				this.updateStatusToFailed("Internal FDC error " + errorHeader + ": " + e.getMessage());
			} catch (Exception ee) {}
			
		} finally {
			try {
				FdcServiceManager.suggestReCache(this.extConfigSei, this.oInfoClient);
			} catch (Exception e) {
				LocalLogger.printStackTrace(e);
			}
		}
	}
/**
	 * Execute query for known FDC nodegroup
	 */
Execute query for known FDC nodegroup
Execute query for known FDC nodegroup
private void executeFdc(AutoGeneratedQueryTypes qt, SparqlResultTypes rt, String targetSparqlID) {
		String errorHeader = "";
		try {
			long startMsec = System.currentTimeMillis();
			SparqlEndpointInterface tempDataSei = this.createTempSei();
			SparqlConnection ingestConn = this.createIngestConn(tempDataSei);
			SparqlConnection expandedConn = this.createExpandedConn(tempDataSei);
			int statusIncrement = 99 / (4 * this.fdcServiceManager.getFdcNodeCount() + 1);
			int percentComplete = 1;
			HashMap<String, Table> paramSetTables = new HashMap<String, Table>();
			while (this.fdcServiceManager.nextFdcNode()) {
				Node node = this.fdcServiceManager.getCurrentFdcNode();
				String fdcName = node.getSparqlID();
				errorHeader = "setting up FDC node " + fdcName;
				this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - getting inputs " );
				HashMap<String, NodeGroup> paramSetNodeGroups = this.fdcServiceManager.getParamNodeGroups(node);
				for (String paramSet : paramSetNodeGroups.keySet()) {
					errorHeader = "querying " + fdcName + " param set " + paramSet;
					NodeGroup ng = paramSetNodeGroups.get(paramSet);
					if (ng == null) {
						paramSetTables.put(paramSet, null);
						break;
					} else {
						ng.setSparqlConnection(expandedConn);
						String query = ng.generateSparqlSelect();
						paramSetTables.put(paramSet, this.querySei.executeQueryToTable(query));
					}
				}
				percentComplete += statusIncrement;   
				this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - running" );
				LocalLogger.logToStdErr("FDC call: " + this.fdcServiceManager.getCurrentServiceUrl());
				if (! paramSetTables.containsValue(null)) {
					errorHeader = "running fdc service " + this.fdcServiceManager.getCurrentServiceUrl();
					FdcClient fdcClient = new FdcClient(FdcClientConfig.fromFullEndpoint(this.fdcServiceManager.getCurrentServiceUrl(), paramSetTables));
					TableResultSet fdcResultSet = fdcClient.executeWithTableResultReturn();
					fdcResultSet.throwExceptionIfUnsuccessful("Error making FDC call");
					Table fdcResults = fdcResultSet.getTable();
					String fdcCsv = fdcResults.toCSVString();
					LocalLogger.logToStdOut("FDC results returned " + String.valueOf(fdcResults.getNumRows()) + " rows");
					percentComplete += statusIncrement; 
					this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: " + fdcName + " - ingesting " + String.valueOf(fdcResults.getNumRows()) + " rows of results" );
					String ingestId = this.fdcServiceManager.getCurrentIngestNodegroupId();
					errorHeader = "ingesting fdc results using " + ingestId;
					FdcClient fdcGetNgClient = new FdcClient(FdcClientConfig.buildGetNodegroup(this.fdcServiceManager.getCurrentServiceUrl(), ingestId));
					SparqlGraphJson sgJson = fdcGetNgClient.executeGetNodegroup();
					if (sgJson == null) {
						sgJson = this.ngStoreClient.executeGetNodeGroupByIdToSGJson(ingestId);
					}
					sgJson.setSparqlConn(ingestConn);
					boolean precheck = false;
					DataLoader.loadFromCsvString(sgJson.getJson(), fdcCsv, this.tmpGraphUser, this.tmpGraphPassword, precheck);	
				}
				percentComplete += 2 * statusIncrement; 
			}
			errorHeader = "running final sparql query";
			this.jobTracker.setJobPercentComplete(this.jobID, percentComplete, "FDC Query: running final query" );
			this.queryNodeGroup.setSparqlConnection(expandedConn);
			String query = this.getSparqlQuery(qt, targetSparqlID);
			this.executePlainSparqlQuery(query, rt);
			long endMsec = System.currentTimeMillis();
			LocalLogger.logToStdErr("FDC query milliseconds: " + String.valueOf(endMsec - startMsec));
			tempDataSei.dropGraph();
		} catch (Exception e) {
			LocalLogger.printStackTrace(e);
			try {
				this.updateStatusToFailed("Internal FDC error " + errorHeader + ": " + e.getMessage());
			} catch (Exception ee) {}
		} finally {
			try {
				FdcServiceManager.suggestReCache(this.extConfigSei, this.oInfoClient);
			} catch (Exception e) {
				LocalLogger.printStackTrace(e);
			}
		}
	}
private void executeFdc ( AutoGeneratedQueryTypes qt , SparqlResultTypes rt , String targetSparqlID ) { String errorHeader = "" ; try { long startMsec = System . currentTimeMillis ( ) ; SparqlEndpointInterface tempDataSei = this . createTempSei ( ) ; SparqlConnection ingestConn = this . createIngestConn ( tempDataSei ) ; SparqlConnection expandedConn = this . createExpandedConn ( tempDataSei ) ; int statusIncrement = 99 / ( 4 * this . fdcServiceManager . getFdcNodeCount ( ) + 1 ) ; int percentComplete = 1 ; HashMap < String , Table > paramSetTables = new HashMap < String , Table > ( ) ; while ( this . fdcServiceManager . nextFdcNode ( ) ) { Node node = this . fdcServiceManager . getCurrentFdcNode ( ) ; String fdcName = node . getSparqlID ( ) ; errorHeader = "setting up FDC node " + fdcName ; this . jobTracker . setJobPercentComplete ( this . jobID , percentComplete , "FDC Query: " + fdcName + " - getting inputs " ) ; HashMap < String , NodeGroup > paramSetNodeGroups = this . fdcServiceManager . getParamNodeGroups ( node ) ; for ( String paramSet : paramSetNodeGroups . keySet ( ) ) { errorHeader = "querying " + fdcName + " param set " + paramSet ; NodeGroup ng = paramSetNodeGroups . get ( paramSet ) ; if ( ng == null ) { paramSetTables . put ( paramSet , null ) ; break ; } else { ng . setSparqlConnection ( expandedConn ) ; String query = ng . generateSparqlSelect ( ) ; paramSetTables . put ( paramSet , this . querySei . executeQueryToTable ( query ) ) ; } } percentComplete += statusIncrement ; this . jobTracker . setJobPercentComplete ( this . jobID , percentComplete , "FDC Query: " + fdcName + " - running" ) ; LocalLogger . logToStdErr ( "FDC call: " + this . fdcServiceManager . getCurrentServiceUrl ( ) ) ; if ( ! paramSetTables . containsValue ( null ) ) { errorHeader = "running fdc service " + this . fdcServiceManager . getCurrentServiceUrl ( ) ; FdcClient fdcClient = new FdcClient ( FdcClientConfig . fromFullEndpoint ( this . fdcServiceManager . getCurrentServiceUrl ( ) , paramSetTables ) ) ; TableResultSet fdcResultSet = fdcClient . executeWithTableResultReturn ( ) ; fdcResultSet . throwExceptionIfUnsuccessful ( "Error making FDC call" ) ; Table fdcResults = fdcResultSet . getTable ( ) ; String fdcCsv = fdcResults . toCSVString ( ) ; LocalLogger . logToStdOut ( "FDC results returned " + String . valueOf ( fdcResults . getNumRows ( ) ) + " rows" ) ; percentComplete += statusIncrement ; this . jobTracker . setJobPercentComplete ( this . jobID , percentComplete , "FDC Query: " + fdcName + " - ingesting " + String . valueOf ( fdcResults . getNumRows ( ) ) + " rows of results" ) ; String ingestId = this . fdcServiceManager . getCurrentIngestNodegroupId ( ) ; errorHeader = "ingesting fdc results using " + ingestId ; FdcClient fdcGetNgClient = new FdcClient ( FdcClientConfig . buildGetNodegroup ( this . fdcServiceManager . getCurrentServiceUrl ( ) , ingestId ) ) ; SparqlGraphJson sgJson = fdcGetNgClient . executeGetNodegroup ( ) ; if ( sgJson == null ) { sgJson = this . ngStoreClient . executeGetNodeGroupByIdToSGJson ( ingestId ) ; } sgJson . setSparqlConn ( ingestConn ) ; boolean precheck = false ; DataLoader . loadFromCsvString ( sgJson . getJson ( ) , fdcCsv , this . tmpGraphUser , this . tmpGraphPassword , precheck ) ; } percentComplete += 2 * statusIncrement ; } errorHeader = "running final sparql query" ; this . jobTracker . setJobPercentComplete ( this . jobID , percentComplete , "FDC Query: running final query" ) ; this . queryNodeGroup . setSparqlConnection ( expandedConn ) ; String query = this . getSparqlQuery ( qt , targetSparqlID ) ; this . executePlainSparqlQuery ( query , rt ) ; long endMsec = System . currentTimeMillis ( ) ; LocalLogger . logToStdErr ( "FDC query milliseconds: " + String . valueOf ( endMsec - startMsec ) ) ; tempDataSei . dropGraph ( ) ; } catch ( Exception e ) { LocalLogger . printStackTrace ( e ) ; try { this . updateStatusToFailed ( "Internal FDC error " + errorHeader + ": " + e . getMessage ( ) ) ; } catch ( Exception ee ) { } } finally { try { FdcServiceManager . suggestReCache ( this . extConfigSei , this . oInfoClient ) ; } catch ( Exception e ) { LocalLogger . printStackTrace ( e ) ; } } }
Execute query for known FDC nodegroup
Execute query for known FDC nodegroup
// loop through FDC nodes // TODO: this should be smarter and manage dependencies/order  // get and execute each param nodegroup // TODO this could break queries into LIMIT/OFFSET chunks // TODO this might be async // if nodegroup doesn't have necessary subgraph to generate param // if param can be calculated with nodegroup, do so // 1 of FDC - finished retrieval // log inputs //for (Table t : paramSetTables.values()) { //	LocalLogger.logToStdErr("Input table: \n" + (t == null ? "null" : t.toCSVString())); //} // if nodegroup had subgraphs available to generate all parameter tables // run the FDC service // execute FDC call // log outputs // ingest // 2 of FDC - finished running FDC client // try retrieving nodegroup from fdcClient first, failures will be silent // try nodegroup store if not found yet.  This one will throw an exception on failure. // 3 and 4 of FDC - finished ingestion // run regular query //--- At this point results have been sent --- // log milliseconds // delete temp graph