Alter CustomQuery to use StreamingResult to download and process result sets:

 - Significantly improve memory usage
 - Minor speedup (1.1x faster?) to overall query/display times
 - Improvements to MCPStreamingResult and MCPConnection to accurately report affected row count

5 files changed, 109 insertions, 59 deletions

diff --git a/Frameworks/MCPKit/MCPFoundationKit/MCPConnection.m b/Frameworks/MCPKit/MCPFoundationKit/MCPConnection.m
index 4413e1c0..2463041a 100644
--- a/Frameworks/MCPKit/MCPFoundationKit/MCPConnection.m
+++ b/Frameworks/MCPKit/MCPFoundationKit/MCPConnection.m
@@ -1365,6 +1365,8 @@ static void forcePingTimeout(int signalNumber)
 		// On success, capture the results
 		if (0 == queryResultCode) {
 			
+			queryAffectedRows = mysql_affected_rows(mConnection);
+
 			if (mysql_field_count(mConnection) != 0) {
 				
 				// For normal result sets, fetch the results and unlock the connection
@@ -1391,10 +1393,8 @@ static void forcePingTimeout(int signalNumber)
 			
 			queryErrorMessage = [[NSString alloc] initWithString:@""];
 			queryErrorId = 0;
-			if (streamResultType == MCP_NO_STREAMING) {
+			if (streamResultType == MCP_NO_STREAMING && queryAffectedRows == -1) {
 				queryAffectedRows = mysql_affected_rows(mConnection);
-			} else {
-				queryAffectedRows = 0;
 			}
 			
 		// On failure, set the error messages and IDs
diff --git a/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.h b/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.h
index 146f81e7..38e59d1d 100644
--- a/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.h
+++ b/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.h
@@ -58,5 +58,6 @@ typedef struct SP_MYSQL_ROWS {
 
 // Results fetching
 - (NSArray *)fetchNextRowAsArray;
+- (void) cancelResultLoad;
 
 @end
\ No newline at end of file
diff --git a/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.m b/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.m
index 3f3e2073..c63a9f0b 100644
--- a/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.m
+++ b/Frameworks/MCPKit/MCPFoundationKit/MCPStreamingResult.m
@@ -285,13 +285,57 @@
 	return returnArray;
 }
 
+/*
+ * Ensure the result set is fully processed and freed without any processing
+ */
+- (void) cancelResultLoad
+{
+	MYSQL_ROW theRow;
+
+	// Loop through all the rows and ensure the rows are fetched.
+	// If fully streaming, loop through the rows directly
+	if (fullyStreaming) {
+		while (1) {
+			theRow = mysql_fetch_row(mResult);
+
+			// If no data was returned, we're at the end of the result set - return.
+			if (theRow == NULL) return;
+		}
+
+	// If in cached-streaming/fast download mode, loop until all data is fetched and freed
+	} else {
+
+		while (1) {
+		
+			// Check to see whether we need to wait for the data to be availabe
+			// - if so, wait 1ms before checking again
+			while (!dataDownloaded && processedRowCount == downloadedRowCount) usleep(1000);
+
+			// If all rows have been processed, we're at the end of the result set - return
+			// once all memory has been freed
+			if (processedRowCount == downloadedRowCount) {
+				while (!dataFreed) usleep(1000);
+				[parentConnection unlockConnection];
+				connectionUnlocked = YES;
+				return;
+			}
+			processedRowCount++;
+		}
+	}
+}
+
 #pragma mark -
 #pragma mark Overrides for safety
 
+/**
+ * If numOfRows is used before the data is fully downloaded, -1 will be returned;
+ * otherwise the number of rows is returned.
+ */
 - (my_ulonglong)numOfRows
 {
-	NSLog(@"numOfRows cannot be used with streaming results");
-	return 0;
+	if (!dataDownloaded) return -1;
+
+	return downloadedRowCount;
 }
 
 - (void)dataSeek:(my_ulonglong) row
diff --git a/Source/CustomQuery.h b/Source/CustomQuery.h
index 33596fc1..2987efed 100644
--- a/Source/CustomQuery.h
+++ b/Source/CustomQuery.h
@@ -137,7 +137,7 @@
 
 // Accessors
 - (NSArray *)currentResult;
-- (NSArray *)fetchResultAsArray:(MCPResult *)theResult;
+- (void)processResultIntoDataStorage:(MCPStreamingResult *)theResult;
 
 // MySQL Help
 - (NSString *)getHTMLformattedMySQLHelpFor:(NSString *)aString;
diff --git a/Source/CustomQuery.m b/Source/CustomQuery.m
index 81532735..daa09cf6 100644
--- a/Source/CustomQuery.m
+++ b/Source/CustomQuery.m
@@ -79,6 +79,7 @@
 	tableReloadAfterEditing = NO;
 
 	[self performQueries:queries];
+
 	// If no error was selected reconstruct a given selection
 	if([textView selectedRange].length == 0)
 		[textView setSelectedRange:curRange];
@@ -403,11 +404,11 @@
 - (void)performQueries:(NSArray *)queries;
 {	
 	
-	NSArray         *theColumns;
-	NSTableColumn   *theCol;
-	MCPResult     *theResult  = nil;
-	NSMutableArray  *menuItems  = [NSMutableArray array];
-	NSMutableString *errors     = [NSMutableString string];
+	NSArray				*theColumns;
+	NSTableColumn		*theCol;
+	MCPStreamingResult	*streamingResult  = nil;
+	NSMutableArray		*menuItems  = [NSMutableArray array];
+	NSMutableString		*errors     = [NSMutableString string];
 	
 	int i, totalQueriesRun = 0, totalAffectedRows = 0;
 	double executionTime = 0;
@@ -454,13 +455,22 @@
 		[tempQueries addObject:query];
 
 		// Run the query, timing execution (note this also includes network and overhead)
-		theResult = [mySQLConnection queryString:query];
+		streamingResult = [mySQLConnection streamingQueryString:query];
 		executionTime += [mySQLConnection lastQueryExecutionTime];
 		totalQueriesRun++;
 
+		// If this is the last query, retrieve and store the result; otherwise,
+		// discard the result without fully loading.
+		if (totalQueriesRun == queryCount)
+			[self processResultIntoDataStorage:streamingResult];
+		else
+			[streamingResult cancelResultLoad];
+
 		// Record any affected rows
 		if ( [mySQLConnection affectedRows] != -1 )
 			totalAffectedRows += [mySQLConnection affectedRows];
+		else if ( [streamingResult numOfRows] )
+			totalAffectedRows += [streamingResult numOfRows];
 
 		// Store any error messages
 		if ( ![[mySQLConnection getLastErrorMessage] isEqualToString:@""] ) {
@@ -536,9 +546,10 @@
 	
 	lastExecutedQuery = [[tempQueries lastObject] retain];
 	
-	//perform empty query if no query is given
+	// Perform empty query if no query is given
 	if ( !queryCount ) {
-		theResult = [mySQLConnection queryString:@""];
+		streamingResult = [mySQLConnection streamingQueryString:@""];
+		[streamingResult cancelResultLoad];
 		[errors setString:[mySQLConnection getLastErrorMessage]];
 	}
 	
@@ -640,9 +651,9 @@
 
 	// Restore automatic query retries
 	[mySQLConnection setAllowQueryRetries:YES];
-
+	
 	// If no results were returned, redraw the empty table and post notifications before returning.
-	if ( !theResult || ![theResult numOfRows] ) {
+	if ( ![fullResult count] ) {
 		[customQueryView reloadData];
 
 		// Notify any listeners that the query has completed
@@ -653,15 +664,12 @@
                                                        description:[NSString stringWithFormat:NSLocalizedString(@"%@",@"description for query finished growl notification"), [errorText stringValue]] 
                                                   notificationName:@"Query Finished"];
 
+		[streamingResult release];
 		return;
 	}
 
 	// get column definitions for the result array
-	cqColumnDefinition = [[theResult fetchResultFieldsStructure] retain];
-
-	// set datasource
-	[fullResult removeAllObjects];
-	[fullResult setArray:[self fetchResultAsArray:theResult]];
+	cqColumnDefinition = [[streamingResult fetchResultFieldsStructure] retain];
 
 	// Find result table name for copying as SQL INSERT.
 	// If more than one table name is found set resultTableName to nil.
@@ -678,10 +686,10 @@
 	}
 
 	// Add columns corresponding to the query result
-	theColumns = [theResult fetchFieldNames];
+	theColumns = [streamingResult fetchFieldNames];
 
 	if(!tableReloadAfterEditing) {
-		for ( i = 0 ; i < [theResult numOfFields] ; i++) {
+		for ( i = 0 ; i < [streamingResult numOfFields] ; i++) {
 			NSDictionary *columnDefinition = NSArrayObjectAtIndex(cqColumnDefinition,i);
 			theCol = [[NSTableColumn alloc] initWithIdentifier:[columnDefinition objectForKey:@"datacolumnindex"]];
 			[theCol setResizingMask:NSTableColumnUserResizingMask];
@@ -739,56 +747,53 @@
 }
 
 /*
- * Fetches the result as an array, with an array for each row in it
+ * Processes a supplied streaming result set, loading it into the data array.
  */
-- (NSArray *)fetchResultAsArray:(MCPResult *)theResult
+- (void)processResultIntoDataStorage:(MCPStreamingResult *)theResult
 {
-	// NSArray *columns;
-	unsigned long numOfRows = [theResult numOfRows];
-	NSMutableArray *tempResult = [NSMutableArray arrayWithCapacity:numOfRows];
-
 	NSArray *tempRow;
-	NSMutableArray *modifiedRow = [NSMutableArray array];
-	int i, j, numOfFields;
+	NSMutableArray *newRow;
+	int i;
+	long rowsProcessed = 0;
+	long columnsCount = 0;
+	NSAutoreleasePool *dataLoadingPool;
+	id prefsNullValue = [[prefs objectForKey:@"NullValue"] retain];
 	Class nullClass = [NSNull class];
-	id prefsNullValue = [prefs objectForKey:@"NullValue"];
-	// BOOL prefsLoadBlobsAsNeeded = [prefs boolForKey:@"LoadBlobsAsNeeded"];
 
-	// columns = [customQueryView columns];
-	// long columnsCount = [cqColumnDefinition count];
+	// Remove all items from the table
+	[fullResult removeAllObjects];
 
-	if (numOfRows) [theResult dataSeek:0];
-	for ( i = 0 ; i < numOfRows ; i++ ) {
-		tempRow = [theResult fetchRowAsArray];
+	// Set up an autorelease pool for row processing
+	dataLoadingPool = [[NSAutoreleasePool alloc] init];
 
-		if ( i == 0 ) numOfFields = [tempRow count];
+	// Loop through the result rows as they become available
+	while (tempRow = [theResult fetchNextRowAsArray]) {
+		if (columnsCount == 0) columnsCount = [tempRow count];
 
-		for ( j = 0; j < numOfFields; j++) {
-			if ( [NSArrayObjectAtIndex(tempRow, j) isMemberOfClass:nullClass] ) {
-				[modifiedRow addObject:prefsNullValue];
-			} else {
-				[modifiedRow addObject:NSArrayObjectAtIndex(tempRow, j)];
-			}
+		NSMutableArrayAddObject(fullResult, [NSMutableArray arrayWithArray:tempRow]);
+		newRow = NSArrayObjectAtIndex(fullResult, rowsProcessed);
+
+		// Process the retrieved row
+		for ( i = 0; i < columnsCount; i++ ) {
+			if ( [NSArrayObjectAtIndex(tempRow, i) isMemberOfClass:nullClass] )
+				[newRow replaceObjectAtIndex:i withObject:prefsNullValue];
 		}
 
-		// Add values for hidden blob and text fields if appropriate
-		// if ( prefsLoadBlobsAsNeeded ) {
-		// 	for ( j = 0 ; j < columnsCount ; j++ ) {
-		// 		if ( [[NSArrayObjectAtIndex(cqColumnDefinition, j) objectForKey:@"typegrouping"] isEqualToString:@"blobdata"] ||
-		// 		 	[[NSArrayObjectAtIndex(cqColumnDefinition, j) objectForKey:@"typegrouping"] isEqualToString:@"textdata"]) {
-		// 			[modifiedRow setObject:NSLocalizedString(@"(not loaded)", @"value shown for hidden blob and text fields") forKey:[NSArrayObjectAtIndex(cqColumnDefinition, j) objectForKey:@"name"]];
-		// 		}
-		// 	}
-		// }
+		// Update the count of rows processed
+		rowsProcessed++;
 
-		[tempResult addObject:[NSArray arrayWithArray:modifiedRow]];
-		[modifiedRow removeAllObjects];
+		// Drain and reset the autorelease pool every ~1024 rows
+		if (!(rowsProcessed % 1024)) {
+			[dataLoadingPool drain];
+			dataLoadingPool = [[NSAutoreleasePool alloc] init];
+		}
 	}
-
-	return tempResult;
+	
+	// Clean up the autorelease pool
+	[dataLoadingPool drain];
+	[prefsNullValue release];
 }
 
-
 /*
  * Retrieve the range of the query at a position specified 
  * within the custom query text view.