Let's try this again... this time with code that actually compiles.

src/org/kered/dko/AbstractQuery.java

@@ -9,7 +9,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -224,14 +224,14 @@ public <S> List<S> asList(final Field<S> field) {
 
 	@Override
 	public Set<T> asSet() {
-		final Set<T> set = new HashSet<T>();
+		final Set<T> set = new LinkedHashSet<T>();
 		for (final T t : this) set.add(t);
 		return set;
 	}
 
 	@Override
 	public <S> Set<S> asSet(final Field<S> field) {
-		final Set<S> ret = new HashSet<S>();
+		final Set<S> ret = new LinkedHashSet<S>();
 		for (final S s : this.distinct().asIterableOf(field)) {
 			ret.add(s);
 		}

src/org/kered/dko/Condition.java

@@ -552,7 +552,7 @@ protected void getSQL(final StringBuffer sb, final List<Object> bindings, final
 				bindings.add(v2);
 			}
 			sb.append(cmp2);
-			if (v2 instanceof Field) {
+			if (v3 instanceof Field) {
 				sb.append(Util.derefField((Field)v3, context));
 			} else if (v3 instanceof Expression) {
 				((Expression)v3).__getSQL(sb, bindings, context);

src/org/kered/dko/Constants.java

@@ -288,6 +288,12 @@ String getDatabaseTableSeparator() {
 	 */
 	public static final String PROPERTY_PERSISTENCE_DB = "org.kered.dko.persistence_db";
 
+	/**
+	 * A Java property (boolean) that controls whether DKO uses the persistence database or not.
+	 * By default: true
+	 */
+	public static final String PROPERTY_USE_PERSISTENCE_DB = "org.kered.dko.use_persistence_db";
+
 	static enum JOIN_TYPE {
 
 		LEFT("left join"),

src/org/kered/dko/Context.java

@@ -156,7 +156,7 @@ static boolean selectOptimizationsEnabled() {
 		String prop = System.getProperty(Constants.PROPERTY_OPTIMIZE_SELECT_FIELDS);
 		if (prop == null) prop = System.getProperty(Constants.PROPERTY_OPTIMIZE_SELECT_FIELDS_OLD);
 		if (prop != null) return Util.truthy(prop);
-		return true;
+		return false;
 	}
 
 	/**
@@ -330,7 +330,7 @@ public Undoer overrideDatabaseName(final DataSource ds, final String originalDat
 		final UUID uuid = UUID.randomUUID();
 		map.put(uuid, newDatabaseName);
 		final Map<UUID, String> map2 = map;
-		return new Undoer() {
+		return new Undoer("OverrideDB:" + originalDatabaseName + "=>" + newDatabaseName) {
 			@Override
 			public void undo() {
 				map2.remove(uuid);
@@ -348,7 +348,7 @@ public void undo() {
 	public Undoer setDataSource(final DataSource ds) {
 		final UUID uuid = UUID.randomUUID();
 		defaultDataSource.put(uuid, ds);
-		return new Undoer() {
+		return new Undoer("setDataSource") {
 			@Override
 			public void undo() {
 				defaultDataSource.remove(uuid);
@@ -374,7 +374,7 @@ public Undoer setDataSource(final Package pkg, final DataSource ds) {
 		final UUID uuid = UUID.randomUUID();
 		map.put(uuid, ds);
 		final Map<UUID, DataSource> map2 = map;
-		return new Undoer() {
+		return new Undoer("setDataSource:pkgName=" + pkg.getName()) {
 			@Override
 			public void undo() {
 				map2.remove(uuid);
@@ -397,7 +397,7 @@ public Undoer setDataSource(final Class<? extends Table> cls, final DataSource d
 		final UUID uuid = UUID.randomUUID();
 		map.put(uuid, ds);
 		final Map<UUID, DataSource> map2 = map;
-		return new Undoer() {
+		return new Undoer("setDataSource:class=" + cls.getName()) {
 			@Override
 			public void undo() {
 				map2.remove(uuid);
@@ -414,7 +414,7 @@ public void undo() {
 	public Undoer enableUsageWarnings(final boolean enable) {
 		final UUID uuid = UUID.randomUUID();
 		enableUsageWarnings.put(uuid, enable);
-		return new Undoer() {
+		return new Undoer("enableUsageWarnings:" + enable) {
 			@Override
 			public void undo() {
 				enableUsageWarnings.remove(uuid);
@@ -430,7 +430,7 @@ public void undo() {
 	public Undoer enableSelectOptimizations(final boolean enable) {
 		final UUID uuid = UUID.randomUUID();
 		enableSelectOptimizations.put(uuid, enable);
-		return new Undoer() {
+		return new Undoer("enableSelectOptimizations:" + enable) {
 			@Override
 			public void undo() {
 				enableSelectOptimizations.remove(uuid);
@@ -445,7 +445,8 @@ public void undo() {
 	 * @author Derek Anderson
 	 */
 	public static abstract class Undoer {
-		private boolean autoRevoke = true;
+		public final String tag;
+		private boolean autoRevoke = false;
 		public abstract void undo();
 		public boolean willAutoUndo() {
 			return autoRevoke ;
@@ -455,7 +456,10 @@ public Undoer setAutoUndo(final boolean v) {
 			return this;
 		}
 
-		private Undoer() {}
+		private Undoer(String tag) {
+			this.tag = tag;
+		}
+
 		protected void finalize() {
 			if (autoRevoke) undo();
 		}

src/org/kered/dko/DBQuery.java

@@ -55,6 +55,7 @@ class DBQuery<T extends Table> extends AbstractQuery<T> {
 	private Set<Field<?>> groupBySet = null;
 	private List<Expression.OrderBy<?>> orderByExpressions = null;
 	long top = 0;
+	long offset = 0;
 	private Map<Field<?>,Object> data = null;
 	boolean distinct = false;
 	DataSource ds = null;
@@ -120,6 +121,7 @@ private void copy(final DBQuery<T> q) {
 			orderByExpressions.addAll(q.orderByExpressions);
 		}
 		top = q.top;
+		offset = q.offset;
 		if (q.data!=null) {
 			data = new HashMap<Field<?>,Object>();
 			data.putAll(q.data);
@@ -675,13 +677,22 @@ public Query<T> limit(final long i) {
 		return q;
 	}
 
+	@Override
+	public Query<T> offset(final long o) {
+		final DBQuery<T> q = new DBQuery<T>(this);
+		q.offset = o;
+		return q;
+	}
+
 	private String genTableName(final Class<? extends Table> table, final Collection<String> tableNames) {
 		final String name = Util.getTableName(table);
 		String base = "";
 		for (final String s : name.split("_")) base += s.length() > 0 ? s.substring(0, 1) : "";
-		String proposed = null;
-		int i = 1;
-		while (tableNames.contains((proposed = base + (i==1 ? "" : String.valueOf(i))))) ++i;
+		String proposed = base + 0;
+		int i = 0;
+		while (tableNames.contains(proposed)) {
+			proposed = base + String.valueOf(++i);
+                }
 		return proposed;
 	}
 
@@ -1543,6 +1554,7 @@ public int hashCode() {
 		result = prime * result
 				+ ((tableInfos == null) ? 0 : tableInfos.hashCode());
 		result = prime * result + (int)top;
+		result = prime * result + (int)offset;
 		result = prime * result
 				+ ((usedTableNames == null) ? 0 : usedTableNames.hashCode());
 		return result;

src/org/kered/dko/DBRowIterator.java

@@ -151,6 +151,7 @@ protected Tuple2<String,List<Object>> getSQL(final SqlContext context) {
 		}
 
 		final List<Expression.OrderBy<?>> obes = query.getOrderByExpressions();
+		boolean orderByApplied = false;
 		if (!context.inInnerQuery() && obes!=null && !obes.isEmpty()) {
 			sb.append(" order by ");
 			final String[] tmp = new String[obes.size()];
@@ -170,14 +171,12 @@ protected Tuple2<String,List<Object>> getSQL(final SqlContext context) {
 					sb2.append(((SQLFunction.OrderBySQLFunction)obe).direction==DESCENDING ? " DESC" : " ASC");
 					tmp[i] = sb2.toString();
 				}
+				orderByApplied = true;
 			}
 			sb.append(Util.join(", ", tmp));
 		}
 
-		if (context.dbType!=DB_TYPE.SQLSERVER && context.dbType!=DB_TYPE.ORACLE && context.dbType!=DB_TYPE.DERBY && 
-				query.top>0 && query.joinsToMany.size()==0) {
-			sb.append(" limit ").append(query.top);
-		}
+		appendLimitAndOffset(query, context, sb, orderByApplied);
 
 		if (selectedFields.length > context.maxFields) {
 			Util.log(sb.toString(), null);
@@ -190,12 +189,37 @@ protected Tuple2<String,List<Object>> getSQL(final SqlContext context) {
 		final String sql = sb.toString();
 		return new Tuple2<String,List<Object>>(sql, bindings);
 	}
+	private static <T extends Table>  void appendLimitAndOffset(DBQuery<T> query, SqlContext context,
+		StringBuffer sb, boolean orderByApplied) {
+
+		if (query.joinsToMany.isEmpty()) {
+			if (context.dbType != DB_TYPE.SQLSERVER && context.dbType!=DB_TYPE.ORACLE && context.dbType!=DB_TYPE.DERBY) {
+				if (query.top > 0) {
+					sb.append(" limit ").append(query.top);
+				}
+				if (query.offset > 0) {
+					sb.append(" offset ").append(query.offset);
+				}
+			}
+			else if (query.offset > 0 && context.dbType == DB_TYPE.SQLSERVER) {
+				// this syntax is more complicated than select top(). If you're restricting the number of rows
+				// you need to provide an offset. This also requires the use of the order by clause. Trap this now.
+				if (!orderByApplied) {
+					throw new RuntimeException("offset requires order by clause on SQL Server.");
+				}
+				sb.append(" offset ").append(Math.max(0, query.offset)).append(" rows");
+				if (query.top > 0) {
+					sb.append(" fetch next ").append(query.top).append(" rows only");
+				}
+			}
+		}
+	}
 
 	private static <T extends Table> void appendSelectFromWhere(DBQuery<T> query, final Expression.Select<?>[] selectedBoundFields,
 			final SqlContext context, final StringBuffer sb, List<Object> bindings) {
 		sb.append("select ");
 		if (query.distinct) sb.append("distinct ");
-		if (context.dbType==DB_TYPE.SQLSERVER && query.top>0 && query.joinsToMany.size()==0) {
+		if (context.dbType==DB_TYPE.SQLSERVER && query.top>0 && query.offset<=0 && query.joinsToMany.size()==0) {
 			sb.append(" top ").append(query.top).append(" ");
 		}
 		if (query.globallyAppliedSelectFunction == null) {

src/org/kered/dko/Field.java

@@ -1120,7 +1120,7 @@ boolean isBound() {
 		return this.boundTable!=null || this.boundTableInfo!=null;
 	}
 
-	boolean sameField(final Expression.Select<?> other) {
+	public boolean sameField(final Expression.Select<?> other) {
 		if (this == other)
 			return true;
 		if (other == null)

src/org/kered/dko/FilteringQuery.java

@@ -18,12 +18,14 @@ class FilteringQuery<T extends Table> extends AbstractQuery<T> implements Matryo
 	private Query<T> q;
 	private Condition condition = null;
 	private long top = -1;
+	private long offset = -1;
 
 	FilteringQuery(final FilteringQuery<T> src) {
 		super(src.getType());
 		this.q = src.q;
 		this.condition = src.condition;
 		this.top = src.top;
+		this.offset = src.offset;
 	}
 
 	FilteringQuery(final Query<T> q, final Condition... conditions) {
@@ -62,6 +64,14 @@ public Query<T> limit(long n) {
 		return ret;
 	}
 
+	@Override
+	public Query<T> offset(long m) {
+		FilteringQuery<T> ret = new FilteringQuery<T>(this);
+		ret.q = ret.q.offset(-1);
+		ret.offset = m;
+		return ret;
+	}
+
 	@Override
 	public Query<T> distinct() {
 		FilteringQuery<T> ret = new FilteringQuery<T>(this);
@@ -179,9 +189,16 @@ public Iterator<T> iterator() {
 			final Iterator<T> i = q.iterator();
 			T next = null;
 			int count = 0;
+			long skip = offset;
 
 			@Override
 			public boolean hasNext() {
+				while(skip > 0 && i.hasNext()) {
+					T candidate = i.next();
+					if (condition.matches(candidate)) {
+						skip--;
+					}	
+				}
 				if (next!=null) return true;
 				if (top > -1 && count >= top) return false;
 				while (i.hasNext()) {

src/org/kered/dko/InMemoryQuery.java

@@ -141,6 +141,17 @@ public Query<T> limit(final long n) {
 		return q;
 	}
 
+	@Override
+	public Query<T> offset(long m) {
+		if (!loaded) load();
+		final InMemoryQuery<T> q = new InMemoryQuery<T>(this);
+		q.cache = new ArrayList<T>();
+		if (cache.size() > m) {
+			q.cache.addAll(cache.subList((int)m, cache.size()));
+		}
+		return q;
+	}
+
 	@Override
 	public Query<T> distinct() {
 		// TODO Auto-generated method stub

src/org/kered/dko/LocalJoin.java

@@ -39,6 +39,7 @@ class LocalJoin<T extends Table> extends AbstractQuery<T> {
 	private Condition joinCondition;
 	private final JOIN_TYPE joinType;
 	private long limit = -1;
+	private long offset = -1;
 
 	private transient List<Field<?>> selectFields;
 
@@ -50,6 +51,7 @@ public LocalJoin(final LocalJoin<T> q) {
 		qR = q.qR;
 		joinCondition = q.joinCondition;
 		limit = q.limit;
+		offset = q.offset;
 	}
 
 	public LocalJoin(final JOIN_TYPE joinType, final Class<? extends Table> type, final Query<? extends Table> q, final Class<? extends Table> t, final Condition on) {
@@ -104,6 +106,16 @@ public Query<T> limit(final long n) {
 		return q;
 	}
 
+	@Override
+	public Query<T> offset(long m) {
+		final LocalJoin<T> q = new LocalJoin<T>(this);
+		q.offset = m;
+		if (joinType==JOIN_TYPE.LEFT || joinType==JOIN_TYPE.OUTER) q.qL = q.qL.offset(m);
+		if (joinType==JOIN_TYPE.RIGHT || joinType==JOIN_TYPE.OUTER) q.qR = q.qR.offset(m);
+		return q;
+	}
+
+
 	@Override
 	public Query<T> distinct() {
 		// TODO Auto-generated method stub

src/org/kered/dko/Query.java

@@ -141,6 +141,15 @@ public interface Query<T extends Table> extends Iterable<T> {
 	 */
 	public Query<T> limit(long n);
 
+	/**
+	 * Returns rows beginning at offset m as defined by the ordering of the query.
+	 *
+	 * @param m
+	 * @return
+	 */
+	public Query<T> offset(long m);
+
+
 	/**
 	 * Sets the distinct keyword in the select statement.
 	 * @return
@@ -284,8 +293,8 @@ public interface Query<T extends Table> extends Iterable<T> {
 	/**
 	 * Executes and update statement populated w/ data from .where() and .set().
 	 * Example:  SomeClass.ALL.set(SomeClass.SOME_FIELD, "xyz")
-	 *                        .where(SomeClass.SOME_FIELD.eq("abc"))
-	 *                        .update();
+	 *			.where(SomeClass.SOME_FIELD.eq("abc"))
+	 *			.update();
 	 * @return
 	 * @throws SQLException
 	 */

src/org/kered/dko/SelectFromOAI.java

@@ -8,6 +8,7 @@
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.logging.Logger;
 
 import javax.sql.DataSource;
@@ -282,6 +283,9 @@ public synchronized void close() {
 
 	@Override
 	public T next() {
+		if (!hasNext()) {
+			throw new NoSuchElementException();
+		}
 		final T t = next;
 		next = null;
 		++count;