<commit_before>'use strict';

// 
// Base class for data frame classes.
//

var ArrayIterator = require('./iterators/array');
var MultiIterator = require('./iterators/multi');
var PairIterator = require('./iterators/pair');
var SkipIterator = require('./iterators/skip');
var SkipWhileIterator = require('./iterators/skip-while');
var BabyParse = require('babyparse');
var SelectIterator = require('../src/iterators/select');
var SelectManyIterator = require('../src/iterators/select-many');
var TakeIterator = require('../src/iterators/take');
var TakeWhileIterator = require('../src/iterators/take-while');
var WhereIterator = require('../src/iterators/where');
var CountIterator = require('../src/iterators/count');
var EmptyIterator = require('../src/iterators/empty');
var utils = require('./utils');
var extend = require('extend');
var inherit = require('./inherit');

var assert = require('chai').assert; 
var E = require('linq');

var validateIterator = require('./iterators/validate');

//
// Creates an iterator that converts rows to JavaScript objects based on passed in column names.
//
var convertRowsToObjects = function (columnNames, rowsIterator) {

	if (Object.isFunction(columnNames)) {
		columnNames = columnNames();
	}

	assert.isArray(columnNames);

	validateIterator(rowsIterator);

	return new SelectIterator(
		rowsIterator,
		function (row) {
			return E.from(columnNames)
				.select(function (columnName, columnIndex) {
					return [columnName, columnIndex];
				})
				.toObject(
					function (column) {
						return column[0];
					},
					function (column) {
						return row[column[1]];
					}
				);							
		}
	);
};

//
// Determine column names from an array of rows. Column names are take from the fields in each JavaScript object.
//
var determineColumnNamesFromObjectRows = function (rows, considerAllRows)  {

	assert.isArray(rows);

	if (considerAllRows) {
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		if (rows.length > 0) {
			// Just consider the first row.
			return Object.keys(rows[0]);
		}
		else {
			// Can't do this, there are no rows.
			return [];
		}
	}
};

//
// Evaluate an iterable of JavaScript objects, look at the fields and figure out column names from that.
//
var determineColumnNamesFromObjectsIterable = function (iterable, considerAllRows) {

	assert.isFunction(iterable);

	var iterator = iterable();

	if (considerAllRows) {
		// Consider all rows, this expensive, so it is optional.
		var rows = [];
		while (iterator.moveNext()) {
			rows.push(iterator.getCurrent());
		}
		
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		// Just consider the first row.		
		if (!iterator.moveNext()) {
			return []; // Nothing in the iterable.
		}

		return Object.keys(iterator.getCurrent());
	}
};

//
// Evaluate an iterable of index/value pairs, look at the fields in the vlaues and figure out column names from that.
//
var determineColumnNamesFromPairsIterable = function (iterable, considerAllRows) {

	assert.isFunction(iterable);

	var iterator = iterable();

	if (considerAllRows) {
		// Consider all rows, this expensive, so it is optional.
		var rows = [];
		while (iterator.moveNext()) {
			rows.push(iterator.getCurrent()[1]);
		}
		
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		// Just consider the first row.
		if (!iterator.moveNext()) {
			return []; // Nothing in the iterable.
		}

		return Object.keys(iterator.getCurrent()[1]);
	}
};

//
// Constuctor.
//
var DataFrame = function (config) {

	var self = this;

	if (!config) {
		self._columnNames = [];
		self.getIterator = function () {
			return new EmptyIterator();
		};
		return;
	}

	if (config.iterable) {

		assert.isFunction(config.iterable, "Expected 'iterable' field of 'config' parameter to DataFrame constructor to be a function that returns an index/value pairs iterator.");

		var iterable = config.iterable;

		if (config.columnNames) {
			assert.isArray(config.columnNames, "Expected 'columnNames' field of 'config' parameter to DataFrame constructor to be an array of column names or function that returns an array of column names.");

			self._columnNames = config.columnNames;
		}
		else {
			self._columnNames = determineColumnNamesFromPairsIterable(iterable, config.considerAllRows);
		}
		self.getIterator = iterable;
		return;
	}

	if (!config.rows &&  !config.columns) {
		self._columnNames = config.columnNames || [];
		self.getIterator = function () {
			return new EmptyIterator();
		};
		return;
	}

	assert.isObject(config, "Expected 'config' parameter to DataFrame constructor to be an object with options for initialisation.");

	if (config.index) {
		if (!Object.isArray(config.index)) {
			assert.isObject(config.index, "Expected 'index' member of 'config' parameter to DataFrame constructor to be an object.");			
		}
	}

	if (config.columnNames) {
		assert.isArray(config.columnNames, "Expected 'columnNames' member of 'config' parameter to DataFrame constructor to be an array of strings or a function that produces an array of strings.");

		config.columnNames.forEach(function (columnName) {
			assert.isString(columnName, "Expected 'columnNames' member of 'config' parameter to DataFrame constructor to be an array of strings or a function that produces an array of strings.");
		});

		if (!config.rows) {
			throw new Error("Expected to find a 'rows' member of 'config' parameter to DataFrame constructor.");
		}

	 	if (!Object.isFunction(config.rows)) {
			assert.isArray(config.rows, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of rows.");

			if (config.debug) {
				config.rows.forEach(function (row) {
					assert.isArray(row, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of arrays, an array of objects or an iterator.");
				});
			}
		}
	}
	else if (config.rows) {
		assert(!config.columns, "Can't use both 'rows' and 'columns' fields of 'config' parameter to DataFrame constructor.");

		if (!Object.isFunction(config.rows)) {
			assert.isArray(config.rows, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of JavaScript objects.");
			
			if (config.rows.length > 0) {
				assert.isObject(config.rows[0], "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of JavaScript objects.")
			}
		}
	}
	else if (config.columns) {
		assert.isObject(config.columns, "Expected 'columns' member of 'config' parameter to DataFrame constructor to be an object with fields that define columns.");
	}

	var rows = config.rows;
	var columns = config.columns;

	if (config.columnNames)	{
		self._columnNames = config.columnNames;

		if (config.index) {
			var index = config.index;
			if (Object.isArray(index)) {

				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, rows()));
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, new ArrayIterator(rows)));
					};
				}
			}
			else {
				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								} 
							),	
							convertRowsToObjects(self._columnNames, rows())
						);
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								}
							),
							convertRowsToObjects(self._columnNames, new ArrayIterator(rows))
						);
					};				
				}
			}
		}
		else {
			if (Object.isFunction(rows)) {
				this.getIterator = function () {
					return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, rows()));
				};
			}
			else {
				this.getIterator = function () {
					return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, new ArrayIterator(rows)));
				};
			}
		}	
	}
	else {
		if (rows) {
			if (Object.isFunction(rows)) {
				self._columnNames = determineColumnNamesFromObjectsIterable(rows, config.considerAllRows);
			}
			else {
				// Derive column names from object fields.
				self._columnNames = determineColumnNamesFromObjectRows(rows, config.considerAllRows);
			}
		}
		else {
			self._columnNames = Object.keys(columns);
		}

		if (config.index) {
			var index = config.index;
			if (Object.isArray(index)) {

				if (rows) {
					if (Object.isFunction(rows)) {
						this.getIterator = function () {
							return new PairIterator(new ArrayIterator(index), rows());
						};
					}
					else {
						this.getIterator = function () {
							return new PairIterator(new ArrayIterator(index), new ArrayIterator(rows));
						};
					}
				}
				else {
					this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)));
					};
				}
			}
			else {
				if (rows) {
					if (Object.isFunction(rows)) {
						this.getIterator = function () {
							return new PairIterator(
								new SelectIterator(
									index.getIterator(),
									function (pair) {
										return pair[1];
									}	
								), 
								rows()
							);
						};
					}
					else {
						this.getIterator = function () {
							return new PairIterator(
								new SelectIterator(
									index.getIterator(),
									function (pair) {
										return pair[1];
									}
								), 
								new ArrayIterator(rows)
							);
						};				
					}
				}
				else {
					this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								}
							), 
							convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)))
						;
					};
				}
			}
		}
		else {
			if (rows) {
				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(new CountIterator(), rows());
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(new CountIterator(), new ArrayIterator(rows));
					};
				}
			}
			else {
				this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)));
				};
			}
		}
	}
};

module.exports = DataFrame;

var Series = require('./series');
var parent = inherit(DataFrame, Series);

var concatDataFrames = require('./concat-dataframes');
var zipDataFrames = require('./zip-dataframes');
var mergeSeries = require('./merge-series');
var mergeDataFrames = require('./merge-dataframes');

/**
 * Get the index of the data frame.
 */
DataFrame.prototype.getIndex = function () {
	var self = this;
	return new Series({
		iterable: function () {		
			return new SelectIterator(
				self.getIterator(),
				function (pair, index) {
					return [index, pair[0]]; // Extract index.
				}
			);
		},
	});
};

/**
 * Get the names of the columns in the data frame.
 */
DataFrame.prototype.getColumnNames = function () {
	var self = this;
	return self._columnNames;
};

/**
 * Get an iterator for the data-frame.
 */
DataFrame.prototype.getIterator = function () {
	return new EmptyIterator(); // This function is defined by the constructor.
};

/**
 * Gets a column index from a column name.
 *
 * @param {string} columnName - The name of the column to retrieve the column index for.
 *
 * @returns {Number} Returns the index of the named column or -1 if the requested column was not found.
 */
DataFrame.prototype.getColumnIndex = function (columnName) {
	assert.isString(columnName, "Expected 'columnName' parameter to getColumnIndex to be a non-empty string.");
	
	var self = this;	
	var columnNames = self.getColumnNames();
	
	for (var i = 0; i < columnNames.length; ++i) {
		if (columnName === columnNames[i]) {
			return i;
		}
	}	
	
	return -1;
};

