package org.opencb.opencga.storage.hadoop.variant.annotation.pending;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.opencb.biodata.models.core.Region;
import org.opencb.biodata.models.variant.Variant;
import org.opencb.commons.datastore.core.Query;
import org.opencb.opencga.storage.core.variant.adaptors.VariantQueryParam;
import org.opencb.opencga.storage.core.variant.query.VariantQueryUtils;
import org.opencb.opencga.storage.hadoop.utils.AbstractHBaseDataReader;
import org.opencb.opencga.storage.hadoop.variant.adaptors.VariantHBaseQueryParser;
import org.opencb.opencga.storage.hadoop.variant.adaptors.VariantHadoopDBAdaptor;
import org.opencb.opencga.storage.hadoop.variant.adaptors.phoenix.VariantPhoenixKeyFactory;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created on 13/02/19.
 *
 * @author Jacobo Coll &lt;jacobo167@gmail.com&gt;
 */
public class PendingVariantsToAnnotateReader extends AbstractHBaseDataReader<Variant> {

    private final VariantHadoopDBAdaptor dbAdaptor;

    public PendingVariantsToAnnotateReader(VariantHadoopDBAdaptor dbAdaptor, Query query) {
        super(dbAdaptor.getHBaseManager(), dbAdaptor.getTableNameGenerator().getPendingAnnotationTableName(), parseScan(query));
        this.dbAdaptor = dbAdaptor;
    }

    @Override
    public boolean pre() {
        try {
            PendingVariantsToAnnotateUtils.createTableIfNeeded(tableName, dbAdaptor.getHBaseManager());
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        return true;
    }

    @Override
    protected List<Variant> convert(List<Result> results) {
        List<Variant> variants = new ArrayList<>(results.size());
        for (Result result : results) {
//            System.out.println("result.isEmpty() = " + result.isEmpty());
//            for (Cell cell : result.rawCells()) {
//                System.out.println("cell = " + Bytes.toString(cell.getQualifier()));
//            }
            variants.add(VariantPhoenixKeyFactory.extractVariantFromVariantRowKey(result.getRow()));
        }
        return variants;
    }

    private static Scan parseScan(Query query) {
        Scan scan = new Scan();
        Region region = null;
        if (VariantQueryUtils.isValidParam(query, VariantQueryParam.REGION)) {
            region = new Region(query.getString(VariantQueryParam.REGION.key()));
        }
        VariantHBaseQueryParser.addRegionFilter(scan, region);
        return scan;
    }


}

@@ -1846,239 +1846,8 @@
 ) %7B%0A
-//            System.out.println(%22result.isEmpty() = %22 + result.isEmpty());%0A//            for (Cell cell : result.rawCells()) %7B%0A//                System.out.println(%22cell = %22 + Bytes.toString(cell.getQualifier()));%0A//            %7D%0A
     
@@ -2004,20 +2004,26 @@
  static 
+List%3C
 Scan
+%3E
  parseSc
@@ -2052,28 +2052,42 @@
     
+List%3C
 Scan
+%3E
  scan
+s
  = new 
-Scan
+ArrayList%3C%3E
 ();%0A
@@ -2098,21 +2098,28 @@
     
+List%3C
 Region
+%3E
  region
+s
  = n
@@ -2224,21 +2224,31 @@
 gion
+s
  = 
-new 
+Region.parse
 Region
+s
 (que
@@ -2296,32 +2296,207 @@
 ()));%0A        %7D%0A
+        if (regions == null) %7B%0A            scans.add(new Scan());%0A        %7D else %7B%0A            for (Region region : regions) %7B%0A                Scan scan = new Scan();%0A        
         VariantH
@@ -2542,16 +2542,73 @@
 egion);%0A
+                scans.add(scan);%0A            %7D%0A        %7D%0A
         
@@ -2618,16 +2618,17 @@
 urn scan
+s
 ;%0A    %7D%0A

Allow read pending variants to annotate from multiple regions