package acstest;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.function.*;
import java.util.function.Predicate;

import org.junit.*;

import Zql.ParseException;
import Zql.ZQuery;
import Zql.ZStatement;
import Zql.ZqlParser;
import simpledb.*;

public class DirtyTest {
	private static class SmudgePredicate extends Operator {
		private final DbIterator child;
		private final int dirtyIndex;
		private final double dirtyPerc;
		private final Predicate<Tuple> shouldDirty;
		
		private final Random rand = new Random(0);
		
		private static final Field missing = new IntField();
		
		public SmudgePredicate(DbIterator child, int dirtyIndex, double dirtyPerc, Predicate<Tuple> shouldDirty) {
			this.child = child;
			this.dirtyIndex = dirtyIndex;
			this.dirtyPerc = dirtyPerc;
			this.shouldDirty = shouldDirty;
		}
		
		public SmudgePredicate(DbIterator child, QuantifiedName dirtyName, double dirtyPerc, Predicate<Tuple> shouldDirty) {
			this(child, child.getTupleDesc().fieldNameToIndex(dirtyName), dirtyPerc, shouldDirty);
		}
		
		@Override
		public void rewind() throws DbException, TransactionAbortedException {
			child.rewind();
		}
		
		@Override
	    public void open() throws DbException, NoSuchElementException,
	            TransactionAbortedException {
	    	super.open();
	        child.open();
	    }
		
		@Override
	    public void close() {
	    	super.close();
	    	child.close();
	    }

		@Override
		protected Tuple fetchNext() throws DbException, TransactionAbortedException {
			while (child.hasNext()) {
	        	Tuple t = child.next();
	        	if (rand.nextDouble() < dirtyPerc) {
	        		if (shouldDirty.test(t)) {
	        			Tuple tt = new Tuple(t);
	        			tt.setField(dirtyIndex, missing);
	        			return tt;
	        		}
	        	} else {
	        		return t;
	        	}
	        }
	        return null;
		}

		@Override
		public DbIterator[] getChildren() {
			throw new RuntimeException("Not implemented.");
		}

		@Override
		public void setChildren(DbIterator[] children) {
			throw new RuntimeException("Not implemented.");
		}

		@Override
		public TupleDesc getTupleDesc() {
			return child.getTupleDesc();
		}	
	}
	
	private static class SmudgeRandom extends SmudgePredicate {
		public SmudgeRandom(DbIterator child, QuantifiedName dirtyName, double d) {
			super(child, dirtyName, d, x -> true);
		}
	}
	
	private double error(DbIterator expected, DbIterator actual) throws DbException, TransactionAbortedException {
		double err = 0.0;
		while(expected.hasNext()) {
			try {
				if (!actual.hasNext()) {
					throw new RuntimeException("BUG: Iterators have different lengths.");
				}
				err += expected.next().error(actual.next());
			} catch (NoSuchElementException e) {
				throw new RuntimeException("BUG: No more tuples.");
			}
		}
		return err;
	}
	
	@Rule public AcsTestRule testDb = new AcsTestRule();
	
	private static DbIterator planQuery(String query, Function<Void, LogicalPlan> planFactory) throws ParseException, TransactionAbortedException, DbException, IOException, ParsingException {
		ZqlParser p = new ZqlParser(new ByteArrayInputStream(query.getBytes("UTF-8")));
		ZStatement s = p.readStatement();
		Parser pp = new Parser(planFactory);
		return pp.handleQueryStatement((ZQuery)s, new TransactionId()).getPhysicalPlan();
	}
	
	@Test 
	public void BldTest() throws DbException, TransactionAbortedException, IOException, ParseException, ParsingException {
		final Function<String, String> query = 
				tbl -> String.format("SELECT BLD as units_in_structure, AVG(BDSP) as estimate FROM %s GROUP BY BLD", tbl);
		
		System.err.print("Creating new table with dirty data... ");
		TransactionId tid = new TransactionId();
		Database.getCatalog().addTable("acsd", "", AcsTestRule.schema);
		int oldTblId = Database.getCatalog().getTableId("acs"),
				newTblId = Database.getCatalog().getTableId("acsd");
		new Insert(tid, new SmudgeRandom(new SeqScan(tid, oldTblId), new QuantifiedName("acs", "BDSP"), 0.1), newTblId).next();
		System.err.println("done.");
		
		System.err.print("Planning queries... ");
		DbIterator clean = planQuery(query.apply("acs"), x -> new LogicalPlan()), 
				imputedDirty = planQuery(query.apply("acsd"), x -> new ImputedLogicalPlan(0.5)), 
				dirty = planQuery(query.apply("acs"), x -> new LogicalPlan());
		System.err.println("done.");
		
		double imputeErr = error(clean, imputedDirty);
		clean.rewind();
		double baseErr = error(clean, dirty);
		Assert.assertTrue("Base error should be higher than imputed error.", baseErr >= imputeErr);
	}
}

@@ -80,53 +80,8 @@
 on;%0A
-import java.io.UnsupportedEncodingException;%0A
 impo
@@ -377,16 +377,86 @@
 rator %7B%0A
+%09%09private static final long serialVersionUID = -5585085012061351883L;%0A
 %09%09privat
@@ -2357,16 +2357,87 @@
 icate %7B%0A
+%09%09private static final long serialVersionUID = -7180182613916428629L;%0A%0A
 %09%09public

fix eclipse warnings