/**
 * Gets a column name from a column index.
 *
 * @param {int} columnIndex - The index of the column to retrieve the column name for.
 *
 * @returns {string} Returns the name of the column or undefined if the requested column was not found.
 */
DataFrame.prototype.getColumnName = function (columnIndex) {
	assert.isNumber(columnIndex, "Expected 'columnIndex' parameter to getColumnIndex to be a non-empty string.");

	var self = this;	
	var columnNames = self.getColumnNames();

	if (columnIndex < 0 || columnIndex >= columnNames.length) {
		return undefined;
	}

	return columnNames[columnIndex];
};

/**
 * Skip a number of rows in the data frame.
 *
 * @param {int} numRows - Number of rows to skip.
 */
DataFrame.prototype.skip = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'skip' function to be a number.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SkipIterator(self.getIterator(), numRows);
		},
	}); 	
};

/**
 * Skips rows in the data-frame while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.skipWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SkipWhileIterator(self.getIterator(), function (pair) {
					return predicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Skips rows in the data-frame until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.skipUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.skipWhile(function (value, index) { 
		return !predicate(value, index); 
	});
};

/**
 * Take a number of rows in the data frame.
 *
 * @param {int} numRows - Number of rows to take.
 */
DataFrame.prototype.take = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'take' function to be a number.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeIterator(self.getIterator(), numRows);
		},
	}); 	
};

/**
 * Take rows from the data-frame while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.takeWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeWhileIterator(self.getIterator(), function (pair) {
					return predicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Take rows from the data-frame until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.takeUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.takeWhile(function (value, index) { 
		return !predicate(value, index); 
	});
};

/**
 * Filter a data frame by a predicate selector.
 *
 * @param {function} filterSelectorPredicate - Predicte function to filter rows of the data frame.
 */
DataFrame.prototype.where = function (filterSelectorPredicate) {
	assert.isFunction(filterSelectorPredicate, "Expected 'filterSelectorPredicate' parameter to 'where' function to be a function.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new WhereIterator(self.getIterator(), function (pair) {
					return filterSelectorPredicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each row to generate a transformed data-frame.
 */
DataFrame.prototype.select = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.select' to be a selector functions.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectIterator(self.getIterator(), function (pair) {
					var newValue = selector(pair[1], pair[0]);
					if (!Object.isObject(newValue)) {
						throw new Error("Expected return value from 'select' selector to be an object that represents a new row in the resulting data-frame.");
					};
					return [
						pair[0],
						newValue,
					];
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index and row pair to generate a transformed data-frame.
 */
DataFrame.prototype.selectPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectPairs' to be a selector functions.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectIterator(self.getIterator(), function (pair) {
					var newPair = selector(pair[1], pair[0]);
					if (!Object.isArray(newPair) || newPair.length !== 2 || !Object.isObject(newPair[1])) {
						throw new Error("Expected return value from 'DataFrame.selectPairs' selector to be a pair, that is an array with two items: [index, object].");
					}
					return newPair;
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each row to create a new data-frame.
 */
DataFrame.prototype.selectMany = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectMany' function to be a function.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), function (pair) {
				var newRows = selector(pair[1], pair[0]);

				if (!Object.isArray(newRows) &&
					!(newRows instanceof Series) &&
					!(newRows instanceof DataFrame)) {
					throw new Error("Expected return value from 'DataFrame.selectMany' selector to be an array of objects, a Series or a DataFrame, each object represents a new row in the resulting data-frame.");
				}

				if (newRows instanceof DataFrame) {
					newRows = newRows.toObjects();
				}
				else if (newRows instanceof Series)
				{
					newRows = newRows.toValues();
				}

				var newPairs = [];
				for (var newRowIndex = 0; newRowIndex < newRows.length; ++newRowIndex) {
					var newRow = newRows[newRowIndex];
					if (!Object.isObject(newRow)) {
						throw new Error("Expected array returned from 'DataFrame.selectMany' selector to contain only objects, each object represents a new row in the resulting data-frame.");
					};

					newPairs.push([
						pair[0], 
						newRow,
					]);
				}

				return newPairs;
			})
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index/row pair to create a new data-frame.
 */
DataFrame.prototype.selectManyPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectManyPairs' function to be a function.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), function (pair) {
				var newPairs = selector(pair[1], pair[0]);
				if (!Object.isArray(newPairs)) {
					throw new Error("Expected return value from 'DataFrame.selectManyPairs' selector to be an array of pairs, each item in the array represents a new pair in the resulting DataFrame.");
				}

				for (var pairIndex = 0; pairIndex < newPairs.length; ++pairIndex) {
					var newPair = newPairs[pairIndex];
					if (!Object.isArray(newPair) || newPair.length !== 2 || !Object.isObject(newPair[1])) {
						throw new Error("Expected return value from 'DataFrame.selectManyPairs' selector to be am array of pairs, but item at index " + pairIndex + " is not an array with two items: [index, object].");
					}
				}

				return newPairs;
			})
		},
	}); 	
};

/**
 * Retreive a time-series from a column of the data-frame.
 *
 * @param {string} columnName - Specifies the column to retreive.
 */
DataFrame.prototype.getSeries = function (columnName) {
	var self = this;

	assert.isString(columnName, "Expected 'columnName' parameter to getSeries function to be a string that specifies the name of the column to retreive.");

	return new Series({
		iterable: function () {
			return new WhereIterator(
				new SelectIterator(self.getIterator(), function (pair) {
					return [
						pair[0],
						pair[1][columnName],
					];
				}),
				function (pair) {
					return pair[1] !== undefined;
				}
			);
		},
	});
};

/**
 * Returns true if the column with the requested name exists in the data frame.
 *
 * @param {string} columnName - Name of the column to check.
 */
DataFrame.prototype.hasSeries = function (columnName) {

	assert.isString(columnName);

	var self = this;
	return self.getColumnIndex(columnName) >= 0;
};

/**
 * 
 * Verify the existance of a column and return it.
 * Throws an exception if the column doesn't exist.
 *
 * @param {string} columnName - Name or index of the column to retreive.
 */
DataFrame.prototype.expectSeries = function (columnName) {

	var self = this;
	if (self.getColumnNames().indexOf(columnName) < 0) {
		throw new Error("Expected data-frame to contain series with column name: '" + columnName + "'.");
	}
	return self;
};

/** 
 * Retreive a collection of all columns.
 */
DataFrame.prototype.getColumns = function () {

	var self = this;

	return E.from(self.getColumnNames())
		.select(function (columnName) {
			return {
				name: columnName,
				series: self.getSeries(columnName),
			};
		})
		.toArray();
};

/**
 * Create a new data-frame from a subset of columns.
 *
 * @param {array} columnNames - Array of column names to include in the new data-frame.
 */
DataFrame.prototype.subset = function (columnNames) {

	var self = this;
	
	assert.isArray(columnNames, "Expected 'columnNames' parameter to 'subset' to be an array.");	
	
	return new DataFrame({
		columnNames: columnNames,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					return [
						pair[0],
						E.from(columnNames)
							.toObject(
								function (columnName) {
									return columnName;
								},
								function (columnName) {
									return pair[1][columnName];
								}
							)
					];					
				}
			);
		},
	});	 
};

//
// Throw an exception if the sort method doesn't make sense.
//
var validateSortMethod = function (sortMethod) {
	assert.isString(sortMethod);
	assert(
		sortMethod === 'orderBy' || 
	   sortMethod === 'orderByDescending' ||
	   sortMethod === 'thenBy' ||
	   sortMethod === 'thenByDescending', 
	   "Expected 'sortMethod' to be one of 'orderBy', 'orderByDescending', 'thenBy' or 'thenByDescending', instead it is '" + sortMethod + "'."
   );
};

//
// Execute a batched sorting command.
//
var executeOrderBy = function (self, batch) {

	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);

	//todo: reconsider how this works wih lazy iterators.

	//
	// Don't invoke the sort until we really know what we need.
	//
	var executeLazySort = function () {

		batch.forEach(function (orderCmd) {
			assert.isObject(orderCmd);
			assert.isFunction(orderCmd.sortSelector);
			validateSortMethod(orderCmd.sortMethod);
		});

		var pairs = [];

		var iterator = self.getIterator();
		while (iterator.moveNext()) {
			pairs.push(iterator.getCurrent());
		}

		return E.from(batch)
			.aggregate(E.from(pairs), function (unsorted, orderCmd) {
				return unsorted[orderCmd.sortMethod](function (pair) {
					assert.isArray(pair);
					assert(pair.length === 2);
					return orderCmd.sortSelector(pair[1]);
				}); 
			})
			.toArray();
	};

	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new ArrayIterator(executeLazySort());
		},
	});
};

//
// Order by values in a partcular column, either ascending or descending
//
var orderBy = function (self, sortMethod, sortSelector) {
	assert.isObject(self);
	validateSortMethod(sortMethod);
	assert.isFunction(sortSelector);

	var batchOrder = [
		{ 
			sortSelector: sortSelector, 
			sortMethod: sortMethod 
		}
	];

	var sortedDataFrame = executeOrderBy(self, batchOrder);
	sortedDataFrame.thenBy = orderThenBy(self, batchOrder, 'thenBy');
	sortedDataFrame.thenByDescending = orderThenBy(self, batchOrder, 'thenByDescending');	
	return sortedDataFrame;
};

