package com.alimama.mdrill.partion.dirtory;
import java.util.HashMap;
import java.util.HashSet;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.solr.common.SolrInputDocument;
import com.alimama.mdrill.json.JSONArray;
import com.alimama.mdrill.json.JSONObject;
import com.alimama.mdrill.partion.MdrillPartions;
import com.alimama.mdrill.partion.MdrillPartionsInterface;
import com.alimama.mdrill.partion.PartionListener;
import com.alimama.mdrill.partion.StatListenerInterface;
import com.alimama.mdrill.partion.thedate.ThedatePartionsUtils;
import com.alimama.mdrill.ui.service.utils.OperateType;
import com.alimama.mdrill.ui.service.utils.WebServiceParams;
public class DirtoryPartions implements MdrillPartionsInterface{
private String parttype="dir@dir";
private String fieldname="dir";
public void setPartionType(String parttype)
{
this.parttype=parttype;
String[] params=this.parttype.split("@");
if(params.length>1)
{
this.fieldname=params[1];
}
}
public String InsertPartion(SolrInputDocument doc) throws Exception{
return String.valueOf(doc.getFieldValue(this.fieldname));
}
@Override
public String[] SqlPartions(String queryStr) throws Exception {
HashSet<String> rtn = new HashSet<String>();
JSONArray jsonStr = new JSONArray(queryStr.trim());
for (int j = 0; j < jsonStr.length(); j++) {
JSONObject obj = jsonStr.getJSONObject(j);
if (obj.has(this.fieldname)) {
JSONObject thedate = obj.getJSONObject(this.fieldname);
Integer operate = Integer.parseInt(thedate.get("operate").toString());
OperateType optype=WebServiceParams.parseOperateType(operate);
String[] val = WebServiceParams.parseFqValue(thedate.getString("value"), operate).split(",");
if (optype.equals(OperateType.eq)||optype.equals(OperateType.in))
{
for (String dir : val) {
rtn.add(dir);
}
}
}
}
String[] rtnarr=new String[rtn.size()];
return rtn.toArray(rtnarr);
}
@Override
public String SqlFilter(String queryStr) throws Exception {
JSONArray rtn = new JSONArray();
JSONArray jsonStr = new JSONArray(queryStr.trim());
for (int j = 0; j < jsonStr.length(); j++) {
JSONObject obj = jsonStr.getJSONObject(j);
if (!obj.has(this.fieldname)) {
rtn.put(obj);
}
}
return rtn.toString();
}
@Override
public HashSet<String> getNameList(FileSystem fs, String inputBase,
String startPoint, int dayDelay, int maxRunDays) throws Exception {
HashSet<String> rtn = new HashSet<String>();
FileStatus[] list = fs.listStatus(new Path(inputBase));
if (list != null) {
for (FileStatus f : list) {
String name = f.getPath().getName().trim();
if (name != null&&!name.isEmpty()) {
rtn.add(name);
}
}
}
return rtn;
}
@Override
public HashMap<String, HashSet<String>> indexPartions(
HashSet<String> namelist, String startday, int dayDelay,
int maxRunDays) throws Exception {
HashMap<String, HashSet<String>> rtn=new HashMap<String, HashSet<String>>();
for(String s:namelist)
{
HashSet<String> list=new HashSet<String>();
list.add(s);
rtn.put(s, list);
}
return rtn;
}
@Override
public HashMap<String, String> indexVertify(
HashMap<String, HashSet<String>> partions, int shards,
String startday, int dayDelay, int maxRunDays) throws Exception {
HashMap<String, String> rtn=new HashMap<String, String>();
for(String key:partions.keySet())
{
String partionvertify = "partionV"+MdrillPartions.PARTION_VERSION+"@001@"+key+"@" + shards ;
rtn.put(key, partionvertify);
}
return rtn;
}
@Override
public StatListenerInterface getStatObj() throws Exception {
StatListenerInterface rtn=new PartionListener();
rtn.setPartionType(this.parttype);
rtn.init();
return rtn;
}
@Override
public String getDropComparePartion(long days) throws Exception {
return null;
}
@Override
public boolean isAllowDropPartion(String partion, String cmp)
throws Exception {
return false;
}
}