/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.gora.accumulo.store;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;

import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IsolatedScanner;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.RowIterator;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.mock.MockConnector;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.mock.MockTabletLocator;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.SortedKeyIterator;
import org.apache.accumulo.core.iterators.user.TimestampFilter;
import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.security.thrift.AuthInfo;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.util.Utf8;
import org.apache.gora.accumulo.encoders.Encoder;
import org.apache.gora.accumulo.query.AccumuloQuery;
import org.apache.gora.accumulo.query.AccumuloResult;
import org.apache.gora.persistency.ListGenericArray;
import org.apache.gora.persistency.Persistent;
import org.apache.gora.persistency.State;
import org.apache.gora.persistency.StateManager;
import org.apache.gora.persistency.StatefulHashMap;
import org.apache.gora.persistency.StatefulMap;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
import org.apache.gora.query.impl.PartitionQueryImpl;
import org.apache.gora.store.DataStoreFactory;
import org.apache.gora.store.impl.DataStoreBase;
import org.apache.gora.util.AvroUtils;
import org.apache.hadoop.io.Text;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;

/**
 * 
 */
public class AccumuloStore<K,T extends Persistent> extends DataStoreBase<K,T> {
  
  protected static final String MOCK_PROPERTY = "accumulo.mock";
  protected static final String INSTANCE_NAME_PROPERTY = "accumulo.instance";
  protected static final String ZOOKEEPERS_NAME_PROPERTY = "accumulo.zookeepers";
  protected static final String USERNAME_PROPERTY = "accumulo.user";
  protected static final String PASSWORD_PROPERTY = "accumulo.password";
  protected static final String DEFAULT_MAPPING_FILE = "gora-accumulo-mapping.xml";

  private Connector conn;
  private BatchWriter batchWriter;
  private AccumuloMapping mapping;
  private AuthInfo authInfo;
  private Encoder encoder;
  
  public Object fromBytes(Schema schema, byte data[]) {
    return fromBytes(encoder, schema, data);
  }

  public static Object fromBytes(Encoder encoder, Schema schema, byte data[]) {
    switch (schema.getType()) {
      case BOOLEAN:
        return encoder.decodeBoolean(data);
      case DOUBLE:
        return encoder.decodeDouble(data);
      case FLOAT:
        return encoder.decodeFloat(data);
      case INT:
        return encoder.decodeInt(data);
      case LONG:
        return encoder.decodeLong(data);
      case STRING:
        return new Utf8(data);
      case BYTES:
        return ByteBuffer.wrap(data);
      case ENUM:
        return AvroUtils.getEnumValue(schema, encoder.decodeInt(data));
    }
    throw new IllegalArgumentException("Unknown type " + schema.getType());
    
  }

  public K fromBytes(Class<K> clazz, byte[] val) {
    return fromBytes(encoder, clazz, val);
  }