//
// Process a column selector that might be a column name, column index or selector function.
// Returns a selector fucntion.
//
var processColumnSelector = function (self, columnNameOrIndexOrSelector, fnName) {
	assert.isObject(self);
	assert.isString(fnName);

	if (!Object.isFunction(columnNameOrIndexOrSelector)) {

		var columnName;

		if (Object.isNumber(columnNameOrIndexOrSelector)) {
			var columnNames = self.getColumnNames();
			assert(
				columnNameOrIndexOrSelector >= 0 && columnNameOrIndexOrSelector < columnNames.length, 
				"Bad column index specified for 'columnNameOrIndexOrSelector' parameter to 'orderBy', expected a column index >= 0 and < " + columnNames.length
			);
			
			columnName = columnNames[columnNameOrIndexOrSelector];
		}
		else {
			assert.isString(
				columnNameOrIndexOrSelector, 
				"Expected 'columnNameOrIndexOrSelector' parameter to '" + fnName + "' to be a column name, a column index or a selector function."
			);

			columnName = columnNameOrIndexOrSelector;
		}

		columnNameOrIndexOrSelector = function (row) {
				return row[columnName];
			};
	}

	return columnNameOrIndexOrSelector;
};


//
// Generates a thenBy function that is attached to already ordered data frames.
//
var orderThenBy = function (self, batch, nextSortMethod) {
	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);
	validateSortMethod(nextSortMethod);
	
	return function (columnNameOrIndexOrSelector) {

		var extendedBatch = batch.concat([
			{
				sortSelector: processColumnSelector(self, columnNameOrIndexOrSelector, 'thenBy'),
				sortMethod: nextSortMethod,
			},
		]);

		var sortedDataFrame = executeOrderBy(self, extendedBatch);
		sortedDataFrame.thenBy = orderThenBy(self, extendedBatch, 'thenBy');
		sortedDataFrame.thenByDescending = orderThenBy(self, extendedBatch, 'thenByDescending');		
		return sortedDataFrame;
	};	
};

/**
 * Sorts a data frame based on a single column (specified by name or index) or by selector (ascending). 
 * 
 * @param {string|index|function} columnNameOrIndexOrSelector - A column name, column index or selector function that indicates the value to sort by.
 */
DataFrame.prototype.orderBy = function (columnNameOrIndexOrSelector) {

	var self = this;
	return orderBy(self, 'orderBy', processColumnSelector(self, columnNameOrIndexOrSelector, 'orderBy'));
};

/**
 * Sorts a data frame based on a single column (specified by name or index) or by selector (descending). 
 * 
 * @param {string|index|function} columnNameOrIndexOrSelector - A column name, column index or selector function that indicates the value to sort by.
 */
DataFrame.prototype.orderByDescending = function (columnNameOrIndexOrSelector) {

	var self = this;
	return orderBy(self, 'orderByDescending', processColumnSelector(self, columnNameOrIndexOrSelector, 'orderByDescending'));
};

/**
 * Create a new data frame with the requested column or columns dropped.
 *
 * @param {string|array} columnOrColumns - Specifies the column name (a string) or columns (array of column names) to drop.
 */
DataFrame.prototype.dropSeries = function (columnOrColumns) {

	var self = this;

	if (!Object.isArray(columnOrColumns)) {
		assert.isString(columnOrColumns, "'DataFrame.dropSeries' expected either a string or an array or strings.");

		columnOrColumns = [columnOrColumns]; // Convert to array for coding convenience.
	}

	var columnNames = self.getColumnNames().slice(0); // Clone array.
	var newColumnNames = E.from(columnNames)
		.where(function (columnName) {
			return !E.from(columnOrColumns).contains(columnName);
		})
		.toArray();

	return new DataFrame({
		columnNames: newColumnNames,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					var row = extend({}, pair[1]);
					columnOrColumns.forEach(function (columnName) {
						delete row[columnName];
					});
					return [
						pair[0],
						row
					];					
				}
			);
		},
	});
};

/**
 * Create a new data frame with only the requested column or columns dropped, other columns are dropped.
 *
 * @param {string|array} columnOrColumns - Specifies the column name (a string) or columns (array of column names) to keep.
 */
DataFrame.prototype.keepSeries = function (columnOrColumns) {

	var self = this;

	if (!Object.isArray(columnOrColumns)) {
		assert.isString(columnOrColumns, "'DataFrame.keepSeries' expected either a string or an array or strings.");

		columnOrColumns = [columnOrColumns]; // Convert to array for coding convenience.
	}

	return new DataFrame({
		columnNames: columnOrColumns,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					var row = extend({}, pair[1]);
					Object.keys(row).forEach(function (fieldName) {
						if (!E.from(columnOrColumns).contains(fieldName)) {
							delete row[fieldName];
						}
					});
					return [
						pair[0],
						row
					];					
				}
			);
		},
	});
};
/**
 * Create a new data frame with an additional column specified by the passed-in series.
 *
 * @param {string} columnName - The name of the column to add or replace.
 * @param {array|column} data - Array of data or column that contains data.
 */
DataFrame.prototype.setSeries = function (columnName, data) {

	assert.isString(columnName, "Expected 'columnName' parameter to 'setSeries' function to be a string that specifies the column to set or replace.");

	var self = this;

	if (Object.isFunction(data)) {
		data = E.from(self.toPairs()) //todo: make this lazy
			.select(function (pair) {
				return data(pair[1], pair[0]);
			})
			.toArray();
	}
	else if (!Object.isArray(data)) {
		assert.isObject(data, "Expected 'data' parameter to 'setSeries' to be either an array or a series object.");
		assert.isFunction(data.reindex, "Expected 'data' parameter to 'setSeries' to have a 'reindex' function that allows the column to be reindexed.");

		data = data.reindex(self.getIndex()).toValues();
	}

	//todo: overview and improve the way this works.

	var columnIndex = self.getColumnIndex(columnName);
	if (columnIndex < 0) {		
		// Add new column.
		return new DataFrame({
			columnNames: self.getColumnNames().concat([columnName]),
			rows: function () {
				return new ArrayIterator(
					E.from(self.toValues())
						.select(function (row, rowIndex) {
							return row.concat([data[rowIndex]]);
						})
						.toArray()
				);
			},
			index: self.getIndex(),
		});
	}
	else {
		// Replace existing column.
		return new DataFrame({
			columnNames: E.from(self.getColumnNames())
				.select(function (thisColumnName, thisColumnIndex) {
					if (thisColumnIndex === columnIndex) {
						return columnName;
					}
					else { 
						return thisColumnName;
					}
				})
				.toArray(),
			rows: function () {
				return new ArrayIterator(
					E.from(self.toValues())
						.select(function (row, rowIndex) {
							return E.from(row)
								.select(function (column, thisColumnIndex) {
									if (thisColumnIndex === columnIndex) {
										return data[rowIndex];
									}
									else {
										return column;
									}
								})
								.toArray();
						})
						.toArray()
				);
			},
			index: self.getIndex(),
		});
	}
};

/**
 * Create a new data-frame from a slice of rows.
 *
 * @param {int|function} startIndexOrStartPredicate - Index where the slice starts or a predicate function that determines where the slice starts.
 * @param {int|function} endIndexOrEndPredicate - Marks the end of the slice, one row past the last row to include. Or a predicate function that determines when the slice has ended.
 * @param {function} [predicate] - Optional predicate to compare index against start/end index. Return true to start or stop the slice.
 */
DataFrame.prototype.slice = function (startIndexOrStartPredicate, endIndexOrEndPredicate, predicate) {

	var self = this;

	var startIndex;
	var endIndex;
	var startPredicate = null;
	var endPredicate = null;

	if (predicate) {
		assert.isFunction(predicate, "Expected 'predicate' parameter to slice function to be function.");
	}

	if (Object.isFunction(startIndexOrStartPredicate)) {
		startPredicate = startIndexOrStartPredicate;
	}
	else {
		startIndex = startIndexOrStartPredicate;
		startPredicate = function (value) {
				return predicate && predicate(value, startIndex) || value < startIndex;
			};
	}

	if (Object.isFunction(endIndexOrEndPredicate)) {
		endPredicate = endIndexOrEndPredicate;
	}
	else {
		endIndex = endIndexOrEndPredicate;
		endPredicate = function (value) {
				return predicate && predicate(value, endIndex) || value < endIndex;
			};
	}

	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeWhileIterator(
				new SkipWhileIterator(
					self.getIterator(),
					function (pair) {
						return startPredicate(pair[0]); // Check index for start condition.
					}
				),
				function (pair) {
					return endPredicate(pair[0]); // Check index for end condition.
				}
			);
		},		
	});

};

/**
 * Set a named column as the index of the data-frame.
 *
 * @param {string} columnName - Name or index of the column to set as the index.
 */
DataFrame.prototype.setIndex = function (columnName) {

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new PairIterator(
				new SelectIterator(
					self.getSeries(columnName).getIterator(),
					function (pair) {
						return pair[1];
					}
				),				
				new SelectIterator(
					self.getIterator(),
					function (pair) {
						return pair[1];
					}
				)
			);
		},
	});
};

/**
 * Reset the index of the data frame back to the default sequential integer index.
 */
DataFrame.prototype.resetIndex = function () {

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair, i) {
					return [i, pair[1]];
				}
			);
		},
	});
};

/** 
 * Format the data frame for display as a string.
 */
DataFrame.prototype.toString = function () {

	var self = this;
	var Table = require('easy-table');

	var index = self.getIndex().toValues();
	var header = ["__index__"].concat(self.getColumnNames());
	var rows = E.from(self.toValues())
		.select(function (row, rowIndex) { 
			return [index[rowIndex]].concat(row);
		})
		.toArray()

	var t = new Table();
	rows.forEach(function (row, rowIndex) {
		row.forEach(function (cell, cellIndex) {
			t.cell(header[cellIndex], cell);
		});
		t.newRow();
	});

	return t.toString();
};

