package org.locationtech.geowave.datastore.kudu.operations;

import java.util.Collection;
import org.apache.kudu.client.KuduException;
import org.locationtech.geowave.core.index.SinglePartitionQueryRanges;
import org.locationtech.geowave.core.store.CloseableIterator;
import org.locationtech.geowave.core.store.entities.GeoWaveRowIteratorTransformer;
import org.locationtech.geowave.core.store.operations.ReaderParams;
import org.locationtech.geowave.core.store.operations.RowReader;
import org.locationtech.geowave.core.store.query.filter.ClientVisibilityFilter;
import org.locationtech.geowave.core.store.util.DataStoreUtils;
import org.locationtech.geowave.datastore.kudu.operations.KuduOperations;
import org.locationtech.geowave.mapreduce.splits.RecordReaderParams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Sets;

public class KuduReader<T> implements RowReader<T> {
  private final ReaderParams<T> readerParams;
  private final RecordReaderParams recordReaderParams;
  private final KuduOperations operations;
  private final GeoWaveRowIteratorTransformer<T> rowTransformer;
  private CloseableIterator<T> iterator;
  private final boolean visibilityEnabled;
  private static final Logger LOGGER = LoggerFactory.getLogger(KuduReader.class);

  public KuduReader(
      final ReaderParams<T> readerParams,
      final KuduOperations operations,
      final boolean visibilityEnabled) {
    this.readerParams = readerParams;
    recordReaderParams = null;
    this.operations = operations;
    this.rowTransformer = readerParams.getRowTransformer();
    this.visibilityEnabled = visibilityEnabled;

    initScanner();
  }

  @SuppressWarnings("unchecked")
  protected void initScanner() {
    final Collection<SinglePartitionQueryRanges> ranges =
        readerParams.getQueryRanges().getPartitionQueryRanges();
    try {
      iterator =
          operations.getKuduRangeRead(
              readerParams.getIndex().getName(),
              readerParams.getAdapterIds(),
              ranges,
              DataStoreUtils.isMergingIteratorRequired(readerParams, visibilityEnabled),
              rowTransformer,
              new ClientVisibilityFilter(
                  Sets.newHashSet(readerParams.getAdditionalAuthorizations())),
              visibilityEnabled).results();
    } catch (final KuduException e) {
      LOGGER.error("Error in initializing reader", e);
    }
  }

  @Override
  public void close() {
    iterator.close();
  }

  @Override
  public boolean hasNext() {
    return iterator.hasNext();
  }

  @Override
  public T next() {
    return iterator.next();
  }

}

@@ -94,44 +94,155 @@
 ort 
-org.apache.kudu.client.KuduException
+java.util.Collections;%0Aimport java.util.Set;%0Aimport org.apache.kudu.client.KuduException;%0Aimport org.locationtech.geowave.core.index.ByteArrayRange
 ;%0Aim
@@ -802,24 +802,90 @@
 Operations;%0A
+import org.locationtech.geowave.mapreduce.splits.GeoWaveRowRange;%0A
 import org.l
@@ -1846,16 +1846,515 @@
 );%0A  %7D%0A%0A
+  public KuduReader(%0A      RecordReaderParams recordReaderParams,%0A      KuduOperations operations,%0A      GeoWaveRowIteratorTransformer%3CT%3E rowTransformer,%0A      boolean visibilityEnabled) %7B%0A    this.readerParams = null;%0A    this.recordReaderParams = recordReaderParams;%0A    this.operations = operations;%0A    this.rowTransformer =%0A        (GeoWaveRowIteratorTransformer%3CT%3E) GeoWaveRowIteratorTransformer.NO_OP_TRANSFORMER;%0A    this.visibilityEnabled = visibilityEnabled;%0A    initRecordScanner();%0A  %7D%0A%0A
   @Suppr
@@ -2872,24 +2872,24 @@
 lityFilter(%0A
-
             
@@ -3100,24 +3100,1306 @@
 %0A    %7D%0A  %7D%0A%0A
+  @SuppressWarnings(%22unchecked%22)%0A  protected void initRecordScanner() %7B%0A    final short%5B%5D adapterIds =%0A        recordReaderParams.getAdapterIds() != null ? recordReaderParams.getAdapterIds()%0A            : new short%5B0%5D;%0A%0A    final GeoWaveRowRange range = recordReaderParams.getRowRange();%0A    final byte%5B%5D startKey = range.isInfiniteStartSortKey() ? null : range.getStartSortKey();%0A    final byte%5B%5D stopKey = range.isInfiniteStopSortKey() ? null : range.getEndSortKey();%0A    final SinglePartitionQueryRanges partitionRange =%0A        new SinglePartitionQueryRanges(%0A            range.getPartitionKey(),%0A            Collections.singleton(new ByteArrayRange(startKey, stopKey)));%0A    try %7B%0A      this.iterator =%0A          operations.getKuduRangeRead(%0A              recordReaderParams.getIndex().getName(),%0A              adapterIds,%0A              Collections.singleton(partitionRange),%0A              DataStoreUtils.isMergingIteratorRequired(recordReaderParams, visibilityEnabled),%0A              rowTransformer,%0A              new ClientVisibilityFilter(%0A                  Sets.newHashSet(recordReaderParams.getAdditionalAuthorizations())),%0A              visibilityEnabled).results();%0A    %7D catch (final KuduException e) %7B%0A      LOGGER.error(%22Error in initializing reader%22, e);%0A    %7D%0A  %7D%0A%0A
   @Override%0A

Add kudu record reader