  @SuppressWarnings("unchecked")
  public static <K> K fromBytes(Encoder encoder, Class<K> clazz, byte[] val) {
    try {
      if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) {
        return (K) Byte.valueOf(encoder.decodeByte(val));
      } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) {
        return (K) Boolean.valueOf(encoder.decodeBoolean(val));
      } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) {
        return (K) Short.valueOf(encoder.decodeShort(val));
      } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) {
        return (K) Integer.valueOf(encoder.decodeInt(val));
      } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) {
        return (K) Long.valueOf(encoder.decodeLong(val));
      } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) {
        return (K) Float.valueOf(encoder.decodeFloat(val));
      } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) {
        return (K) Double.valueOf(encoder.decodeDouble(val));
      } else if (clazz.equals(String.class)) {
        return (K) new String(val, "UTF-8");
      } else if (clazz.equals(Utf8.class)) {
        return (K) new Utf8(val);
      }
      
      throw new IllegalArgumentException("Unknown type " + clazz.getName());
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }
  }

  private static byte[] copyIfNeeded(byte b[], int offset, int len) {
    if (len != b.length || offset != 0) {
      byte copy[] = new byte[len];
      System.arraycopy(b, offset, copy, 0, copy.length);
      b = copy;
    }
    return b;
  }

  public byte[] toBytes(Object o) {
    return toBytes(encoder, o);
  }
  
  public static byte[] toBytes(Encoder encoder, Object o) {
    
    try {
      if (o instanceof String) {
        return ((String) o).getBytes("UTF-8");
      } else if (o instanceof Utf8) {
        return copyIfNeeded(((Utf8) o).getBytes(), 0, ((Utf8) o).getLength());
      } else if (o instanceof ByteBuffer) {
        return copyIfNeeded(((ByteBuffer) o).array(), ((ByteBuffer) o).arrayOffset() + ((ByteBuffer) o).position(), ((ByteBuffer) o).remaining());
      } else if (o instanceof Long) {
        return encoder.encodeLong((Long) o);
      } else if (o instanceof Integer) {
        return encoder.encodeInt((Integer) o);
      } else if (o instanceof Short) {
        return encoder.encodeShort((Short) o);
      } else if (o instanceof Byte) {
        return encoder.encodeByte((Byte) o);
      } else if (o instanceof Boolean) {
        return encoder.encodeBoolean((Boolean) o);
      } else if (o instanceof Float) {
        return encoder.encodeFloat((Float) o);
      } else if (o instanceof Double) {
        return encoder.encodeDouble((Double) o);
      } else if (o instanceof Enum) {
        return encoder.encodeInt(((Enum) o).ordinal());
      }
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }
    
    throw new IllegalArgumentException("Uknown type " + o.getClass().getName());
  }

  private BatchWriter getBatchWriter() throws IOException {
    if (batchWriter == null)
      try {
        batchWriter = conn.createBatchWriter(mapping.tableName, 10000000, 60000l, 4);
      } catch (TableNotFoundException e) {
        throw new IOException(e);
      }
    return batchWriter;
  }

  @Override
  public void initialize(Class<K> keyClass, Class<T> persistentClass, Properties properties) throws IOException {
    super.initialize(keyClass, persistentClass, properties);

    String mock = DataStoreFactory.findProperty(properties, this, MOCK_PROPERTY, null);
    String mappingFile = DataStoreFactory.getMappingFile(properties, this, DEFAULT_MAPPING_FILE);
    String user = DataStoreFactory.findProperty(properties, this, USERNAME_PROPERTY, null);
    String password = DataStoreFactory.findProperty(properties, this, PASSWORD_PROPERTY, null);
    
    mapping = readMapping(mappingFile);

    if (mapping.encoder == null || mapping.encoder.equals("")) {
      encoder = new org.apache.gora.accumulo.encoders.BinaryEncoder();
    } else {
      try {
        encoder = (Encoder) getClass().getClassLoader().loadClass(mapping.encoder).newInstance();
      } catch (InstantiationException e) {
        throw new IOException(e);
      } catch (IllegalAccessException e) {
        throw new IOException(e);
      } catch (ClassNotFoundException e) {
        throw new IOException(e);
      }
    }

    try {
      if (mock == null || !mock.equals("true")) {
        String instance = DataStoreFactory.findProperty(properties, this, INSTANCE_NAME_PROPERTY, null);
        String zookeepers = DataStoreFactory.findProperty(properties, this, ZOOKEEPERS_NAME_PROPERTY, null);
        conn = new ZooKeeperInstance(instance, zookeepers).getConnector(user, password);
        authInfo = new AuthInfo(user, ByteBuffer.wrap(password.getBytes()), conn.getInstance().getInstanceID());
      } else {
        conn = new MockInstance().getConnector(user, password);
      }

      if (autoCreateSchema)
        createSchema();
    } catch (AccumuloException e) {
      throw new IOException(e);
    } catch (AccumuloSecurityException e) {
      throw new IOException(e);
    }
  }
  
  protected AccumuloMapping readMapping(String filename) throws IOException {
    try {
      
      AccumuloMapping mapping = new AccumuloMapping();

      DocumentBuilder db = DocumentBuilderFactory.newInstance().newDocumentBuilder();
      Document dom = db.parse(getClass().getClassLoader().getResourceAsStream(filename));
      
      Element root = dom.getDocumentElement();
      
      NodeList nl = root.getElementsByTagName("class");
      for (int i = 0; i < nl.getLength(); i++) {
        
        Element classElement = (Element) nl.item(i);
        if (classElement.getAttribute("keyClass").equals(keyClass.getCanonicalName())
            && classElement.getAttribute("name").equals(persistentClass.getCanonicalName())) {

          mapping.tableName = getSchemaName(classElement.getAttribute("table"), persistentClass);
          mapping.encoder = classElement.getAttribute("encoder");
          
          NodeList fields = classElement.getElementsByTagName("field");
          for (int j = 0; j < fields.getLength(); j++) {
            Element fieldElement = (Element) fields.item(j);

            String name = fieldElement.getAttribute("name");
            String family = fieldElement.getAttribute("family");
            String qualifier = fieldElement.getAttribute("qualifier");
            if (qualifier.equals(""))
              qualifier = null;

            Pair<Text,Text> col = new Pair<Text,Text>(new Text(family), qualifier == null ? null : new Text(qualifier));
            mapping.fieldMap.put(name, col);
            mapping.columnMap.put(col, name);
          }
        }

      }
      
      nl = root.getElementsByTagName("table");
      for (int i = 0; i < nl.getLength(); i++) {
        Element tableElement = (Element) nl.item(i);
        if (tableElement.getAttribute("name").equals(mapping.tableName)) {
          NodeList configs = tableElement.getElementsByTagName("config");
          for (int j = 0; j < configs.getLength(); j++) {
            Element configElement = (Element) configs.item(j);
            String key = configElement.getAttribute("key");
            String val = configElement.getAttribute("value");
            mapping.tableConfig.put(key, val);
          }
        }
      }

      return mapping;
    } catch (Exception ex) {
      throw new IOException(ex);
    }

  }
  
  @Override
  public String getSchemaName() {
    return mapping.tableName;
  }
  
  @Override
  public void createSchema() throws IOException {
    try {
      conn.tableOperations().create(mapping.tableName);
      Set<Entry<String,String>> es = mapping.tableConfig.entrySet();
      for (Entry<String,String> entry : es) {
        conn.tableOperations().setProperty(mapping.tableName, entry.getKey(), entry.getValue());
      }

    } catch (AccumuloException e) {
      throw new IOException(e);
    } catch (AccumuloSecurityException e) {
      throw new IOException(e);
    } catch (TableExistsException e) {
      return;
    }
  }

  @Override
  public void deleteSchema() throws IOException {
    try {
      if (batchWriter != null)
        batchWriter.close();
      batchWriter = null;
      conn.tableOperations().delete(mapping.tableName);
    } catch (AccumuloException e) {
      throw new IOException(e);
    } catch (AccumuloSecurityException e) {
      throw new IOException(e);
    } catch (TableNotFoundException e) {
      return;
    }
  }

  @Override
  public boolean schemaExists() throws IOException {
    return conn.tableOperations().exists(mapping.tableName);
  }

  public ByteSequence populate(Iterator<Entry<Key,Value>> iter, T persistent) throws IOException {
    ByteSequence row = null;
    
    Map currentMap = null;
    ArrayList currentArray = null;
    Text currentFam = null;
    int currentPos = 0;
    Schema currentSchema = null;
    Field currentField = null;

    while (iter.hasNext()) {
      Entry<Key,Value> entry = iter.next();
      
      if (currentMap != null) {
        if (currentFam.equals(entry.getKey().getColumnFamily())) {
          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), fromBytes(currentSchema, entry.getValue().get()));
          continue;
        } else {
          persistent.put(currentPos, currentMap);
          currentMap = null;
        }
      } else if (currentArray != null) {
        if (currentFam.equals(entry.getKey().getColumnFamily())) {
          currentArray.add(fromBytes(currentSchema, entry.getValue().get()));
          continue;
        } else {
          persistent.put(currentPos, new ListGenericArray<T>(currentField.schema(), currentArray));
          currentArray = null;
        }
      }

      if (row == null)
        row = entry.getKey().getRowData();
      
      String fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), entry.getKey().getColumnQualifier()));
      if (fieldName == null)
        fieldName = mapping.columnMap.get(new Pair<Text,Text>(entry.getKey().getColumnFamily(), null));

      Field field = fieldMap.get(fieldName);

      switch (field.schema().getType()) {
        case MAP:
          currentMap = new StatefulHashMap();
          currentPos = field.pos();
          currentFam = entry.getKey().getColumnFamily();
          currentSchema = field.schema().getValueType();
          
          currentMap.put(new Utf8(entry.getKey().getColumnQualifierData().toArray()), fromBytes(currentSchema, entry.getValue().get()));

          break;
        case ARRAY:
          currentArray = new ArrayList();
          currentPos = field.pos();
          currentFam = entry.getKey().getColumnFamily();
          currentSchema = field.schema().getElementType();
          currentField = field;
          
          currentArray.add(fromBytes(currentSchema, entry.getValue().get()));

          break;
        case RECORD:
          SpecificDatumReader reader = new SpecificDatumReader(field.schema());
          byte[] val = entry.getValue().get();
          // TODO reuse decoder
          BinaryDecoder decoder = DecoderFactory.defaultFactory().createBinaryDecoder(val, null);
          persistent.put(field.pos(), reader.read(null, decoder));
          break;
        default:
          persistent.put(field.pos(), fromBytes(field.schema(), entry.getValue().get()));
      }
    }
    
    if (currentMap != null) {
      persistent.put(currentPos, currentMap);
    } else if (currentArray != null) {
      persistent.put(currentPos, new ListGenericArray<T>(currentField.schema(), currentArray));
    }

    persistent.clearDirty();

    return row;
  }

  private void setFetchColumns(Scanner scanner, String fields[]) {
    fields = getFieldsToQuery(fields);
    for (String field : fields) {
      Pair<Text,Text> col = mapping.fieldMap.get(field);
      if (col.getSecond() == null) {
        scanner.fetchColumnFamily(col.getFirst());
      } else {
        scanner.fetchColumn(col.getFirst(), col.getSecond());
      }
    }
  }

  @Override
  public T get(K key, String[] fields) throws IOException {
    try {
      // TODO make isolated scanner optional?
      Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS));
      Range rowRange = new Range(new Text(toBytes(key)));
      
      scanner.setRange(rowRange);
      setFetchColumns(scanner, fields);
      
      T persistent = newPersistent();
      ByteSequence row = populate(scanner.iterator(), persistent);
      if (row == null)
        return null;
      return persistent;
    } catch (TableNotFoundException e) {
      return null;
    }
  }
  
  @Override
  public void put(K key, T val) throws IOException {

    Mutation m = new Mutation(new Text(toBytes(key)));
    
    Schema schema = val.getSchema();
    StateManager stateManager = val.getStateManager();
    
    Iterator<Field> iter = schema.getFields().iterator();
    
    int count = 0;
    for (int i = 0; iter.hasNext(); i++) {
      Field field = iter.next();
      if (!stateManager.isDirty(val, i)) {
        continue;
      }
      
      Object o = val.get(i);
      Pair<Text,Text> col = mapping.fieldMap.get(field.name());

      switch (field.schema().getType()) {
        case MAP:
          if (o instanceof StatefulMap) {
            StatefulMap map = (StatefulMap) o;
            Set<?> es = map.states().entrySet();
            for (Object entry : es) {
              Object mapKey = ((Entry) entry).getKey();
              State state = (State) ((Entry) entry).getValue();

              switch (state) {
                case NEW:
                case DIRTY:
                  m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(map.get(mapKey))));
                  count++;
                  break;
                case DELETED:
                  m.putDelete(col.getFirst(), new Text(toBytes(mapKey)));
                  count++;
                  break;
              }
              
            }
          } else {
            Map map = (Map) o;
            Set<?> es = map.entrySet();
            for (Object entry : es) {
              Object mapKey = ((Entry) entry).getKey();
              Object mapVal = ((Entry) entry).getValue();
              m.put(col.getFirst(), new Text(toBytes(mapKey)), new Value(toBytes(mapVal)));
              count++;
            }
          }
          break;
        case ARRAY:
          GenericArray array = (GenericArray) o;
          int j = 0;
          for (Object item : array) {
            m.put(col.getFirst(), new Text(toBytes(j++)), new Value(toBytes(item)));
            count++;
          }
          break;
        case RECORD:
          SpecificDatumWriter writer = new SpecificDatumWriter(field.schema());
          ByteArrayOutputStream os = new ByteArrayOutputStream();
          BinaryEncoder encoder = new BinaryEncoder(os);
          writer.write(o, encoder);
          encoder.flush();
          m.put(col.getFirst(), col.getSecond(), new Value(os.toByteArray()));
          break;
        default:
          m.put(col.getFirst(), col.getSecond(), new Value(toBytes(o)));
          count++;
      }

    }
    
    if (count > 0)
      try {
        getBatchWriter().addMutation(m);
      } catch (MutationsRejectedException e) {
        throw new IOException(e);
      }
  }
  
  @Override
  public boolean delete(K key) throws IOException {
    Query<K,T> q = newQuery();
    q.setKey(key);
    return deleteByQuery(q) > 0;
  }

  @Override
  public long deleteByQuery(Query<K,T> query) throws IOException {
    try {
      Scanner scanner = createScanner(query);
      // add iterator that drops values on the server side
      scanner.addScanIterator(new IteratorSetting(Integer.MAX_VALUE, SortedKeyIterator.class));
      RowIterator iterator = new RowIterator(scanner.iterator());
      
      long count = 0;

      while (iterator.hasNext()) {
        Iterator<Entry<Key,Value>> row = iterator.next();
        Mutation m = null;
        while (row.hasNext()) {
          Entry<Key,Value> entry = row.next();
          Key key = entry.getKey();
          if (m == null)
            m = new Mutation(key.getRow());
          // TODO optimize to avoid continually creating column vis? prob does not matter for empty
          m.putDelete(key.getColumnFamily(), key.getColumnQualifier(), new ColumnVisibility(key.getColumnVisibility()), key.getTimestamp());
        }
        getBatchWriter().addMutation(m);
        count++;
      }
      
      return count;
    } catch (TableNotFoundException e) {
      // TODO return 0?
      throw new IOException(e);
    } catch (MutationsRejectedException e) {
      throw new IOException(e);
    }
  }

  private Range createRange(Query<K,T> query) {
    Text startRow = null;
    Text endRow = null;
    
    if (query.getStartKey() != null)
      startRow = new Text(toBytes(query.getStartKey()));
    
    if (query.getEndKey() != null)
      endRow = new Text(toBytes(query.getEndKey()));
    
    return new Range(startRow, true, endRow, true);
    
  }
  
  private Scanner createScanner(Query<K,T> query) throws TableNotFoundException {
    // TODO make isolated scanner optional?
    Scanner scanner = new IsolatedScanner(conn.createScanner(mapping.tableName, Constants.NO_AUTHS));
    setFetchColumns(scanner, query.getFields());
    
    scanner.setRange(createRange(query));
    
    if (query.getStartTime() != -1 || query.getEndTime() != -1) {
      IteratorSetting is = new IteratorSetting(30, TimestampFilter.class);
      if (query.getStartTime() != -1)
        TimestampFilter.setStart(is, query.getStartTime(), true);
      if (query.getEndTime() != -1)
        TimestampFilter.setEnd(is, query.getEndTime(), true);
      
      scanner.addScanIterator(is);
    }
    
    return scanner;
  }

  @Override
  public Result<K,T> execute(Query<K,T> query) throws IOException {
    try {
      Scanner scanner = createScanner(query);
      return new AccumuloResult<K,T>(this, query, scanner);
    } catch (TableNotFoundException e) {
      // TODO return empty result?
      throw new IOException(e);
    }
  }
  
  @Override
  public Query<K,T> newQuery() {
    return new AccumuloQuery<K,T>(this);
  }

  Text pad(Text key, int bytes) {
    if (key.getLength() < bytes)
      key = new Text(key);
    
    while (key.getLength() < bytes) {
      key.append(new byte[] {0}, 0, 1);
    }
    
    return key;
  }
  
  @Override
  public List<PartitionQuery<K,T>> getPartitions(Query<K,T> query) throws IOException {
    try {
      TabletLocator tl;
      if (conn instanceof MockConnector)
        tl = new MockTabletLocator();
      else
        tl = TabletLocator.getInstance(conn.getInstance(), authInfo, new Text(Tables.getTableId(conn.getInstance(), mapping.tableName)));
      
      Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
      
      tl.invalidateCache();
      while (tl.binRanges(Collections.singletonList(createRange(query)), binnedRanges).size() > 0) {
        // TODO log?
        if (!Tables.exists(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName)))
          throw new TableDeletedException(Tables.getTableId(conn.getInstance(), mapping.tableName));
        else if (Tables.getTableState(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName)) == TableState.OFFLINE)
          throw new TableOfflineException(conn.getInstance(), Tables.getTableId(conn.getInstance(), mapping.tableName));
        UtilWaitThread.sleep(100);
        tl.invalidateCache();
      }
      
      List<PartitionQuery<K,T>> ret = new ArrayList<PartitionQuery<K,T>>();
      
      Text startRow = null;
      Text endRow = null;
      if (query.getStartKey() != null)
        startRow = new Text(toBytes(query.getStartKey()));
      if (query.getEndKey() != null)
        endRow = new Text(toBytes(query.getEndKey()));
     
      //hadoop expects hostnames, accumulo keeps track of IPs... so need to convert
      HashMap<String,String> hostNameCache = new HashMap<String,String>();
 
      for (Entry<String,Map<KeyExtent,List<Range>>> entry : binnedRanges.entrySet()) {
        String ip = entry.getKey().split(":", 2)[0];
        String location = hostNameCache.get(ip);
        if (location == null) {
          InetAddress inetAddress = InetAddress.getByName(ip);
          location = inetAddress.getHostName();
          hostNameCache.put(ip, location);
        }

        Map<KeyExtent,List<Range>> tablets = entry.getValue();
        for (KeyExtent ke : tablets.keySet()) {
          
          K startKey = null;
          if (startRow == null || !ke.contains(startRow)) {
            if (ke.getPrevEndRow() != null) {
              startKey = followingKey(encoder, getKeyClass(), TextUtil.getBytes(ke.getPrevEndRow()));
            }
          } else {
            startKey = fromBytes(getKeyClass(), TextUtil.getBytes(startRow));
          }
          
          K endKey = null;
          if (endRow == null || !ke.contains(endRow)) {
            if (ke.getEndRow() != null)
              endKey = lastPossibleKey(encoder, getKeyClass(), TextUtil.getBytes(ke.getEndRow()));
          } else {
            endKey = fromBytes(getKeyClass(), TextUtil.getBytes(endRow));
          }
          
          PartitionQueryImpl pqi = new PartitionQueryImpl<K,T>(query, startKey, endKey, new String[] {location});
          ret.add(pqi);
        }
      }
      
      return ret;
    } catch (TableNotFoundException e) {
      throw new IOException(e);
    } catch (AccumuloException e) {
      throw new IOException(e);
    } catch (AccumuloSecurityException e) {
      throw new IOException(e);
    }
    
  }
  
  static <K> K lastPossibleKey(Encoder encoder, Class<K> clazz, byte[] er) {
    
    if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) {
      throw new UnsupportedOperationException();
    } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) {
      throw new UnsupportedOperationException();
    } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) {
      return fromBytes(encoder, clazz, encoder.lastPossibleKey(2, er));
    } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) {
      return fromBytes(encoder, clazz, encoder.lastPossibleKey(4, er));
    } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) {
      return fromBytes(encoder, clazz, encoder.lastPossibleKey(8, er));
    } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) {
      return fromBytes(encoder, clazz, encoder.lastPossibleKey(4, er));
    } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) {
      return fromBytes(encoder, clazz, encoder.lastPossibleKey(8, er));
    } else if (clazz.equals(String.class)) {
      throw new UnsupportedOperationException();
    } else if (clazz.equals(Utf8.class)) {
      return fromBytes(encoder, clazz, er);
    }
    
    throw new IllegalArgumentException("Unknown type " + clazz.getName());
  }


  /**
   * @param keyClass
   * @param bytes
   * @return
   */
  static <K> K followingKey(Encoder encoder, Class<K> clazz, byte[] per) {
    
    if (clazz.equals(Byte.TYPE) || clazz.equals(Byte.class)) {
      return (K) Byte.valueOf(encoder.followingKey(1, per)[0]);
    } else if (clazz.equals(Boolean.TYPE) || clazz.equals(Boolean.class)) {
      throw new UnsupportedOperationException();
    } else if (clazz.equals(Short.TYPE) || clazz.equals(Short.class)) {
      return fromBytes(encoder, clazz, encoder.followingKey(2, per));
    } else if (clazz.equals(Integer.TYPE) || clazz.equals(Integer.class)) {
      return fromBytes(encoder, clazz, encoder.followingKey(4, per));
    } else if (clazz.equals(Long.TYPE) || clazz.equals(Long.class)) {
      return fromBytes(encoder, clazz, encoder.followingKey(8, per));
    } else if (clazz.equals(Float.TYPE) || clazz.equals(Float.class)) {
      return fromBytes(encoder, clazz, encoder.followingKey(4, per));
    } else if (clazz.equals(Double.TYPE) || clazz.equals(Double.class)) {
      return fromBytes(encoder, clazz, encoder.followingKey(8, per));
    } else if (clazz.equals(String.class)) {
      throw new UnsupportedOperationException();
    } else if (clazz.equals(Utf8.class)) {
      return fromBytes(encoder, clazz, Arrays.copyOf(per, per.length + 1));
    }

    throw new IllegalArgumentException("Unknown type " + clazz.getName());
  }

  @Override
  public void flush() throws IOException {
    try {
      if (batchWriter != null) {
        batchWriter.flush();
      }
    } catch (MutationsRejectedException e) {
      throw new IOException(e);
    }
  }

  @Override
  public void close() throws IOException {
    try {
      if (batchWriter != null) {
        batchWriter.close();
        batchWriter = null;
      }
    } catch (MutationsRejectedException e) {
      throw new IOException(e);
    }
    
  }
}