/**
 * Parse a column with string values to a column with int values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 */
DataFrame.prototype.parseInts = function (columnNameOrNames) {

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseInts());
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseInts());
	}
};

/**
 * Parse a column with string values to a column with float values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 */
DataFrame.prototype.parseFloats = function (columnNameOrNames) {

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseFloats());
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseFloats());
	}
};

/**
 * Parse a column with string values to a column with date values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 * @param {string} [formatString] - Optional formatting string for dates.
 */
DataFrame.prototype.parseDates = function (columnNameOrNames, formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseDates(formatString));
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseDates(formatString));
	}
};

/**
 * Convert a column of values of different types to a column of string values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to convert to strings.
 * @param {string} [formatString] - Optional formatting string for dates.
 */
DataFrame.prototype.toStrings = function (columnNameOrNames, formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).toStrings(formatString));
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).toStrings(formatString));
	}
};

/**
 * Detect actual types and their frequencies contained within columns in the data frame.
 */
DataFrame.prototype.detectTypes = function () {

	var self = this;

	var dataFrames = E.from(self.getColumns())
		.select(function (column) {
			var series = column.series;
			var numValues = series.toValues().length;
			//todo: broad-cast column
			var newSeries = new Series({
				values: E.range(0, numValues)
					.select(function () { 
						return column.name; 
					})
					.toArray()
			});
			return column.series
				.detectTypes()
				.setSeries('Column', newSeries);
		})
		.toArray();
	return concatDataFrames(dataFrames).resetIndex();
};

/**
 * Detect values and their frequencies contained within columns in the data frame.
 */
DataFrame.prototype.detectValues = function () {

	var self = this;

	var dataFrames = E.from(self.getColumns())
		.select(function (column) {
			var numValues = column.series.toValues().length;
			//todo: broad-cast column
			var newSeries = new Series({
				values: E.range(0, numValues)
					.select(function () { 
						return column.name 
					})
					.toArray()
			});
			return column.series.detectValues().setSeries('Column', newSeries);
		})
		.toArray();
	return concatDataFrames(dataFrames).resetIndex();
};
/**
 * Produces a new data frame with all string values truncated to the requested maximum length.
 *
 * @param {int} maxLength - The maximum length of the string values after truncation.
 */
DataFrame.prototype.truncateStrings = function (maxLength) {
	assert.isNumber(maxLength, "Expected 'maxLength' parameter to 'truncateStrings' to be an integer.");

	var self = this;
	var truncatedValues = E.from(self.toValues()) //todo: make this function lazy.
		.select(function (row) {
			return E.from(row)
				.select(function (value) {
					if (Object.isString(value)) {
						if (value.length > maxLength) {
							return value.substring(0, maxLength);
						}
					}

					return value;
				})
				.toArray();
		})
		.toArray();

	return new DataFrame({
			columnNames: self.getColumnNames(),
			rows: truncatedValues,
		});
};

/**
 * Create a new data frame with columns reordered.
 * New column names create new columns (with undefined values), omitting existing column names causes those columns to be dropped.
 * 
 * @param {array} columnNames - The new order for columns. 
 */
DataFrame.prototype.remapColumns = function (columnNames) {

	assert.isArray(columnNames, "Expected parameter 'columnNames' to remapColumns to be an array with column names.");

	columnNames.forEach(function (columnName) {
		assert.isString(columnName, "Expected parameter 'columnNames' to remapColumns to be an array with column names.");
	});

	var self = this;

	return new DataFrame({
		columnNames: columnNames,
		rows: function () { //todo: make this properly lazy.
			return new ArrayIterator(
				E.from(self.toValues())
					.select(function (row) {
						return E.from(columnNames)
							.select(function (columnName) {
								var columnIndex = self.getColumnIndex(columnName);
								if (columnIndex >= 0) {
									return row[columnIndex];
								}
								else { 
									// Column doesn't exist.
									return undefined;
								}
							})
							.toArray();
					})
					.toArray()
			);
		},
	});
};

/**
 * Create a new data-frame with renamed series.
 *
 * @param {array|object} newColumnNames|columnsMap - Array of strings, with an element for each existing column that specifies the new name of that column. Or, a hash that maps old column name to new column name.
 */
DataFrame.prototype.renameSeries = function (newColumnNames) {

	var self = this;

	if (Object.isObject(newColumnNames)) {
		var self = this;
		var renamedColumns = self.getColumnNames().slice(0); // Clone array.

		Object.keys(newColumnNames).forEach(function (existingColumnName, columnIndex) {
			var columnIndex = self.getColumnIndex(existingColumnName);
			if (columnIndex === -1) {
				return; // No column to be renamed.
			}

			renamedColumns[columnIndex] = newColumnNames[existingColumnName];
		});

		return self.renameSeries(renamedColumns);
	}
	else {
		var existingColumns = self.getColumnNames();
		var numExistingColumns = existingColumns.length;

		assert.isArray(newColumnNames, "Expected parameter 'newColumnNames' to renameColumns to be an array with column names.");
		assert(newColumnNames.length === numExistingColumns, "Expected 'newColumnNames' array to have an element for each existing column. There are " + numExistingColumns + "existing columns.");

		return new DataFrame({
			columnNames: newColumnNames,
			iterable: function () {
				var columnMap = E.from(existingColumns)
					.zip(newColumnNames, function (oldName, newName) {
						return [oldName, newName];
					})
					.toArray();

				return new SelectIterator(
					self.getIterator(),
					function (pair) {
						return [
							pair[0],
							E.from(columnMap).toObject(
								function (remap) {
									return remap[1];
								},
								function (remap) {
									return pair[1][remap[0]];								
								}
							)
						];
					}
				);
			},
		});
	}
};

/**
 * Bake the data frame to an array of rows.
 */
DataFrame.prototype.toValues = function () {

	var self = this;

	var iterator = self.getIterator();
	validateIterator(iterator);

	var values = [];
	var columnNames = self.getColumnNames();

	while (iterator.moveNext()) {
		var curRow = iterator.getCurrent()[1];

		var asArray = [];
		for (var columnIndex = 0; columnIndex < columnNames.length; ++columnIndex) {
			asArray.push(curRow[columnNames[columnIndex]]);
		}

		values.push(asArray);
	}

	return values;
};

/**
 * Bake the data frame to an array of JavaScript objects.
 */
DataFrame.prototype.toObjects = function () {

	var self = this;

	var iterator = self.getIterator();
	validateIterator(iterator);

	var objects = [];

	while (iterator.moveNext()) {
		objects.push(iterator.getCurrent()[1]); // Extract values.
	}

	return objects;
};

/**
 * Serialize the data frame to JSON.
 */
DataFrame.prototype.toJSON = function () {
	var self = this;
	return JSON.stringify(self.toObjects(), null, 4);
};

/**
 * Serialize the data frame to CSV.
 */
DataFrame.prototype.toCSV = function () {

	var self = this;
	var data = [self.getColumnNames()].concat(self.toValues());
	return BabyParse.unparse(data);

	/*Old csv stringify.
	var header = self.getColumnNames().join(',');
	var rows = E.from(self.toValues())
			.select(function (row) {
				return row.join(',');
			})
			.select(function (col) { // Strip newlines... these don't work in CSV files.
				if (Object.isString(col)) { //todo: not necessar if all columns are converted to strings.
					return col.replace(/\r\n/g, ' ').replace(/\n/g, ' ');
				}
				else {
					return col;
				}
			})					
			.toArray();
	return [header].concat(rows).join('\r\n');	
	*/
};

/**
 * Retreive the data as pairs of [index, objects].
 */
DataFrame.prototype.toPairs = function () {

	var self = this;

	var iterator = self.getIterator();
	validateIterator(iterator);

	var pairs = [];

	while (iterator.moveNext()) {
		pairs.push(iterator.getCurrent());
	}

	return pairs;
};

/**
 * Forces lazy evaluation to complete and 'bakes' the data frame into memory.
 */
DataFrame.prototype.bake = function () {

	var self = this;
	if (self._baked) {
		// Already baked, just return self.
		return self;
	}

	var pairs = self.toPairs();
	var baked = new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new ArrayIterator(pairs);
		},
	});
	baked._baked = true;
	return baked;
};
<commit_msg>Removed the toPairs fn from DataFrame. The Series function does the job.<commit_after>'use strict';

// 
// Base class for data frame classes.
//

var ArrayIterator = require('./iterators/array');
var MultiIterator = require('./iterators/multi');
var PairIterator = require('./iterators/pair');
var SkipIterator = require('./iterators/skip');
var SkipWhileIterator = require('./iterators/skip-while');
var BabyParse = require('babyparse');
var SelectIterator = require('../src/iterators/select');
var SelectManyIterator = require('../src/iterators/select-many');
var TakeIterator = require('../src/iterators/take');
var TakeWhileIterator = require('../src/iterators/take-while');
var WhereIterator = require('../src/iterators/where');
var CountIterator = require('../src/iterators/count');
var EmptyIterator = require('../src/iterators/empty');
var utils = require('./utils');
var extend = require('extend');
var inherit = require('./inherit');

var assert = require('chai').assert; 
var E = require('linq');

var validateIterator = require('./iterators/validate');

