package org.aksw.es.bsbmloader.reader;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

import org.apache.log4j.Logger;
import org.apache.metamodel.DataContext;
import org.apache.metamodel.data.DataSet;
import org.apache.metamodel.data.Row;
import org.apache.metamodel.query.Query;
import org.apache.metamodel.schema.Column;
import org.apache.metamodel.schema.Table;

import com.codahale.metrics.Gauge;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;

public class DataReader implements Runnable {
	private DataContext dataContext;
	private BlockingQueue<Row> queue = null;
	private Table table;
	private int offset = 0;
	private int limit = 0;
	private static org.apache.log4j.Logger log = Logger.getLogger(DataReader.class);
	private CountDownLatch latch;
	private final static int BORDER = 3000;
	private int numbers = 0;
	private boolean isFinish = false;
	private int key = 0;
	private int result = 0;
	private boolean horizontal = false;
	private String columnName;
	private Column hashedColumn;
	private final MetricRegistry metrics;
	private Meter requests;
	private int threads;

	public void setThreads(int threads) {
		this.threads = threads;
	}

	public DataReader(MetricRegistry metrics) {
		this.metrics = metrics;

	}

	public void setKey(int key) {
		this.key = key;
	}

	public void setResult(int result) {
		this.result = result;
	}

	public void setHorizontal(boolean horizontal) {
		this.horizontal = horizontal;
	}

	public void setColumnName(String columnName) {
		this.columnName = columnName;
	}

	public void setLatch(CountDownLatch latch) {
		this.latch = latch;
	}

	public void setOffset(int offset) {
		this.offset = offset;
	}

	public void setLimit(int limit) {
		this.limit = limit;
	}

	public void setDataContext(DataContext dataContext) {
		this.dataContext = dataContext;
	}

	public void setQueue(BlockingQueue<Row> queue) {
		this.queue = queue;
	}

	public void setTable(Table table) {
		this.table = table;
	}

	private void insertPosion() throws Exception {
		for (int i = 0; i < 2 * threads; i++) {
			queue.put(PosionRow.posionRow);
		}

	}

	private Query selectAll() {
		return dataContext.query().from(table).selectAll().offset(offset).limit(limit).toQuery();
	}

	private DataSet createDataSet() {

		if (numbers == 0 && !isFinish) {
			DataSet number = dataContext.query().from(table).selectCount().execute();
			number.next();
			Number n = (Number) number.getRow().getValue(0);
			numbers = n.intValue();
			limit = BORDER;
			number.close();
		}

		return dataContext.executeQuery(selectAll());

	}

	public void readData() throws Exception {

		table = dataContext.getTableByQualifiedLabel(table.getName());
		DataSet dataSet = createDataSet();

		while (numbers > 0) {
			insertRowIntoBlockingQueue(dataSet);
			dataSet.close();
			setNewOffset();
			dataSet = createDataSet();
		}

		insertPosion();
	}

	private void setNewOffset() {
		offset += BORDER;
		numbers -= BORDER;
		if (numbers < BORDER && numbers <= 0) {
			numbers = 0;
			isFinish = true;
		}
	}

	private void insertRowIntoBlockingQueue(DataSet dataset) throws InterruptedException {

		hashedColumn = table.getColumnByName(columnName);
		requests = metrics.meter("read thread");

		while (dataset.next()) {
			Row row = dataset.getRow();
			if (horizontal && isKey(row)) {
				queue.put(row);
				requests.mark();
			} else {
				queue.put(row);
				requests.mark();
			}

		}

	}

	private boolean isKey(Row row) {
		return ((Integer) row.getValue(hashedColumn) % key) == result;
	}

	public void run() {
		try {
			readData();
			latch.countDown();
		} catch (Exception e) {

			log.error(e);
		}

	}

}

@@ -3411,16 +3411,40 @@
  else %7B%0A
+%09%09%09    if(!horizontal)%7B%0A
 %09%09%09%09queu
@@ -3472,24 +3472,33 @@
 sts.mark();%0A
+%09%09%09    %7D%0A
 %09%09%09%7D%0A%0A%09%09%7D%0A%0A%09

Fix Error in datareader