//
// Creates an iterator that converts rows to JavaScript objects based on passed in column names.
//
var convertRowsToObjects = function (columnNames, rowsIterator) {

	if (Object.isFunction(columnNames)) {
		columnNames = columnNames();
	}

	assert.isArray(columnNames);

	validateIterator(rowsIterator);

	return new SelectIterator(
		rowsIterator,
		function (row) {
			return E.from(columnNames)
				.select(function (columnName, columnIndex) {
					return [columnName, columnIndex];
				})
				.toObject(
					function (column) {
						return column[0];
					},
					function (column) {
						return row[column[1]];
					}
				);							
		}
	);
};

//
// Determine column names from an array of rows. Column names are take from the fields in each JavaScript object.
//
var determineColumnNamesFromObjectRows = function (rows, considerAllRows)  {

	assert.isArray(rows);

	if (considerAllRows) {
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		if (rows.length > 0) {
			// Just consider the first row.
			return Object.keys(rows[0]);
		}
		else {
			// Can't do this, there are no rows.
			return [];
		}
	}
};

//
// Evaluate an iterable of JavaScript objects, look at the fields and figure out column names from that.
//
var determineColumnNamesFromObjectsIterable = function (iterable, considerAllRows) {

	assert.isFunction(iterable);

	var iterator = iterable();

	if (considerAllRows) {
		// Consider all rows, this expensive, so it is optional.
		var rows = [];
		while (iterator.moveNext()) {
			rows.push(iterator.getCurrent());
		}
		
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		// Just consider the first row.		
		if (!iterator.moveNext()) {
			return []; // Nothing in the iterable.
		}

		return Object.keys(iterator.getCurrent());
	}
};

//
// Evaluate an iterable of index/value pairs, look at the fields in the vlaues and figure out column names from that.
//
var determineColumnNamesFromPairsIterable = function (iterable, considerAllRows) {

	assert.isFunction(iterable);

	var iterator = iterable();

	if (considerAllRows) {
		// Consider all rows, this expensive, so it is optional.
		var rows = [];
		while (iterator.moveNext()) {
			rows.push(iterator.getCurrent()[1]);
		}
		
		return E.from(rows)
			.selectMany(function (row) {
				return Object.keys(row);
			})
			.distinct()
			.toArray();
	}
	else {
		// Just consider the first row.
		if (!iterator.moveNext()) {
			return []; // Nothing in the iterable.
		}

		return Object.keys(iterator.getCurrent()[1]);
	}
};

//
// Constuctor.
//
var DataFrame = function (config) {

	var self = this;

	if (!config) {
		self._columnNames = [];
		self.getIterator = function () {
			return new EmptyIterator();
		};
		return;
	}

	if (config.iterable) {

		assert.isFunction(config.iterable, "Expected 'iterable' field of 'config' parameter to DataFrame constructor to be a function that returns an index/value pairs iterator.");

		var iterable = config.iterable;

		if (config.columnNames) {
			assert.isArray(config.columnNames, "Expected 'columnNames' field of 'config' parameter to DataFrame constructor to be an array of column names or function that returns an array of column names.");

			self._columnNames = config.columnNames;
		}
		else {
			self._columnNames = determineColumnNamesFromPairsIterable(iterable, config.considerAllRows);
		}
		self.getIterator = iterable;
		return;
	}

	if (!config.rows &&  !config.columns) {
		self._columnNames = config.columnNames || [];
		self.getIterator = function () {
			return new EmptyIterator();
		};
		return;
	}

	assert.isObject(config, "Expected 'config' parameter to DataFrame constructor to be an object with options for initialisation.");

	if (config.index) {
		if (!Object.isArray(config.index)) {
			assert.isObject(config.index, "Expected 'index' member of 'config' parameter to DataFrame constructor to be an object.");			
		}
	}

	if (config.columnNames) {
		assert.isArray(config.columnNames, "Expected 'columnNames' member of 'config' parameter to DataFrame constructor to be an array of strings or a function that produces an array of strings.");

		config.columnNames.forEach(function (columnName) {
			assert.isString(columnName, "Expected 'columnNames' member of 'config' parameter to DataFrame constructor to be an array of strings or a function that produces an array of strings.");
		});

		if (!config.rows) {
			throw new Error("Expected to find a 'rows' member of 'config' parameter to DataFrame constructor.");
		}

	 	if (!Object.isFunction(config.rows)) {
			assert.isArray(config.rows, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of rows.");

			if (config.debug) {
				config.rows.forEach(function (row) {
					assert.isArray(row, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of arrays, an array of objects or an iterator.");
				});
			}
		}
	}
	else if (config.rows) {
		assert(!config.columns, "Can't use both 'rows' and 'columns' fields of 'config' parameter to DataFrame constructor.");

		if (!Object.isFunction(config.rows)) {
			assert.isArray(config.rows, "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of JavaScript objects.");
			
			if (config.rows.length > 0) {
				assert.isObject(config.rows[0], "Expected 'rows' member of 'config' parameter to DataFrame constructor to be an array of JavaScript objects.")
			}
		}
	}
	else if (config.columns) {
		assert.isObject(config.columns, "Expected 'columns' member of 'config' parameter to DataFrame constructor to be an object with fields that define columns.");
	}

	var rows = config.rows;
	var columns = config.columns;

	if (config.columnNames)	{
		self._columnNames = config.columnNames;

		if (config.index) {
			var index = config.index;
			if (Object.isArray(index)) {

				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, rows()));
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, new ArrayIterator(rows)));
					};
				}
			}
			else {
				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								} 
							),	
							convertRowsToObjects(self._columnNames, rows())
						);
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								}
							),
							convertRowsToObjects(self._columnNames, new ArrayIterator(rows))
						);
					};				
				}
			}
		}
		else {
			if (Object.isFunction(rows)) {
				this.getIterator = function () {
					return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, rows()));
				};
			}
			else {
				this.getIterator = function () {
					return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, new ArrayIterator(rows)));
				};
			}
		}	
	}
	else {
		if (rows) {
			if (Object.isFunction(rows)) {
				self._columnNames = determineColumnNamesFromObjectsIterable(rows, config.considerAllRows);
			}
			else {
				// Derive column names from object fields.
				self._columnNames = determineColumnNamesFromObjectRows(rows, config.considerAllRows);
			}
		}
		else {
			self._columnNames = Object.keys(columns);
		}

		if (config.index) {
			var index = config.index;
			if (Object.isArray(index)) {

				if (rows) {
					if (Object.isFunction(rows)) {
						this.getIterator = function () {
							return new PairIterator(new ArrayIterator(index), rows());
						};
					}
					else {
						this.getIterator = function () {
							return new PairIterator(new ArrayIterator(index), new ArrayIterator(rows));
						};
					}
				}
				else {
					this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(new ArrayIterator(index), convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)));
					};
				}
			}
			else {
				if (rows) {
					if (Object.isFunction(rows)) {
						this.getIterator = function () {
							return new PairIterator(
								new SelectIterator(
									index.getIterator(),
									function (pair) {
										return pair[1];
									}	
								), 
								rows()
							);
						};
					}
					else {
						this.getIterator = function () {
							return new PairIterator(
								new SelectIterator(
									index.getIterator(),
									function (pair) {
										return pair[1];
									}
								), 
								new ArrayIterator(rows)
							);
						};				
					}
				}
				else {
					this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(
							new SelectIterator(
								index.getIterator(),
								function (pair) {
									return pair[1];
								}
							), 
							convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)))
						;
					};
				}
			}
		}
		else {
			if (rows) {
				if (Object.isFunction(rows)) {
					this.getIterator = function () {
						return new PairIterator(new CountIterator(), rows());
					};
				}
				else {
					this.getIterator = function () {
						return new PairIterator(new CountIterator(), new ArrayIterator(rows));
					};
				}
			}
			else {
				this.getIterator = function () {
						var columnIterators = E.from(self._columnNames)
							.select(function (columnName) {
								return new ArrayIterator(columns[columnName]);
							})
							.toArray();
						return new PairIterator(new CountIterator(), convertRowsToObjects(self._columnNames, new MultiIterator(columnIterators)));
				};
			}
		}
	}
};

module.exports = DataFrame;

var Series = require('./series');
var parent = inherit(DataFrame, Series);

var concatDataFrames = require('./concat-dataframes');
var zipDataFrames = require('./zip-dataframes');
var mergeSeries = require('./merge-series');
var mergeDataFrames = require('./merge-dataframes');

/**
 * Get the index of the data frame.
 */
DataFrame.prototype.getIndex = function () {
	var self = this;
	return new Series({
		iterable: function () {		
			return new SelectIterator(
				self.getIterator(),
				function (pair, index) {
					return [index, pair[0]]; // Extract index.
				}
			);
		},
	});
};

/**
 * Get the names of the columns in the data frame.
 */
DataFrame.prototype.getColumnNames = function () {
	var self = this;
	return self._columnNames;
};

/**
 * Get an iterator for the data-frame.
 */
DataFrame.prototype.getIterator = function () {
	return new EmptyIterator(); // This function is defined by the constructor.
};

/**
 * Gets a column index from a column name.
 *
 * @param {string} columnName - The name of the column to retrieve the column index for.
 *
 * @returns {Number} Returns the index of the named column or -1 if the requested column was not found.
 */
DataFrame.prototype.getColumnIndex = function (columnName) {
	assert.isString(columnName, "Expected 'columnName' parameter to getColumnIndex to be a non-empty string.");
	
	var self = this;	
	var columnNames = self.getColumnNames();
	
	for (var i = 0; i < columnNames.length; ++i) {
		if (columnName === columnNames[i]) {
			return i;
		}
	}	
	
	return -1;
};

/**
 * Gets a column name from a column index.
 *
 * @param {int} columnIndex - The index of the column to retrieve the column name for.
 *
 * @returns {string} Returns the name of the column or undefined if the requested column was not found.
 */
DataFrame.prototype.getColumnName = function (columnIndex) {
	assert.isNumber(columnIndex, "Expected 'columnIndex' parameter to getColumnIndex to be a non-empty string.");

	var self = this;	
	var columnNames = self.getColumnNames();

	if (columnIndex < 0 || columnIndex >= columnNames.length) {
		return undefined;
	}

	return columnNames[columnIndex];
};

/**
 * Skip a number of rows in the data frame.
 *
 * @param {int} numRows - Number of rows to skip.
 */
DataFrame.prototype.skip = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'skip' function to be a number.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SkipIterator(self.getIterator(), numRows);
		},
	}); 	
};

/**
 * Skips rows in the data-frame while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.skipWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SkipWhileIterator(self.getIterator(), function (pair) {
					return predicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Skips rows in the data-frame until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.skipUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'skipUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.skipWhile(function (value, index) { 
		return !predicate(value, index); 
	});
};

/**
 * Take a number of rows in the data frame.
 *
 * @param {int} numRows - Number of rows to take.
 */
DataFrame.prototype.take = function (numRows) {
	assert.isNumber(numRows, "Expected 'numRows' parameter to 'take' function to be a number.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeIterator(self.getIterator(), numRows);
		},
	}); 	
};

/**
 * Take rows from the data-frame while a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.takeWhile = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeWhile' function to be a predicate function that returns true/false.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeWhileIterator(self.getIterator(), function (pair) {
					return predicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Take rows from the data-frame until a condition is met.
 *
 * @param {function} predicate - Return true to indicate the condition met.
 */
DataFrame.prototype.takeUntil = function (predicate) {
	assert.isFunction(predicate, "Expected 'predicate' parameter to 'takeUntil' function to be a predicate function that returns true/false.");

	var self = this;
	return self.takeWhile(function (value, index) { 
		return !predicate(value, index); 
	});
};

/**
 * Filter a data frame by a predicate selector.
 *
 * @param {function} filterSelectorPredicate - Predicte function to filter rows of the data frame.
 */
DataFrame.prototype.where = function (filterSelectorPredicate) {
	assert.isFunction(filterSelectorPredicate, "Expected 'filterSelectorPredicate' parameter to 'where' function to be a function.");

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new WhereIterator(self.getIterator(), function (pair) {
					return filterSelectorPredicate(pair[1], pair[0]);
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each row to generate a transformed data-frame.
 */
DataFrame.prototype.select = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.select' to be a selector functions.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectIterator(self.getIterator(), function (pair) {
					var newValue = selector(pair[1], pair[0]);
					if (!Object.isObject(newValue)) {
						throw new Error("Expected return value from 'select' selector to be an object that represents a new row in the resulting data-frame.");
					};
					return [
						pair[0],
						newValue,
					];
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index and row pair to generate a transformed data-frame.
 */
DataFrame.prototype.selectPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectPairs' to be a selector functions.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectIterator(self.getIterator(), function (pair) {
					var newPair = selector(pair[1], pair[0]);
					if (!Object.isArray(newPair) || newPair.length !== 2 || !Object.isObject(newPair[1])) {
						throw new Error("Expected return value from 'DataFrame.selectPairs' selector to be a pair, that is an array with two items: [index, object].");
					}
					return newPair;
				});
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each row to create a new data-frame.
 */
DataFrame.prototype.selectMany = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectMany' function to be a function.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), function (pair) {
				var newRows = selector(pair[1], pair[0]);

				if (!Object.isArray(newRows) &&
					!(newRows instanceof Series) &&
					!(newRows instanceof DataFrame)) {
					throw new Error("Expected return value from 'DataFrame.selectMany' selector to be an array of objects, a Series or a DataFrame, each object represents a new row in the resulting data-frame.");
				}

				if (newRows instanceof DataFrame) {
					newRows = newRows.toObjects();
				}
				else if (newRows instanceof Series)
				{
					newRows = newRows.toValues();
				}

				var newPairs = [];
				for (var newRowIndex = 0; newRowIndex < newRows.length; ++newRowIndex) {
					var newRow = newRows[newRowIndex];
					if (!Object.isObject(newRow)) {
						throw new Error("Expected array returned from 'DataFrame.selectMany' selector to contain only objects, each object represents a new row in the resulting data-frame.");
					};

					newPairs.push([
						pair[0], 
						newRow,
					]);
				}

				return newPairs;
			})
		},
	}); 	
};

/**
 * Generate a new data frame based on the results of the selector function.
 *
 * @param {function} selector - Selector function that transforms each index/row pair to create a new data-frame.
 */
DataFrame.prototype.selectManyPairs = function (selector) {
	assert.isFunction(selector, "Expected 'selector' parameter to 'DataFrame.selectManyPairs' function to be a function.");

	var self = this;
	return new DataFrame({
		iterable: function () {
			return new SelectManyIterator(self.getIterator(), function (pair) {
				var newPairs = selector(pair[1], pair[0]);
				if (!Object.isArray(newPairs)) {
					throw new Error("Expected return value from 'DataFrame.selectManyPairs' selector to be an array of pairs, each item in the array represents a new pair in the resulting DataFrame.");
				}

				for (var pairIndex = 0; pairIndex < newPairs.length; ++pairIndex) {
					var newPair = newPairs[pairIndex];
					if (!Object.isArray(newPair) || newPair.length !== 2 || !Object.isObject(newPair[1])) {
						throw new Error("Expected return value from 'DataFrame.selectManyPairs' selector to be am array of pairs, but item at index " + pairIndex + " is not an array with two items: [index, object].");
					}
				}

				return newPairs;
			})
		},
	}); 	
};

/**
 * Retreive a time-series from a column of the data-frame.
 *
 * @param {string} columnName - Specifies the column to retreive.
 */
DataFrame.prototype.getSeries = function (columnName) {
	var self = this;

	assert.isString(columnName, "Expected 'columnName' parameter to getSeries function to be a string that specifies the name of the column to retreive.");

	return new Series({
		iterable: function () {
			return new WhereIterator(
				new SelectIterator(self.getIterator(), function (pair) {
					return [
						pair[0],
						pair[1][columnName],
					];
				}),
				function (pair) {
					return pair[1] !== undefined;
				}
			);
		},
	});
};

/**
 * Returns true if the column with the requested name exists in the data frame.
 *
 * @param {string} columnName - Name of the column to check.
 */
DataFrame.prototype.hasSeries = function (columnName) {

	assert.isString(columnName);

	var self = this;
	return self.getColumnIndex(columnName) >= 0;
};

/**
 * 
 * Verify the existance of a column and return it.
 * Throws an exception if the column doesn't exist.
 *
 * @param {string} columnName - Name or index of the column to retreive.
 */
DataFrame.prototype.expectSeries = function (columnName) {

	var self = this;
	if (self.getColumnNames().indexOf(columnName) < 0) {
		throw new Error("Expected data-frame to contain series with column name: '" + columnName + "'.");
	}
	return self;
};

/** 
 * Retreive a collection of all columns.
 */
DataFrame.prototype.getColumns = function () {

	var self = this;

	return E.from(self.getColumnNames())
		.select(function (columnName) {
			return {
				name: columnName,
				series: self.getSeries(columnName),
			};
		})
		.toArray();
};

/**
 * Create a new data-frame from a subset of columns.
 *
 * @param {array} columnNames - Array of column names to include in the new data-frame.
 */
DataFrame.prototype.subset = function (columnNames) {

	var self = this;
	
	assert.isArray(columnNames, "Expected 'columnNames' parameter to 'subset' to be an array.");	
	
	return new DataFrame({
		columnNames: columnNames,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					return [
						pair[0],
						E.from(columnNames)
							.toObject(
								function (columnName) {
									return columnName;
								},
								function (columnName) {
									return pair[1][columnName];
								}
							)
					];					
				}
			);
		},
	});	 
};

//
// Throw an exception if the sort method doesn't make sense.
//
var validateSortMethod = function (sortMethod) {
	assert.isString(sortMethod);
	assert(
		sortMethod === 'orderBy' || 
	   sortMethod === 'orderByDescending' ||
	   sortMethod === 'thenBy' ||
	   sortMethod === 'thenByDescending', 
	   "Expected 'sortMethod' to be one of 'orderBy', 'orderByDescending', 'thenBy' or 'thenByDescending', instead it is '" + sortMethod + "'."
   );
};

//
// Execute a batched sorting command.
//
var executeOrderBy = function (self, batch) {

	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);

	//todo: reconsider how this works wih lazy iterators.

	//
	// Don't invoke the sort until we really know what we need.
	//
	var executeLazySort = function () {

		batch.forEach(function (orderCmd) {
			assert.isObject(orderCmd);
			assert.isFunction(orderCmd.sortSelector);
			validateSortMethod(orderCmd.sortMethod);
		});

		var pairs = [];

		var iterator = self.getIterator();
		while (iterator.moveNext()) {
			pairs.push(iterator.getCurrent());
		}

		return E.from(batch)
			.aggregate(E.from(pairs), function (unsorted, orderCmd) {
				return unsorted[orderCmd.sortMethod](function (pair) {
					assert.isArray(pair);
					assert(pair.length === 2);
					return orderCmd.sortSelector(pair[1]);
				}); 
			})
			.toArray();
	};

	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new ArrayIterator(executeLazySort());
		},
	});
};

//
// Order by values in a partcular column, either ascending or descending
//
var orderBy = function (self, sortMethod, sortSelector) {
	assert.isObject(self);
	validateSortMethod(sortMethod);
	assert.isFunction(sortSelector);

	var batchOrder = [
		{ 
			sortSelector: sortSelector, 
			sortMethod: sortMethod 
		}
	];

	var sortedDataFrame = executeOrderBy(self, batchOrder);
	sortedDataFrame.thenBy = orderThenBy(self, batchOrder, 'thenBy');
	sortedDataFrame.thenByDescending = orderThenBy(self, batchOrder, 'thenByDescending');	
	return sortedDataFrame;
};

//
// Process a column selector that might be a column name, column index or selector function.
// Returns a selector fucntion.
//
var processColumnSelector = function (self, columnNameOrIndexOrSelector, fnName) {
	assert.isObject(self);
	assert.isString(fnName);

	if (!Object.isFunction(columnNameOrIndexOrSelector)) {

		var columnName;

		if (Object.isNumber(columnNameOrIndexOrSelector)) {
			var columnNames = self.getColumnNames();
			assert(
				columnNameOrIndexOrSelector >= 0 && columnNameOrIndexOrSelector < columnNames.length, 
				"Bad column index specified for 'columnNameOrIndexOrSelector' parameter to 'orderBy', expected a column index >= 0 and < " + columnNames.length
			);
			
			columnName = columnNames[columnNameOrIndexOrSelector];
		}
		else {
			assert.isString(
				columnNameOrIndexOrSelector, 
				"Expected 'columnNameOrIndexOrSelector' parameter to '" + fnName + "' to be a column name, a column index or a selector function."
			);

			columnName = columnNameOrIndexOrSelector;
		}

		columnNameOrIndexOrSelector = function (row) {
				return row[columnName];
			};
	}

	return columnNameOrIndexOrSelector;
};


//
// Generates a thenBy function that is attached to already ordered data frames.
//
var orderThenBy = function (self, batch, nextSortMethod) {
	assert.isObject(self);
	assert.isArray(batch);
	assert(batch.length > 0);
	validateSortMethod(nextSortMethod);
	
	return function (columnNameOrIndexOrSelector) {

		var extendedBatch = batch.concat([
			{
				sortSelector: processColumnSelector(self, columnNameOrIndexOrSelector, 'thenBy'),
				sortMethod: nextSortMethod,
			},
		]);

		var sortedDataFrame = executeOrderBy(self, extendedBatch);
		sortedDataFrame.thenBy = orderThenBy(self, extendedBatch, 'thenBy');
		sortedDataFrame.thenByDescending = orderThenBy(self, extendedBatch, 'thenByDescending');		
		return sortedDataFrame;
	};	
};

/**
 * Sorts a data frame based on a single column (specified by name or index) or by selector (ascending). 
 * 
 * @param {string|index|function} columnNameOrIndexOrSelector - A column name, column index or selector function that indicates the value to sort by.
 */
DataFrame.prototype.orderBy = function (columnNameOrIndexOrSelector) {

	var self = this;
	return orderBy(self, 'orderBy', processColumnSelector(self, columnNameOrIndexOrSelector, 'orderBy'));
};

/**
 * Sorts a data frame based on a single column (specified by name or index) or by selector (descending). 
 * 
 * @param {string|index|function} columnNameOrIndexOrSelector - A column name, column index or selector function that indicates the value to sort by.
 */
DataFrame.prototype.orderByDescending = function (columnNameOrIndexOrSelector) {

	var self = this;
	return orderBy(self, 'orderByDescending', processColumnSelector(self, columnNameOrIndexOrSelector, 'orderByDescending'));
};

/**
 * Create a new data frame with the requested column or columns dropped.
 *
 * @param {string|array} columnOrColumns - Specifies the column name (a string) or columns (array of column names) to drop.
 */
DataFrame.prototype.dropSeries = function (columnOrColumns) {

	var self = this;

	if (!Object.isArray(columnOrColumns)) {
		assert.isString(columnOrColumns, "'DataFrame.dropSeries' expected either a string or an array or strings.");

		columnOrColumns = [columnOrColumns]; // Convert to array for coding convenience.
	}

	var columnNames = self.getColumnNames().slice(0); // Clone array.
	var newColumnNames = E.from(columnNames)
		.where(function (columnName) {
			return !E.from(columnOrColumns).contains(columnName);
		})
		.toArray();

	return new DataFrame({
		columnNames: newColumnNames,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					var row = extend({}, pair[1]);
					columnOrColumns.forEach(function (columnName) {
						delete row[columnName];
					});
					return [
						pair[0],
						row
					];					
				}
			);
		},
	});
};

/**
 * Create a new data frame with only the requested column or columns dropped, other columns are dropped.
 *
 * @param {string|array} columnOrColumns - Specifies the column name (a string) or columns (array of column names) to keep.
 */
DataFrame.prototype.keepSeries = function (columnOrColumns) {

	var self = this;

	if (!Object.isArray(columnOrColumns)) {
		assert.isString(columnOrColumns, "'DataFrame.keepSeries' expected either a string or an array or strings.");

		columnOrColumns = [columnOrColumns]; // Convert to array for coding convenience.
	}

	return new DataFrame({
		columnNames: columnOrColumns,
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair) {
					var row = extend({}, pair[1]);
					Object.keys(row).forEach(function (fieldName) {
						if (!E.from(columnOrColumns).contains(fieldName)) {
							delete row[fieldName];
						}
					});
					return [
						pair[0],
						row
					];					
				}
			);
		},
	});
};
/**
 * Create a new data frame with an additional column specified by the passed-in series.
 *
 * @param {string} columnName - The name of the column to add or replace.
 * @param {array|column} data - Array of data or column that contains data.
 */
DataFrame.prototype.setSeries = function (columnName, data) {

	assert.isString(columnName, "Expected 'columnName' parameter to 'setSeries' function to be a string that specifies the column to set or replace.");

	var self = this;

	if (Object.isFunction(data)) {
		data = E.from(self.toPairs()) //todo: make this lazy
			.select(function (pair) {
				return data(pair[1], pair[0]);
			})
			.toArray();
	}
	else if (!Object.isArray(data)) {
		assert.isObject(data, "Expected 'data' parameter to 'setSeries' to be either an array or a series object.");
		assert.isFunction(data.reindex, "Expected 'data' parameter to 'setSeries' to have a 'reindex' function that allows the column to be reindexed.");

		data = data.reindex(self.getIndex()).toValues();
	}

	//todo: overview and improve the way this works.

	var columnIndex = self.getColumnIndex(columnName);
	if (columnIndex < 0) {		
		// Add new column.
		return new DataFrame({
			columnNames: self.getColumnNames().concat([columnName]),
			rows: function () {
				return new ArrayIterator(
					E.from(self.toValues())
						.select(function (row, rowIndex) {
							return row.concat([data[rowIndex]]);
						})
						.toArray()
				);
			},
			index: self.getIndex(),
		});
	}
	else {
		// Replace existing column.
		return new DataFrame({
			columnNames: E.from(self.getColumnNames())
				.select(function (thisColumnName, thisColumnIndex) {
					if (thisColumnIndex === columnIndex) {
						return columnName;
					}
					else { 
						return thisColumnName;
					}
				})
				.toArray(),
			rows: function () {
				return new ArrayIterator(
					E.from(self.toValues())
						.select(function (row, rowIndex) {
							return E.from(row)
								.select(function (column, thisColumnIndex) {
									if (thisColumnIndex === columnIndex) {
										return data[rowIndex];
									}
									else {
										return column;
									}
								})
								.toArray();
						})
						.toArray()
				);
			},
			index: self.getIndex(),
		});
	}
};

/**
 * Create a new data-frame from a slice of rows.
 *
 * @param {int|function} startIndexOrStartPredicate - Index where the slice starts or a predicate function that determines where the slice starts.
 * @param {int|function} endIndexOrEndPredicate - Marks the end of the slice, one row past the last row to include. Or a predicate function that determines when the slice has ended.
 * @param {function} [predicate] - Optional predicate to compare index against start/end index. Return true to start or stop the slice.
 */
DataFrame.prototype.slice = function (startIndexOrStartPredicate, endIndexOrEndPredicate, predicate) {

	var self = this;

	var startIndex;
	var endIndex;
	var startPredicate = null;
	var endPredicate = null;

	if (predicate) {
		assert.isFunction(predicate, "Expected 'predicate' parameter to slice function to be function.");
	}

	if (Object.isFunction(startIndexOrStartPredicate)) {
		startPredicate = startIndexOrStartPredicate;
	}
	else {
		startIndex = startIndexOrStartPredicate;
		startPredicate = function (value) {
				return predicate && predicate(value, startIndex) || value < startIndex;
			};
	}

	if (Object.isFunction(endIndexOrEndPredicate)) {
		endPredicate = endIndexOrEndPredicate;
	}
	else {
		endIndex = endIndexOrEndPredicate;
		endPredicate = function (value) {
				return predicate && predicate(value, endIndex) || value < endIndex;
			};
	}

	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new TakeWhileIterator(
				new SkipWhileIterator(
					self.getIterator(),
					function (pair) {
						return startPredicate(pair[0]); // Check index for start condition.
					}
				),
				function (pair) {
					return endPredicate(pair[0]); // Check index for end condition.
				}
			);
		},		
	});

};

/**
 * Set a named column as the index of the data-frame.
 *
 * @param {string} columnName - Name or index of the column to set as the index.
 */
DataFrame.prototype.setIndex = function (columnName) {

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new PairIterator(
				new SelectIterator(
					self.getSeries(columnName).getIterator(),
					function (pair) {
						return pair[1];
					}
				),				
				new SelectIterator(
					self.getIterator(),
					function (pair) {
						return pair[1];
					}
				)
			);
		},
	});
};

/**
 * Reset the index of the data frame back to the default sequential integer index.
 */
DataFrame.prototype.resetIndex = function () {

	var self = this;
	return new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new SelectIterator(
				self.getIterator(),
				function (pair, i) {
					return [i, pair[1]];
				}
			);
		},
	});
};

/** 
 * Format the data frame for display as a string.
 */
DataFrame.prototype.toString = function () {

	var self = this;
	var Table = require('easy-table');

	var index = self.getIndex().toValues();
	var header = ["__index__"].concat(self.getColumnNames());
	var rows = E.from(self.toValues())
		.select(function (row, rowIndex) { 
			return [index[rowIndex]].concat(row);
		})
		.toArray()

	var t = new Table();
	rows.forEach(function (row, rowIndex) {
		row.forEach(function (cell, cellIndex) {
			t.cell(header[cellIndex], cell);
		});
		t.newRow();
	});

	return t.toString();
};

/**
 * Parse a column with string values to a column with int values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 */
DataFrame.prototype.parseInts = function (columnNameOrNames) {

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseInts());
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseInts());
	}
};

/**
 * Parse a column with string values to a column with float values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 */
DataFrame.prototype.parseFloats = function (columnNameOrNames) {

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseFloats());
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseFloats());
	}
};

/**
 * Parse a column with string values to a column with date values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to parse.
 * @param {string} [formatString] - Optional formatting string for dates.
 */
DataFrame.prototype.parseDates = function (columnNameOrNames, formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).parseDates(formatString));
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).parseDates(formatString));
	}
};

/**
 * Convert a column of values of different types to a column of string values.
 *
 * @param {string|array} columnNameOrNames - Specifies the column name or array of column names to convert to strings.
 * @param {string} [formatString] - Optional formatting string for dates.
 */
DataFrame.prototype.toStrings = function (columnNameOrNames, formatString) {

	if (formatString) {
		assert.isString(formatString, "Expected optional 'formatString' parameter to parseDates to be a string (if specified).");
	}

	var self = this;
	if (Object.isArray(columnNameOrNames)) {
		return E.from(columnNameOrNames)
			.aggregate(self, function (self, columnName) {
				return self.setSeries(columnName, self.getSeries(columnName).toStrings(formatString));
			});
	}
	else {
		return self.setSeries(columnNameOrNames, self.getSeries(columnNameOrNames).toStrings(formatString));
	}
};

/**
 * Detect actual types and their frequencies contained within columns in the data frame.
 */
DataFrame.prototype.detectTypes = function () {

	var self = this;

	var dataFrames = E.from(self.getColumns())
		.select(function (column) {
			var series = column.series;
			var numValues = series.toValues().length;
			//todo: broad-cast column
			var newSeries = new Series({
				values: E.range(0, numValues)
					.select(function () { 
						return column.name; 
					})
					.toArray()
			});
			return column.series
				.detectTypes()
				.setSeries('Column', newSeries);
		})
		.toArray();
	return concatDataFrames(dataFrames).resetIndex();
};

/**
 * Detect values and their frequencies contained within columns in the data frame.
 */
DataFrame.prototype.detectValues = function () {

	var self = this;

	var dataFrames = E.from(self.getColumns())
		.select(function (column) {
			var numValues = column.series.toValues().length;
			//todo: broad-cast column
			var newSeries = new Series({
				values: E.range(0, numValues)
					.select(function () { 
						return column.name 
					})
					.toArray()
			});
			return column.series.detectValues().setSeries('Column', newSeries);
		})
		.toArray();
	return concatDataFrames(dataFrames).resetIndex();
};
/**
 * Produces a new data frame with all string values truncated to the requested maximum length.
 *
 * @param {int} maxLength - The maximum length of the string values after truncation.
 */
DataFrame.prototype.truncateStrings = function (maxLength) {
	assert.isNumber(maxLength, "Expected 'maxLength' parameter to 'truncateStrings' to be an integer.");

	var self = this;
	var truncatedValues = E.from(self.toValues()) //todo: make this function lazy.
		.select(function (row) {
			return E.from(row)
				.select(function (value) {
					if (Object.isString(value)) {
						if (value.length > maxLength) {
							return value.substring(0, maxLength);
						}
					}

					return value;
				})
				.toArray();
		})
		.toArray();

	return new DataFrame({
			columnNames: self.getColumnNames(),
			rows: truncatedValues,
		});
};

/**
 * Create a new data frame with columns reordered.
 * New column names create new columns (with undefined values), omitting existing column names causes those columns to be dropped.
 * 
 * @param {array} columnNames - The new order for columns. 
 */
DataFrame.prototype.remapColumns = function (columnNames) {

	assert.isArray(columnNames, "Expected parameter 'columnNames' to remapColumns to be an array with column names.");

	columnNames.forEach(function (columnName) {
		assert.isString(columnName, "Expected parameter 'columnNames' to remapColumns to be an array with column names.");
	});

	var self = this;

	return new DataFrame({
		columnNames: columnNames,
		rows: function () { //todo: make this properly lazy.
			return new ArrayIterator(
				E.from(self.toValues())
					.select(function (row) {
						return E.from(columnNames)
							.select(function (columnName) {
								var columnIndex = self.getColumnIndex(columnName);
								if (columnIndex >= 0) {
									return row[columnIndex];
								}
								else { 
									// Column doesn't exist.
									return undefined;
								}
							})
							.toArray();
					})
					.toArray()
			);
		},
	});
};

/**
 * Create a new data-frame with renamed series.
 *
 * @param {array|object} newColumnNames|columnsMap - Array of strings, with an element for each existing column that specifies the new name of that column. Or, a hash that maps old column name to new column name.
 */
DataFrame.prototype.renameSeries = function (newColumnNames) {

	var self = this;

	if (Object.isObject(newColumnNames)) {
		var self = this;
		var renamedColumns = self.getColumnNames().slice(0); // Clone array.

		Object.keys(newColumnNames).forEach(function (existingColumnName, columnIndex) {
			var columnIndex = self.getColumnIndex(existingColumnName);
			if (columnIndex === -1) {
				return; // No column to be renamed.
			}

			renamedColumns[columnIndex] = newColumnNames[existingColumnName];
		});

		return self.renameSeries(renamedColumns);
	}
	else {
		var existingColumns = self.getColumnNames();
		var numExistingColumns = existingColumns.length;

		assert.isArray(newColumnNames, "Expected parameter 'newColumnNames' to renameColumns to be an array with column names.");
		assert(newColumnNames.length === numExistingColumns, "Expected 'newColumnNames' array to have an element for each existing column. There are " + numExistingColumns + "existing columns.");

		return new DataFrame({
			columnNames: newColumnNames,
			iterable: function () {
				var columnMap = E.from(existingColumns)
					.zip(newColumnNames, function (oldName, newName) {
						return [oldName, newName];
					})
					.toArray();

				return new SelectIterator(
					self.getIterator(),
					function (pair) {
						return [
							pair[0],
							E.from(columnMap).toObject(
								function (remap) {
									return remap[1];
								},
								function (remap) {
									return pair[1][remap[0]];								
								}
							)
						];
					}
				);
			},
		});
	}
};

/**
 * Bake the data frame to an array of rows.
 */
DataFrame.prototype.toValues = function () {

	var self = this;

	var iterator = self.getIterator();
	validateIterator(iterator);

	var values = [];
	var columnNames = self.getColumnNames();

	while (iterator.moveNext()) {
		var curRow = iterator.getCurrent()[1];

		var asArray = [];
		for (var columnIndex = 0; columnIndex < columnNames.length; ++columnIndex) {
			asArray.push(curRow[columnNames[columnIndex]]);
		}

		values.push(asArray);
	}

	return values;
};

/**
 * Bake the data frame to an array of JavaScript objects.
 */
DataFrame.prototype.toObjects = function () {

	var self = this;

	var iterator = self.getIterator();
	validateIterator(iterator);

	var objects = [];

	while (iterator.moveNext()) {
		objects.push(iterator.getCurrent()[1]); // Extract values.
	}

	return objects;
};

/**
 * Serialize the data frame to JSON.
 */
DataFrame.prototype.toJSON = function () {
	var self = this;
	return JSON.stringify(self.toObjects(), null, 4);
};

/**
 * Serialize the data frame to CSV.
 */
DataFrame.prototype.toCSV = function () {

	var self = this;
	var data = [self.getColumnNames()].concat(self.toValues());
	return BabyParse.unparse(data);

	/*Old csv stringify.
	var header = self.getColumnNames().join(',');
	var rows = E.from(self.toValues())
			.select(function (row) {
				return row.join(',');
			})
			.select(function (col) { // Strip newlines... these don't work in CSV files.
				if (Object.isString(col)) { //todo: not necessar if all columns are converted to strings.
					return col.replace(/\r\n/g, ' ').replace(/\n/g, ' ');
				}
				else {
					return col;
				}
			})					
			.toArray();
	return [header].concat(rows).join('\r\n');	
	*/
};

/**
 * Forces lazy evaluation to complete and 'bakes' the data frame into memory.
 */
DataFrame.prototype.bake = function () {

	var self = this;
	if (self._baked) {
		// Already baked, just return self.
		return self;
	}

	var pairs = self.toPairs();
	var baked = new DataFrame({
		columnNames: self.getColumnNames(),
		iterable: function () {
			return new ArrayIterator(pairs);
		},
	});
	baked._baked = true;
	return baked;
};

{'fraction_non_alphanumeric': 0.1228735473012559, 'fraction_numerical': 0.0017043578299263931, 'mean_word_length': 3.4321797837684715, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6153308', 'n_tokens_mistral': 32867, 'n_tokens_neox': 27453, 'n_words': 8153}