carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [3/4] carbondata git commit: [CARBONDATA-2690][CarbonStore] implement RESTful API: create table, load data and select
Date Fri, 06 Jul 2018 07:35:20 GMT
http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/StoreServiceImpl.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/StoreServiceImpl.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/StoreServiceImpl.java
new file mode 100644
index 0000000..ac3b199
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/impl/StoreServiceImpl.java
@@ -0,0 +1,78 @@
+/*
+ * 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.carbondata.store.rpc.impl;
+
+import java.io.IOException;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.store.rpc.StoreService;
+import org.apache.carbondata.store.rpc.model.BaseResponse;
+import org.apache.carbondata.store.rpc.model.LoadDataRequest;
+import org.apache.carbondata.store.rpc.model.QueryRequest;
+import org.apache.carbondata.store.rpc.model.QueryResponse;
+import org.apache.carbondata.store.rpc.model.ShutdownRequest;
+import org.apache.carbondata.store.rpc.model.ShutdownResponse;
+import org.apache.carbondata.store.worker.Worker;
+
+import org.apache.hadoop.ipc.ProtocolSignature;
+
+@InterfaceAudience.Internal
+public class StoreServiceImpl implements StoreService {
+
+  private Worker worker;
+  RequestHandler handler;
+
+  public StoreServiceImpl(Worker worker) {
+    this.worker = worker;
+    this.handler = new RequestHandler(worker.getConf(), worker.getHadoopConf());
+  }
+
+  @Override
+  public BaseResponse loadData(LoadDataRequest request) {
+    return handler.handleLoadData(request);
+  }
+
+  @Override
+  public QueryResponse query(QueryRequest request) {
+    return handler.handleSearch(request);
+  }
+
+  @Override
+  public ShutdownResponse shutdown(ShutdownRequest request) {
+    return handler.handleShutdown(request);
+  }
+
+  @Override
+  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
+    return versionID;
+  }
+
+  @Override
+  public ProtocolSignature getProtocolSignature(String protocol, long clientVersion,
+      int clientMethodsHash) throws IOException {
+    return null;
+  }
+
+  public Worker getWorker() {
+    return worker;
+  }
+
+  public void setWorker(Worker worker) {
+    this.worker = worker;
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/model/BaseResponse.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/BaseResponse.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/BaseResponse.java
new file mode 100644
index 0000000..d826b32
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/BaseResponse.java
@@ -0,0 +1,69 @@
+/*
+ * 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.carbondata.store.rpc.model;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+
+import org.apache.hadoop.io.Writable;
+
+@InterfaceAudience.Internal
+public class BaseResponse implements Serializable, Writable {
+  private int status;
+  private String message;
+
+  public BaseResponse() {
+  }
+
+  public BaseResponse(int status, String message) {
+    this.status = status;
+    this.message = message;
+  }
+
+  public int getStatus() {
+    return status;
+  }
+
+  public void setStatus(int status) {
+    this.status = status;
+  }
+
+  public String getMessage() {
+    return message;
+  }
+
+  public void setMessage(String message) {
+    this.message = message;
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    out.writeInt(status);
+    out.writeUTF(message);
+  }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    status = in.readInt();
+    message = in.readUTF();
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/model/LoadDataRequest.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/LoadDataRequest.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/LoadDataRequest.java
new file mode 100644
index 0000000..e79fad2
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/LoadDataRequest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.carbondata.store.rpc.model;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
+import org.apache.carbondata.store.util.StoreUtil;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+public class LoadDataRequest implements Serializable, Writable {
+
+  private CarbonLoadModel model;
+
+  public LoadDataRequest() {
+  }
+
+  public LoadDataRequest(CarbonLoadModel model) {
+    this.model = model;
+  }
+
+  public CarbonLoadModel getModel() {
+    return model;
+  }
+
+  public void setModel(CarbonLoadModel model) {
+    this.model = model;
+  }
+
+  @Override
+  public void write(DataOutput out) throws IOException {
+    WritableUtils.writeCompressedByteArray(out, StoreUtil.serialize(model));
+  }
+
+  @Override
+  public void readFields(DataInput in) throws IOException {
+    byte[] bytes = WritableUtils.readCompressedByteArray(in);
+    model = (CarbonLoadModel) StoreUtil.deserialize(bytes);
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/model/QueryResponse.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/QueryResponse.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/QueryResponse.java
index 033f1a5..7ad9210 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/QueryResponse.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/QueryResponse.java
@@ -29,19 +29,17 @@ import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
 
 @InterfaceAudience.Internal
-public class QueryResponse implements Serializable, Writable {
+public class QueryResponse extends BaseResponse implements Serializable, Writable {
   private int queryId;
-  private int status;
-  private String message;
   private Object[][] rows;
 
   public QueryResponse() {
+    super();
   }
 
   public QueryResponse(int queryId, int status, String message, Object[][] rows) {
+    super(status, message);
     this.queryId = queryId;
-    this.status = status;
-    this.message = message;
     this.rows = rows;
   }
 
@@ -49,13 +47,6 @@ public class QueryResponse implements Serializable, Writable {
     return queryId;
   }
 
-  public int getStatus() {
-    return status;
-  }
-
-  public String getMessage() {
-    return message;
-  }
 
   public Object[][] getRows() {
     return rows;
@@ -63,17 +54,15 @@ public class QueryResponse implements Serializable, Writable {
 
   @Override
   public void write(DataOutput out) throws IOException {
+    super.write(out);
     out.writeInt(queryId);
-    out.writeInt(status);
-    out.writeUTF(message);
     WritableUtils.writeCompressedByteArray(out, ObjectSerializationUtil.serialize(rows));
   }
 
   @Override
   public void readFields(DataInput in) throws IOException {
+    super.readFields(in);
     queryId = in.readInt();
-    status = in.readInt();
-    message = in.readUTF();
     try {
       rows = (Object[][])ObjectSerializationUtil.deserialize(
           WritableUtils.readCompressedByteArray(in));

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/rpc/model/RegisterWorkerRequest.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/RegisterWorkerRequest.java b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/RegisterWorkerRequest.java
index 894948b..2131e3b 100644
--- a/store/core/src/main/java/org/apache/carbondata/store/rpc/model/RegisterWorkerRequest.java
+++ b/store/core/src/main/java/org/apache/carbondata/store/rpc/model/RegisterWorkerRequest.java
@@ -66,4 +66,8 @@ public class RegisterWorkerRequest implements Serializable, Writable {
     port = in.readInt();
     cores = in.readInt();
   }
+
+  @Override public String toString() {
+    return "RegisterWorkerRequest{" + "hostAddress='" + hostAddress + '\'' + ", port=" + port + '}';
+  }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/scheduler/Schedulable.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/scheduler/Schedulable.java b/store/core/src/main/java/org/apache/carbondata/store/scheduler/Schedulable.java
new file mode 100644
index 0000000..65d0786
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/scheduler/Schedulable.java
@@ -0,0 +1,74 @@
+/*
+ * 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.carbondata.store.scheduler;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.carbondata.store.rpc.StoreService;
+
+public class Schedulable {
+
+  private String id;
+  private String address;
+  private int port;
+  private int cores;
+  public StoreService service;
+  public AtomicInteger workload;
+
+  public Schedulable(String id, String address, int port, int cores, StoreService service) {
+    this.id = id;
+    this.address = address;
+    this.port = port;
+    this.cores = cores;
+    this.service = service;
+    this.workload = new AtomicInteger();
+  }
+
+  public String getId() {
+    return id;
+  }
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+  public String getAddress() {
+    return address;
+  }
+
+  public void setAddress(String address) {
+    this.address = address;
+  }
+
+  public int getPort() {
+    return port;
+  }
+
+  public void setPort(int port) {
+    this.port = port;
+  }
+
+  int getCores() {
+    return cores;
+  }
+
+  @Override public String toString() {
+    return "Schedulable{" + "id='" + id + '\'' + ", address='" + address + '\'' + ", port=" + port
+        + '}';
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/scheduler/Scheduler.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/scheduler/Scheduler.java b/store/core/src/main/java/org/apache/carbondata/store/scheduler/Scheduler.java
new file mode 100644
index 0000000..1b4cdde
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/scheduler/Scheduler.java
@@ -0,0 +1,136 @@
+/*
+ * 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.carbondata.store.scheduler;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.store.exception.WorkerTooBusyException;
+import org.apache.carbondata.store.rpc.model.BaseResponse;
+import org.apache.carbondata.store.rpc.model.LoadDataRequest;
+import org.apache.carbondata.store.rpc.model.QueryRequest;
+import org.apache.carbondata.store.rpc.model.QueryResponse;
+
+/**
+ * [[Master]] uses Scheduler to pick a Worker to send request
+ */
+public class Scheduler {
+
+  private static LogService LOGGER = LogServiceFactory.getLogService(Scheduler.class.getName());
+
+  // mapping of worker IP address to worker instance
+  private Map<String, Schedulable> ipMapWorker = new HashMap<>();
+  private List<Schedulable> workers = new ArrayList<>();
+  private AtomicLong index = new AtomicLong(0);
+  private ExecutorService executors = Executors.newCachedThreadPool();
+
+  /**
+   * Pick a Worker according to the address and workload of the Worker
+   * Invoke the RPC and return Future result
+   */
+  public Future<QueryResponse> sendRequestAsync(final Schedulable worker,
+      final QueryRequest request) {
+
+    LOGGER.info("sending search request to worker " + worker);
+    worker.workload.incrementAndGet();
+    return executors.submit(new Callable<QueryResponse>() {
+      @Override public QueryResponse call() {
+        return worker.service.query(request);
+      }
+    });
+  }
+
+  public BaseResponse sendRequest(final Schedulable worker,
+      final LoadDataRequest request) {
+
+    LOGGER.info("sending load data request to worker " + worker);
+    worker.workload.incrementAndGet();
+    return worker.service.loadData(request);
+  }
+
+  public Schedulable pickWorker(String splitAddress) {
+    Schedulable worker = ipMapWorker.get(splitAddress);
+    // no local worker available, choose one worker randomly
+    if (worker == null) {
+      worker = pickNexWorker();
+    }
+    // check whether worker exceed max workload, if exceeded, pick next worker
+    int maxWorkload = CarbonProperties.getMaxWorkloadForWorker(worker.getCores());
+    int numTry = workers.size();
+    do {
+      if (worker.workload.get() >= maxWorkload) {
+        LOGGER.info("worker " + worker + " reach limit, re-select worker...");
+        worker = pickNexWorker();
+        numTry = numTry - 1;
+      } else {
+        numTry = -1;
+      }
+    } while (numTry > 0);
+    if (numTry == 0) {
+      // tried so many times and still not able to find Worker
+      throw new WorkerTooBusyException(
+          "All workers are busy, number of workers: " + workers.size() + ", workload limit:"
+              + maxWorkload);
+    }
+
+    return worker;
+  }
+
+  public Schedulable pickNexWorker() {
+    return workers.get((int) (index.get() % workers.size()));
+  }
+
+  /**
+   * A new searcher is trying to register, add it to the map and connect to this searcher
+   */
+  public void addWorker(Schedulable schedulable) {
+    workers.add(schedulable);
+    ipMapWorker.put(schedulable.getAddress(), schedulable);
+  }
+
+  public void removeWorker(String address) {
+    Schedulable schedulable = ipMapWorker.get(address);
+    if (schedulable != null) {
+      ipMapWorker.remove(address);
+      workers.remove(schedulable);
+    }
+  }
+
+  public List<Schedulable> getAllWorkers() {
+    return workers;
+  }
+
+  public List<String> getAllWorkerAddresses() {
+    List<String> addresses = new ArrayList<>(workers.size());
+    for (Schedulable worker : workers) {
+      addresses.add(worker.getAddress());
+    }
+    return addresses;
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/util/StoreUtil.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/util/StoreUtil.java b/store/core/src/main/java/org/apache/carbondata/store/util/StoreUtil.java
new file mode 100644
index 0000000..fba3413
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/util/StoreUtil.java
@@ -0,0 +1,132 @@
+/*
+ * 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.carbondata.store.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.memory.UnsafeMemoryManager;
+import org.apache.carbondata.core.memory.UnsafeSortMemoryManager;
+import org.apache.carbondata.core.util.CarbonProperties;
+import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat;
+import org.apache.carbondata.processing.loading.model.CarbonLoadModel;
+import org.apache.carbondata.store.conf.StoreConf;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.log4j.BasicConfigurator;
+import org.apache.log4j.PropertyConfigurator;
+
+public class StoreUtil {
+
+  private static LogService LOGGER = LogServiceFactory.getLogService(StoreUtil.class.getName());
+
+  public static void loadProperties(String filePath, StoreConf conf) {
+    InputStream input = null;
+    try {
+      input = new FileInputStream(filePath);
+      Properties prop = new Properties();
+      prop.load(input);
+      for (Map.Entry<Object, Object> entry : prop.entrySet()) {
+        conf.conf(entry.getKey().toString(), entry.getValue().toString());
+      }
+      LOGGER.audit("loaded properties: " + filePath);
+    } catch (IOException ex) {
+      LOGGER.error(ex, "Failed to load properties file " + filePath);
+    } finally {
+      if (input != null) {
+        try {
+          input.close();
+        } catch (IOException e) {
+          LOGGER.error(e);
+        }
+      }
+    }
+  }
+
+  public static void initLog4j(String propertiesFilePath) {
+    BasicConfigurator.configure();
+    PropertyConfigurator.configure(propertiesFilePath);
+  }
+
+  public static byte[] serialize(Object object) {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream(1024);
+    try {
+      ObjectOutputStream oos = new ObjectOutputStream(baos);
+      oos.writeObject(object);
+    } catch (IOException e) {
+      LOGGER.error(e);
+    }
+    return baos.toByteArray();
+  }
+
+  public static Object deserialize(byte[] bytes) {
+    if (bytes == null) {
+      return null;
+    }
+    try {
+      ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
+      return ois.readObject();
+    } catch (IOException e) {
+      LOGGER.error(e);
+    } catch (ClassNotFoundException e) {
+      LOGGER.error(e);
+    }
+    return null;
+  }
+
+  public static void configureCSVInputFormat(Configuration configuration,
+      CarbonLoadModel carbonLoadModel) {
+    CSVInputFormat.setCommentCharacter(configuration, carbonLoadModel.getCommentChar());
+    CSVInputFormat.setCSVDelimiter(configuration, carbonLoadModel.getCsvDelimiter());
+    CSVInputFormat.setSkipEmptyLine(configuration, carbonLoadModel.getSkipEmptyLine());
+    CSVInputFormat.setEscapeCharacter(configuration, carbonLoadModel.getEscapeChar());
+    CSVInputFormat.setMaxColumns(configuration, carbonLoadModel.getMaxColumns());
+    CSVInputFormat.setNumberOfColumns(configuration,
+        "" + carbonLoadModel.getCsvHeaderColumns().length);
+
+    CSVInputFormat.setHeaderExtractionEnabled(
+        configuration,
+        carbonLoadModel.getCsvHeader() == null ||
+            StringUtils.isEmpty(carbonLoadModel.getCsvHeader()));
+
+    CSVInputFormat.setQuoteCharacter(configuration, carbonLoadModel.getQuoteChar());
+
+    CSVInputFormat.setReadBufferSize(
+        configuration,
+        CarbonProperties.getInstance().getProperty(
+            CarbonCommonConstants.CSV_READ_BUFFER_SIZE,
+            CarbonCommonConstants.CSV_READ_BUFFER_SIZE_DEFAULT));
+  }
+
+  public static void clearUnsafeMemory(long taskId) {
+    UnsafeMemoryManager.INSTANCE.freeMemoryAll(taskId);
+    UnsafeSortMemoryManager.INSTANCE.freeMemoryAll(taskId);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/java/org/apache/carbondata/store/worker/Worker.java
----------------------------------------------------------------------
diff --git a/store/core/src/main/java/org/apache/carbondata/store/worker/Worker.java b/store/core/src/main/java/org/apache/carbondata/store/worker/Worker.java
new file mode 100644
index 0000000..6fa2191
--- /dev/null
+++ b/store/core/src/main/java/org/apache/carbondata/store/worker/Worker.java
@@ -0,0 +1,166 @@
+/*
+ * 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.carbondata.store.worker;
+
+import java.io.IOException;
+import java.net.BindException;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.store.conf.StoreConf;
+import org.apache.carbondata.store.rpc.RegistryService;
+import org.apache.carbondata.store.rpc.ServiceFactory;
+import org.apache.carbondata.store.rpc.StoreService;
+import org.apache.carbondata.store.rpc.impl.StoreServiceImpl;
+import org.apache.carbondata.store.rpc.model.RegisterWorkerRequest;
+import org.apache.carbondata.store.rpc.model.RegisterWorkerResponse;
+import org.apache.carbondata.store.util.StoreUtil;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RPC;
+
+public class Worker {
+
+  private static LogService LOGGER = LogServiceFactory.getLogService(Worker.class.getName());
+
+  private String id;
+  private RegistryService registry;
+  private StoreConf conf;
+  private Configuration hadoopConf;
+  private RPC.Server server;
+
+  public Worker(StoreConf conf) {
+    this.conf = conf;
+    this.hadoopConf = this.conf.newHadoopConf();
+  }
+
+  public void start() {
+    try {
+      startService();
+      registerToMaster();
+    } catch (IOException e) {
+      LOGGER.error(e, "worker failed to start");
+    }
+  }
+
+  private void startService() throws IOException {
+    BindException exception;
+    // we will try to create service at worse case 100 times
+    int numTry = 100;
+    int coreNum = conf.workerCoreNum();
+    String host = conf.workerHost();
+    int port = conf.workerPort();
+    StoreService queryService = new StoreServiceImpl(this);
+    do {
+      try {
+        server = new RPC.Builder(hadoopConf)
+            .setNumHandlers(coreNum)
+            .setBindAddress(host)
+            .setPort(port)
+            .setProtocol(StoreService.class)
+            .setInstance(queryService)
+            .build();
+        server.start();
+
+        numTry = 0;
+        exception = null;
+      } catch (BindException e) {
+        // port is occupied, increase the port number and try again
+        exception = e;
+        port = port + 1;
+        numTry = numTry - 1;
+      }
+    } while (numTry > 0);
+
+    if (exception != null) {
+      // we have tried many times, but still failed to find an available port
+      LOGGER.error(exception, "worker failed to start");
+      throw exception;
+    }
+
+    conf.conf(StoreConf.WORKER_PORT, port);
+    LOGGER.info("worker started on " + host + ":" + port + " successfully");
+
+  }
+
+  public void stop() {
+    try {
+      stopService();
+    } catch (InterruptedException e) {
+      LOGGER.error(e, "worker failed to start");
+    }
+  }
+
+  private void stopService() throws InterruptedException {
+    if (server != null) {
+      server.stop();
+      server.join();
+      server = null;
+    }
+  }
+
+  private void registerToMaster() throws IOException {
+
+    LOGGER.info("trying to register to master " + conf.masterHost() + ":" + conf.masterPort());
+    if (registry == null) {
+      registry = ServiceFactory.createRegistryService(conf.masterHost(), conf.masterPort());
+    }
+    RegisterWorkerRequest request =
+        new RegisterWorkerRequest(conf.workerHost(), conf.workerPort(), conf.workerCoreNum());
+    try {
+      RegisterWorkerResponse response = registry.registerWorker(request);
+      id = response.getWorkerId();
+    } catch (Throwable throwable) {
+      LOGGER.error(throwable, "worker failed to register");
+      throw new IOException(throwable);
+    }
+
+    LOGGER.info("worker " + id + " registered successfully");
+  }
+
+  public String getId() {
+    return id;
+  }
+
+  public static void main(String[] args) {
+    if (args.length != 2) {
+      System.err.println("Usage: Worker <log4j file> <properties file>");
+      return;
+    }
+
+    StoreUtil.initLog4j(args[0]);
+    Worker worker = new Worker(new StoreConf(args[1]));
+    worker.start();
+  }
+
+  public StoreConf getConf() {
+    return conf;
+  }
+
+  public void setConf(StoreConf conf) {
+    this.conf = conf;
+  }
+
+  public Configuration getHadoopConf() {
+    return hadoopConf;
+  }
+
+  public void setHadoopConf(Configuration hadoopConf) {
+    this.hadoopConf = hadoopConf;
+  }
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/scala/org/apache/carbondata/store/Master.scala
----------------------------------------------------------------------
diff --git a/store/core/src/main/scala/org/apache/carbondata/store/Master.scala b/store/core/src/main/scala/org/apache/carbondata/store/Master.scala
deleted file mode 100644
index 2109251..0000000
--- a/store/core/src/main/scala/org/apache/carbondata/store/Master.scala
+++ /dev/null
@@ -1,283 +0,0 @@
-/*
- * 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.carbondata.store
-
-import java.io.IOException
-import java.net.{BindException, InetAddress}
-import java.util.{List => JList, Map => JMap, Objects, Random, UUID}
-import java.util.concurrent.{ExecutionException, Future, TimeoutException, TimeUnit}
-import java.util.concurrent.atomic.AtomicBoolean
-
-import scala.collection.JavaConverters._
-import scala.collection.mutable.ArrayBuffer
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.ipc.RPC
-import org.apache.hadoop.mapred.JobConf
-import org.apache.hadoop.mapreduce.Job
-
-import org.apache.carbondata.common.annotations.InterfaceAudience
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datastore.block.Distributable
-import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.datastore.row.CarbonRow
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable
-import org.apache.carbondata.core.scan.expression.Expression
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.hadoop.CarbonMultiBlockSplit
-import org.apache.carbondata.hadoop.api.CarbonInputFormat
-import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
-import org.apache.carbondata.processing.util.CarbonLoaderUtil
-import org.apache.carbondata.store.rpc.{RegistryService, ServiceFactory}
-import org.apache.carbondata.store.rpc.impl.{RegistryServiceImpl, Status}
-import org.apache.carbondata.store.rpc.model._
-
-/**
- * Master of CarbonSearch.
- * It provides a Registry service for worker to register.
- * And it provides search API to fire RPC call to workers.
- */
-@InterfaceAudience.Internal
-private[store] class Master {
-  private val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  // worker host address map to EndpointRef
-
-  private val random = new Random
-
-  private var registryServer: RPC.Server = _
-
-  private val scheduler: Scheduler = new Scheduler
-
-  def buildServer(serverHost: String, serverPort: Int): RPC.Server = {
-    val hadoopConf = FileFactory.getConfiguration
-    val builder = new RPC.Builder(hadoopConf)
-    builder
-      .setBindAddress(serverHost)
-      .setPort(serverPort)
-      .setProtocol(classOf[RegistryService])
-      .setInstance(new RegistryServiceImpl(this))
-      .build
-  }
-
-  /** start service and listen on port passed in constructor */
-  def startService(): Unit = {
-    if (registryServer == null) {
-      LOG.info("Start search mode master thread")
-      val isStarted: AtomicBoolean = new AtomicBoolean(false)
-      new Thread(new Runnable {
-        override def run(): Unit = {
-          val hostAddress = InetAddress.getLocalHost.getHostAddress
-          var port = CarbonProperties.getSearchMasterPort
-          var exception: BindException = null
-          var numTry = 100  // we will try to create service at worse case 100 times
-          do {
-            try {
-              LOG.info(s"building registry-service on $hostAddress:$port")
-              registryServer = buildServer(hostAddress, port)
-              numTry = 0
-            } catch {
-              case e: BindException =>
-                // port is occupied, increase the port number and try again
-                exception = e
-                LOG.error(s"start registry-service failed: ${e.getMessage}")
-                port = port + 1
-                numTry = numTry - 1
-            }
-          } while (numTry > 0)
-          if (registryServer == null) {
-            // we have tried many times, but still failed to find an available port
-            throw exception
-          }
-          if (isStarted.compareAndSet(false, false)) {
-            synchronized {
-              isStarted.compareAndSet(false, true)
-            }
-          }
-          LOG.info("starting registry-service")
-          registryServer.start()
-          LOG.info("registry-service started")
-        }
-      }).start()
-      var count = 0
-      val countThreshold = 5000
-      while (isStarted.compareAndSet(false, false) && count < countThreshold) {
-        LOG.info(s"Waiting search mode master to start, retrying $count times")
-        Thread.sleep(10)
-        count = count + 1
-      }
-      if (count >= countThreshold) {
-        LOG.error(s"Search mode try $countThreshold times to start master but failed")
-        throw new RuntimeException(
-          s"Search mode try $countThreshold times to start master but failed")
-      } else {
-        LOG.info("Search mode master started")
-      }
-    } else {
-      LOG.info("Search mode master has already started")
-    }
-  }
-
-  def stopService(): Unit = {
-    if (registryServer != null) {
-      registryServer.stop()
-      registryServer.join()
-      registryServer = null
-    }
-  }
-
-  def stopAllWorkers(): Unit = {
-    scheduler.getAllWorkers.toSeq.foreach { case (address, schedulable) =>
-      val response = try {
-        schedulable.service.shutdown(new ShutdownRequest("user"))
-      } catch {
-        case throwable: Throwable =>
-          throw new IOException(throwable)
-      }
-      scheduler.removeWorker(address)
-    }
-  }
-
-  /** A new searcher is trying to register, add it to the map and connect to this searcher */
-  def addWorker(request: RegisterWorkerRequest): RegisterWorkerResponse = {
-    LOG.info(s"Receive Register request from worker ${request.getHostAddress}:${request.getPort} " +
-             s"with ${request.getCores} cores")
-    val workerId = UUID.randomUUID().toString
-    val workerAddress = request.getHostAddress
-    val workerPort = request.getPort
-    LOG.info(s"connecting to worker ${request.getHostAddress}:${request.getPort}, " +
-             s"workerId $workerId")
-
-    val searchService = ServiceFactory.createSearchService(workerAddress, workerPort)
-    scheduler.addWorker(workerAddress,
-      new Schedulable(workerId, workerAddress, workerPort, request.getCores, searchService))
-    LOG.info(s"worker ${request.getHostAddress}:${request.getPort} registered")
-    new RegisterWorkerResponse(workerId)
-  }
-
-  /**
-   * Execute search by firing RPC call to worker, return the result rows
-   * @param table table to search
-   * @param columns projection column names
-   * @param filter filter expression
-   * @param globalLimit max number of rows required in Master
-   * @param localLimit max number of rows required in Worker
-   * @return
-   */
-  def search(table: CarbonTable, columns: Array[String], filter: Expression,
-      globalLimit: Long, localLimit: Long): Array[CarbonRow] = {
-    Objects.requireNonNull(table)
-    Objects.requireNonNull(columns)
-    if (globalLimit < 0 || localLimit < 0) {
-      throw new IllegalArgumentException("limit should be positive")
-    }
-
-    val queryId = random.nextInt
-    var rowCount = 0
-    val output = new ArrayBuffer[CarbonRow]
-
-    def onSuccess(result: QueryResponse): Unit = {
-      // in case of RPC success, collect all rows in response message
-      if (result.getQueryId != queryId) {
-        throw new IOException(
-          s"queryId in response does not match request: ${result.getQueryId} != $queryId")
-      }
-      if (result.getStatus != Status.SUCCESS.ordinal()) {
-        throw new IOException(s"failure in worker: ${ result.getMessage }")
-      }
-
-      val itor = result.getRows.iterator
-      while (itor.hasNext && rowCount < globalLimit) {
-        output += new CarbonRow(itor.next())
-        rowCount = rowCount + 1
-      }
-      LOG.info(s"[QueryId:$queryId] accumulated result size $rowCount")
-    }
-    def onFaiure(e: Throwable) = throw new IOException(s"exception in worker: ${ e.getMessage }")
-    def onTimedout() = throw new ExecutionTimeoutException()
-
-    // prune data and get a mapping of worker hostname to list of blocks,
-    // then add these blocks to the QueryRequest and fire the RPC call
-    val nodeBlockMapping: JMap[String, JList[Distributable]] = pruneBlock(table, columns, filter)
-    val tuple = nodeBlockMapping.asScala.map { case (splitAddress, blocks) =>
-      // Build a QueryRequest
-      val split = new CarbonMultiBlockSplit(blocks, splitAddress)
-      val request =
-        new QueryRequest(queryId, split, table.getTableInfo, columns, filter, localLimit)
-
-      // Find an Endpoind and send the request to it
-      // This RPC is non-blocking so that we do not need to wait before send to next worker
-      scheduler.sendRequestAsync(splitAddress, request)
-    }
-
-    // loop to get the result of each Worker
-    tuple.foreach { case (worker: Schedulable, future: Future[QueryResponse]) =>
-
-      // if we have enough data already, we do not need to collect more result
-      if (rowCount < globalLimit) {
-        // wait for worker
-        val response = try {
-          future.get(CarbonProperties.getInstance().getQueryTimeout.toLong, TimeUnit.SECONDS)
-        } catch {
-          case e: ExecutionException => onFaiure(e)
-          case t: TimeoutException => onTimedout()
-        } finally {
-          worker.workload.decrementAndGet()
-        }
-        LOG.info(s"[QueryId:$queryId] receive search response from worker " +
-                 s"${worker.address}:${worker.port}")
-        onSuccess(response)
-      }
-    }
-    output.toArray
-  }
-
-  /**
-   * Prune data by using CarbonInputFormat.getSplit
-   * Return a mapping of host address to list of block
-   */
-  private def pruneBlock(
-      table: CarbonTable,
-      columns: Array[String],
-      filter: Expression): JMap[String, JList[Distributable]] = {
-    val jobConf = new JobConf(new Configuration)
-    val job = new Job(jobConf)
-    val format = CarbonInputFormatUtil.createCarbonTableInputFormat(
-      job, table, columns, filter, null, null)
-
-    // We will do FG pruning in reader side, so don't do it here
-    CarbonInputFormat.setFgDataMapPruning(job.getConfiguration, false)
-    val splits = format.getSplits(job)
-    val distributables = splits.asScala.map { split =>
-      split.asInstanceOf[Distributable]
-    }
-    CarbonLoaderUtil.nodeBlockMapping(
-      distributables.asJava,
-      -1,
-      getWorkers.asJava,
-      CarbonLoaderUtil.BlockAssignmentStrategy.BLOCK_NUM_FIRST,
-      null)
-  }
-
-  /** return hostname of all workers */
-  def getWorkers: Seq[String] = scheduler.getAllWorkers.map(_._1).toSeq
-}
-
-// Exception if execution timed out in search mode
-class ExecutionTimeoutException extends RuntimeException

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/scala/org/apache/carbondata/store/Scheduler.scala
----------------------------------------------------------------------
diff --git a/store/core/src/main/scala/org/apache/carbondata/store/Scheduler.scala b/store/core/src/main/scala/org/apache/carbondata/store/Scheduler.scala
deleted file mode 100644
index fb3ef86..0000000
--- a/store/core/src/main/scala/org/apache/carbondata/store/Scheduler.scala
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
- * 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.carbondata.store
-
-import java.io.IOException
-import java.util.concurrent.{Callable, Executors, Future}
-import java.util.concurrent.atomic.AtomicInteger
-
-import scala.collection.mutable
-import scala.reflect.ClassTag
-import scala.util.Random
-
-import org.apache.carbondata.common.annotations.InterfaceAudience
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.store.rpc.QueryService
-import org.apache.carbondata.store.rpc.model.{QueryRequest, QueryResponse}
-
-/**
- * [[Master]] uses Scheduler to pick a Worker to send request
- */
-@InterfaceAudience.Internal
-private[store] class Scheduler {
-  // mapping of worker IP address to worker instance
-  private val workers = mutable.Map[String, Schedulable]()
-  private val random = new Random()
-  private val executors = Executors.newCachedThreadPool()
-  private val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-
-  /**
-   * Pick a Worker according to the address and workload of the Worker
-   * Invoke the RPC and return Future result
-   */
-  def sendRequestAsync(
-      splitAddress: String,
-      request: QueryRequest): (Schedulable, Future[QueryResponse]) = {
-    require(splitAddress != null)
-    if (workers.isEmpty) {
-      throw new IOException("No worker is available")
-    }
-    var worker: Schedulable = pickWorker(splitAddress)
-
-    // check whether worker exceed max workload, if exceeded, pick next worker
-    val maxWorkload = CarbonProperties.getMaxWorkloadForWorker(worker.cores)
-    var numTry = workers.size
-    do {
-      if (worker.workload.get() >= maxWorkload) {
-        LOG.info(s"worker ${worker.address}:${worker.port} reach limit, re-select worker...")
-        worker = pickNextWorker(worker)
-        numTry = numTry - 1
-      } else {
-        numTry = -1
-      }
-    } while (numTry > 0)
-    if (numTry == 0) {
-      // tried so many times and still not able to find Worker
-      throw new WorkerTooBusyException(
-        s"All workers are busy, number of workers: ${workers.size}, workload limit: $maxWorkload")
-    }
-    LOG.info(s"sending search request to worker ${worker.address}:${worker.port}")
-    val future = executors.submit(
-      new Callable[QueryResponse] {
-        override def call(): QueryResponse = worker.service.query(request)
-      }
-    )
-    worker.workload.incrementAndGet()
-    (worker, future)
-  }
-
-  private def pickWorker[T: ClassTag](splitAddress: String) = {
-    try {
-      workers(splitAddress)
-    } catch {
-      case e: NoSuchElementException =>
-        // no local worker available, choose one worker randomly
-        pickRandomWorker()
-    }
-  }
-
-  /** pick a worker randomly */
-  private def pickRandomWorker() = {
-    val index = random.nextInt(workers.size)
-    workers.toSeq(index)._2
-  }
-
-  /** pick the next worker of the input worker in the [[Scheduler.workers]] */
-  private def pickNextWorker(worker: Schedulable) = {
-    val index = workers.zipWithIndex.find { case ((address, w), index) =>
-      w == worker
-    }.get._2
-    if (index == workers.size - 1) {
-      workers.toSeq.head._2
-    } else {
-      workers.toSeq(index + 1)._2
-    }
-  }
-
-  /** A new searcher is trying to register, add it to the map and connect to this searcher */
-  def addWorker(address: String, schedulable: Schedulable): Unit = {
-    require(schedulable != null)
-    require(address.equals(schedulable.address))
-    workers(address) = schedulable
-  }
-
-  def removeWorker(address: String): Unit = {
-    workers.remove(address)
-  }
-
-  def getAllWorkers: Iterator[(String, Schedulable)] = workers.iterator
-}
-
-/**
- * Represent a Worker which [[Scheduler]] can send
- * Search request on it
- * @param id Worker ID, a UUID string
- * @param cores, number of cores in Worker
- * @param service RPC service reference
- * @param workload number of outstanding request sent to Worker
- */
-private[store] class Schedulable(
-    val id: String,
-    val address: String,
-    val port: Int,
-    val cores: Int,
-    val service: QueryService,
-    var workload: AtomicInteger) {
-  def this(id: String, address: String, port: Int, cores: Int, service: QueryService) = {
-    this(id, address, port, cores, service, new AtomicInteger())
-  }
-}
-
-class WorkerTooBusyException(message: String) extends RuntimeException(message)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/main/scala/org/apache/carbondata/store/Worker.scala
----------------------------------------------------------------------
diff --git a/store/core/src/main/scala/org/apache/carbondata/store/Worker.scala b/store/core/src/main/scala/org/apache/carbondata/store/Worker.scala
deleted file mode 100644
index 2ded00b..0000000
--- a/store/core/src/main/scala/org/apache/carbondata/store/Worker.scala
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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.carbondata.store
-
-import java.io.IOException
-import java.net.{BindException, InetAddress}
-
-import org.apache.hadoop.ipc.RPC
-
-import org.apache.carbondata.common.annotations.InterfaceAudience
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.store.rpc.{QueryService, RegistryService, ServiceFactory}
-import org.apache.carbondata.store.rpc.impl.QueryServiceImpl
-import org.apache.carbondata.store.rpc.model.RegisterWorkerRequest
-
-@InterfaceAudience.Internal
-private[store] object Worker {
-  private val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
-  private val hostAddress = InetAddress.getLocalHost.getHostAddress
-  private var port: Int = _
-  private var registry: RegistryService = _
-
-  def init(masterHostAddress: String, masterPort: Int): Unit = {
-    LOG.info(s"initializing worker...")
-    startService()
-    LOG.info(s"registering to master $masterHostAddress:$masterPort")
-    val workerId = registerToMaster(masterHostAddress, masterPort)
-    LOG.info(s"worker registered to master, workerId: $workerId")
-  }
-
-  def buildServer(serverHost: String, serverPort: Int): RPC.Server = {
-    val hadoopConf = FileFactory.getConfiguration
-    val builder = new RPC.Builder(hadoopConf)
-    builder
-      .setNumHandlers(Runtime.getRuntime.availableProcessors)
-      .setBindAddress(serverHost)
-      .setPort(serverPort)
-      .setProtocol(classOf[QueryService])
-      .setInstance(new QueryServiceImpl)
-      .build
-  }
-
-  /**
-   * Start to listen on port [[CarbonProperties.getSearchWorkerPort]]
-   */
-  private def startService(): Unit = {
-    new Thread(new Runnable {
-      override def run(): Unit = {
-        port = CarbonProperties.getSearchWorkerPort
-        var searchServer: RPC.Server = null
-        var exception: BindException = null
-        var numTry = 100  // we will try to create service at worse case 100 times
-        do {
-          try {
-            LOG.info(s"building search-service on $hostAddress:$port")
-            searchServer = buildServer(hostAddress, port)
-            numTry = 0
-          } catch {
-            case e: BindException =>
-              // port is occupied, increase the port number and try again
-              exception = e
-              LOG.error(s"start search-service failed: ${e.getMessage}")
-              port = port + 1
-              numTry = numTry - 1
-          }
-        } while (numTry > 0)
-        if (searchServer == null) {
-          // we have tried many times, but still failed to find an available port
-          throw exception
-        }
-        LOG.info("starting search-service")
-        searchServer.start()
-        LOG.info("search-service started")
-      }
-    }).start()
-  }
-
-  private def registerToMaster(registryHostAddress: String, registryPort: Int): String = {
-    LOG.info(s"trying to register to master $registryHostAddress:$registryPort")
-    if (registry == null) {
-      registry = ServiceFactory.createRegistryService(registryHostAddress, registryPort)
-    }
-    val cores = Runtime.getRuntime.availableProcessors()
-    val request = new RegisterWorkerRequest(hostAddress, port, cores)
-    val response = try {
-      registry.registerWorker(request)
-    } catch {
-      case throwable: Throwable =>
-        LOG.error(s"worker failed to registered: $throwable")
-        throw new IOException(throwable)
-    }
-
-    LOG.info("worker registered")
-    response.getWorkerId
-  }
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/core/src/test/scala/org/apache/carbondata/store/SchedulerSuite.scala
----------------------------------------------------------------------
diff --git a/store/core/src/test/scala/org/apache/carbondata/store/SchedulerSuite.scala b/store/core/src/test/scala/org/apache/carbondata/store/SchedulerSuite.scala
deleted file mode 100644
index 95e7335..0000000
--- a/store/core/src/test/scala/org/apache/carbondata/store/SchedulerSuite.scala
+++ /dev/null
@@ -1,155 +0,0 @@
-/*
- * 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.carbondata.store
-
-import org.apache.hadoop.ipc.ProtocolSignature
-import org.scalatest.{BeforeAndAfterEach, FunSuite}
-
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.store.rpc.QueryService
-import org.apache.carbondata.store.rpc.model.{QueryRequest, QueryResponse, ShutdownRequest, ShutdownResponse}
-
-class SchedulerSuite extends FunSuite with BeforeAndAfterEach {
-
-  var scheduler: Scheduler = _
-  var w1: Schedulable = _
-  var w2: Schedulable = _
-  var w3: Schedulable = _
-
-  override def beforeEach(): Unit = {
-    scheduler = new Scheduler()
-    w1 = new Schedulable("id1", "1.1.1.1", 1000, 4, new DummyRef())
-    w2 = new Schedulable("id2", "1.1.1.2", 1000, 4, new DummyRef())
-    w3 = new Schedulable("id3", "1.1.1.3", 1000, 4, new DummyRef())
-
-    scheduler.addWorker("1.1.1.1", w1)
-    scheduler.addWorker("1.1.1.2", w2)
-    scheduler.addWorker("1.1.1.3", w3)
-  }
-
-  test("test addWorker, removeWorker, getAllWorkers") {
-    assertResult(Set("1.1.1.1", "1.1.1.2", "1.1.1.3"))(scheduler.getAllWorkers.toMap.keySet)
-
-    scheduler.removeWorker("1.1.1.2")
-    assertResult(Set("1.1.1.1", "1.1.1.3"))(scheduler.getAllWorkers.toMap.keySet)
-
-    val w4 = new Schedulable("id4", "1.1.1.4", 1000, 4, new DummyRef())
-    scheduler.addWorker("1.1.1.4", w4)
-    assertResult(Set("1.1.1.1", "1.1.1.3", "1.1.1.4"))(scheduler.getAllWorkers.toMap.keySet)
-    assertResult("id4")(scheduler.getAllWorkers.toMap.get("1.1.1.4").get.id)
-  }
-
-  test("test normal schedule") {
-    val (r1, _) = scheduler.sendRequestAsync("1.1.1.1", null)
-    assertResult(w1.id)(r1.id)
-    val (r2, _) = scheduler.sendRequestAsync("1.1.1.2", null)
-    assertResult(w2.id)(r2.id)
-    val (r3, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    assertResult(w3.id)(r3.id)
-    val (r4, _) = scheduler.sendRequestAsync("1.1.1.1", null)
-    assertResult(w1.id)(r4.id)
-    val (r5, _) = scheduler.sendRequestAsync("1.1.1.2", null)
-    assertResult(w2.id)(r5.id)
-    val (r6, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    assertResult(w3.id)(r6.id)
-  }
-
-  test("test worker unavailable") {
-    val (r1, _) = scheduler.sendRequestAsync("1.1.1.5", null)
-    assert(scheduler.getAllWorkers.map(_._2.id).contains(r1.id))
-  }
-
-  test("test reschedule when target worker is overload") {
-    // by default, maxWorkload is number of core * 10, so it is 40 in this test suite
-    (1 to 40).foreach { i =>
-      val (r2, _) = scheduler.sendRequestAsync("1.1.1.2", null)
-      val (r3, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    }
-    val (r, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    // it must be worker1 since worker3 exceed max workload
-    assertResult(w1.id)(r.id)
-  }
-
-  test("test all workers are overload") {
-    // by default, maxWorkload is number of core * 10, so it is 40 in this test suite
-    (1 to 40).foreach { i =>
-      val (r1, _) = scheduler.sendRequestAsync("1.1.1.1", null)
-      val (r2, _) = scheduler.sendRequestAsync("1.1.1.2", null)
-      val (r3, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    }
-
-    val e = intercept[WorkerTooBusyException] {
-      scheduler.sendRequestAsync("1.1.1.3", null)
-    }
-  }
-
-  test("test user configured overload param") {
-    val original = CarbonProperties.getInstance().getProperty(
-      CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT)
-
-    CarbonProperties.getInstance().addProperty(
-      CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT, "3")
-
-    (1 to 3).foreach { i =>
-      val (r1, _) = scheduler.sendRequestAsync("1.1.1.1", null)
-      val (r2, _) = scheduler.sendRequestAsync("1.1.1.2", null)
-      val (r3, _) = scheduler.sendRequestAsync("1.1.1.3", null)
-    }
-
-    val e = intercept[WorkerTooBusyException] {
-      scheduler.sendRequestAsync("1.1.1.3", null)
-    }
-
-    if (original != null) {
-      CarbonProperties.getInstance().addProperty(
-        CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT, original)
-    }
-  }
-
-  test("test invalid property") {
-    intercept[IllegalArgumentException] {
-      CarbonProperties.getInstance().addProperty(
-        CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT, "-3")
-    }
-    var value = CarbonProperties.getInstance().getProperty(
-      CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT)
-    assertResult(null)(value)
-
-    intercept[NumberFormatException] {
-      CarbonProperties.getInstance().addProperty(
-        CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT, "3s")
-    }
-    value = CarbonProperties.getInstance().getProperty(
-      CarbonCommonConstants.CARBON_SEARCH_MODE_WORKER_WORKLOAD_LIMIT)
-    assertResult(null)(value)
-  }
-}
-
-class DummyRef extends QueryService {
-  override def query(request: QueryRequest): QueryResponse = ???
-
-  override def shutdown(request: ShutdownRequest): ShutdownResponse = ???
-
-  override def getProtocolVersion(protocol: String,
-      clientVersion: Long): Long = ???
-
-  override def getProtocolSignature(protocol: String,
-      clientVersion: Long,
-      clientMethodsHash: Int): ProtocolSignature = ???
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/pom.xml
----------------------------------------------------------------------
diff --git a/store/horizon/pom.xml b/store/horizon/pom.xml
new file mode 100644
index 0000000..3665e53
--- /dev/null
+++ b/store/horizon/pom.xml
@@ -0,0 +1,95 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.apache.carbondata</groupId>
+    <artifactId>carbondata-parent</artifactId>
+    <version>1.5.0-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>carbondata-horizon</artifactId>
+  <name>Apache CarbonData :: Horizon </name>
+
+  <properties>
+    <dev.path>${basedir}/../../dev</dev.path>
+    <spring.version>2.0.2.RELEASE</spring.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.carbondata</groupId>
+      <artifactId>carbondata-store-core</artifactId>
+      <version>${project.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>om.google.code.gson</groupId>
+          <artifactId>gson</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-web</artifactId>
+      <version>${spring.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>ch.qos.logback</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.antlr</groupId>
+      <artifactId>antlr4-runtime</artifactId>
+      <version>4.7.1</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.scalatest</groupId>
+      <artifactId>scalatest_${scala.binary.version}</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <version>${spring.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+      <version>2.8.2</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <version>2.9.5</version>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>1.7</source>
+          <target>1.7</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+</project>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/src/main/anltr/Expression.g4
----------------------------------------------------------------------
diff --git a/store/horizon/src/main/anltr/Expression.g4 b/store/horizon/src/main/anltr/Expression.g4
new file mode 100644
index 0000000..81688cd
--- /dev/null
+++ b/store/horizon/src/main/anltr/Expression.g4
@@ -0,0 +1,163 @@
+/*
+ * Licensed 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.
+ *
+ * copy from SqlBase.g4 of Presto and Spark.
+ */
+
+grammar Expression;
+
+parseFilter
+ : booleanExpression EOF
+ ;
+
+booleanExpression
+    : predicate
+    | left=booleanExpression operator=AND right=booleanExpression
+    | left=booleanExpression operator=OR right=booleanExpression
+    | '(' booleanExpression ')'
+    ;
+
+predicate
+    : left=primaryExpression comparisonOperator right=primaryExpression
+    | left=primaryExpression NOT? BETWEEN lower=primaryExpression AND upper=primaryExpression
+    | left=primaryExpression NOT? IN '(' primaryExpression (',' primaryExpression)* ')'
+    | left=primaryExpression IS NOT? NULL
+    ;
+
+primaryExpression
+    : constant                                           #constantDefault
+    | identifier                                         #columnReference
+    | base=identifier '.' fieldName=identifier           #dereference
+    | '(' booleanExpression ')'                          #parenthesizedExpression
+    ;
+
+constant
+    : NULL                                               #nullLiteral
+    | number                                             #numericLiteral
+    | booleanValue                                       #booleanLiteral
+    | STRING+                                            #stringLiteral
+    ;
+
+identifier
+    : IDENTIFIER                                         #unquotedIdentifier
+    | BACKQUOTED_IDENTIFIER                              #backQuotedIdentifier
+    ;
+
+comparisonOperator
+    : EQ | NEQ | LT | LTE | GT | GTE
+    ;
+
+booleanValue
+    : TRUE | FALSE
+    ;
+
+number
+    : MINUS? DECIMAL_VALUE                               #decimalLiteral
+    | MINUS? INTEGER_VALUE                               #integerLiteral
+    | MINUS? BIGINT_LITERAL                              #bigIntLiteral
+    | MINUS? SMALLINT_LITERAL                            #smallIntLiteral
+    | MINUS? TINYINT_LITERAL                             #tinyIntLiteral
+    | MINUS? DOUBLE_LITERAL                              #doubleLiteral
+    | MINUS? BIGDECIMAL_LITERAL                          #bigDecimalLiteral
+    ;
+
+AND: 'AND';
+BETWEEN: 'BETWEEN';
+FALSE: 'FALSE';
+IN: 'IN';
+IS: 'IS';
+NOT: 'NOT';
+NULL: 'NULL';
+OR: 'OR';
+TRUE: 'TRUE';
+
+EQ  : '=';
+NEQ : '<>' | '!=';
+LT  : '<';
+LTE : '<=';
+GT  : '>';
+GTE : '>=';
+
+MINUS: '-';
+
+STRING
+    : '\'' ( ~('\''|'\\') | ('\\' .) )* '\''
+    | '"' ( ~('"'|'\\') | ('\\' .) )* '"'
+    ;
+
+BIGINT_LITERAL
+    : DIGIT+ 'L'
+    ;
+
+SMALLINT_LITERAL
+    : DIGIT+ 'S'
+    ;
+
+TINYINT_LITERAL
+    : DIGIT+ 'Y'
+    ;
+
+INTEGER_VALUE
+    : DIGIT+
+    ;
+
+DECIMAL_VALUE
+    : DIGIT+ EXPONENT
+    | DECIMAL_DIGITS EXPONENT?
+    ;
+
+DOUBLE_LITERAL
+    : DIGIT+ EXPONENT? 'D'
+    | DECIMAL_DIGITS EXPONENT? 'D'
+    ;
+
+BIGDECIMAL_LITERAL
+    : DIGIT+ EXPONENT? 'BD'
+    | DECIMAL_DIGITS EXPONENT? 'BD'
+    ;
+
+IDENTIFIER
+    : (LETTER | DIGIT | '_')+
+    ;
+
+BACKQUOTED_IDENTIFIER
+    : '`' ( ~'`' | '``' )* '`'
+    ;
+
+fragment DECIMAL_DIGITS
+    : DIGIT+ '.' DIGIT*
+    | '.' DIGIT+
+    ;
+
+fragment EXPONENT
+    : 'E' [+-]? DIGIT+
+    ;
+
+fragment DIGIT
+    : [0-9]
+    ;
+
+fragment LETTER
+    : [A-Z]
+    ;
+
+WS
+    : [ \r\n\t]+ -> channel(HIDDEN)
+    ;
+
+// Catch-all for anything we can't recognize.
+// We use this to be able to ignore and recover all the text
+// when splitting statements with DelimiterLexer
+UNRECOGNIZED
+    : .
+    ;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/ANTLRNoCaseStringStream.java
----------------------------------------------------------------------
diff --git a/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/ANTLRNoCaseStringStream.java b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/ANTLRNoCaseStringStream.java
new file mode 100644
index 0000000..1032c51
--- /dev/null
+++ b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/ANTLRNoCaseStringStream.java
@@ -0,0 +1,38 @@
+/*
+ * 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.carbondata.horizon.antlr;
+
+import org.antlr.v4.runtime.ANTLRInputStream;
+import org.antlr.v4.runtime.IntStream;
+
+public class ANTLRNoCaseStringStream extends ANTLRInputStream {
+
+  public ANTLRNoCaseStringStream(String input) {
+    super(input);
+  }
+
+  @Override public int LA(int i) {
+    int la = super.LA(i);
+    if (la == 0 || la == IntStream.EOF) {
+      return la;
+    } else {
+      return Character.toUpperCase(la);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/FilterVisitor.java
----------------------------------------------------------------------
diff --git a/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/FilterVisitor.java b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/FilterVisitor.java
new file mode 100644
index 0000000..0f28ea9
--- /dev/null
+++ b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/FilterVisitor.java
@@ -0,0 +1,215 @@
+/*
+ * 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.carbondata.horizon.antlr;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
+import org.apache.carbondata.core.scan.expression.ColumnExpression;
+import org.apache.carbondata.core.scan.expression.Expression;
+import org.apache.carbondata.core.scan.expression.LiteralExpression;
+import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression;
+import org.apache.carbondata.core.scan.expression.conditional.GreaterThanEqualToExpression;
+import org.apache.carbondata.core.scan.expression.conditional.GreaterThanExpression;
+import org.apache.carbondata.core.scan.expression.conditional.InExpression;
+import org.apache.carbondata.core.scan.expression.conditional.LessThanEqualToExpression;
+import org.apache.carbondata.core.scan.expression.conditional.LessThanExpression;
+import org.apache.carbondata.core.scan.expression.conditional.ListExpression;
+import org.apache.carbondata.core.scan.expression.conditional.NotEqualsExpression;
+import org.apache.carbondata.core.scan.expression.conditional.NotInExpression;
+import org.apache.carbondata.core.scan.expression.logical.AndExpression;
+import org.apache.carbondata.core.scan.expression.logical.OrExpression;
+import org.apache.carbondata.core.scan.expression.logical.RangeExpression;
+import org.apache.carbondata.horizon.antlr.gen.ExpressionBaseVisitor;
+import org.apache.carbondata.horizon.antlr.gen.ExpressionParser;
+
+public class FilterVisitor extends ExpressionBaseVisitor<Expression> {
+
+  private CarbonTable carbonTable;
+
+  public FilterVisitor(CarbonTable carbonTable) {
+    this.carbonTable = carbonTable;
+  }
+
+  public ColumnExpression getColumnExpression(String columnName) {
+    return getColumnExpression(carbonTable.getTableName(), columnName);
+  }
+
+  public ColumnExpression getColumnExpression(String tableName, String columnName) {
+    CarbonColumn column = carbonTable.getColumnByName(tableName, columnName);
+    if (column == null) {
+      throw new RuntimeException("column not exists: " + tableName + "." + columnName);
+    }
+    return new ColumnExpression(column.getColName(), column.getDataType());
+  }
+
+  @Override
+  public Expression visitParseFilter(ExpressionParser.ParseFilterContext ctx) {
+    return visitBooleanExpression(ctx.booleanExpression());
+  }
+
+  @Override
+  public Expression visitBooleanExpression(ExpressionParser.BooleanExpressionContext ctx) {
+    if (ctx.AND() != null) {
+      return new AndExpression(visitBooleanExpression(ctx.left), visitBooleanExpression(ctx.right));
+    } else if (ctx.OR() != null) {
+      return new OrExpression(visitBooleanExpression(ctx.left), visitBooleanExpression(ctx.right));
+    } else if (ctx.predicate() != null) {
+      return visitPredicate(ctx.predicate());
+    } else if (!ctx.booleanExpression().isEmpty()) {
+      return visitBooleanExpression(ctx.booleanExpression().get(0));
+    }
+
+    return super.visitBooleanExpression(ctx);
+  }
+
+  @Override
+  public Expression visitPredicate(ExpressionParser.PredicateContext ctx) {
+    ExpressionParser.ComparisonOperatorContext comparision = ctx.comparisonOperator();
+    if (comparision != null) {
+      if (comparision.EQ() != null) {
+        return new EqualToExpression(visit(ctx.left), visit(ctx.right));
+      } else if (comparision.GT() != null) {
+        return new GreaterThanExpression(visit(ctx.left), visit(ctx.right));
+      } else if (comparision.GTE() != null) {
+        return new GreaterThanEqualToExpression(visit(ctx.left), visit(ctx.right));
+      } else if (comparision.LT() != null) {
+        return new LessThanExpression(visit(ctx.left), visit(ctx.right));
+      } else if (comparision.LTE() != null) {
+        return new LessThanEqualToExpression(visit(ctx.left), visit(ctx.right));
+      } else if (comparision.NEQ() != null) {
+        return new NotEqualsExpression(visit(ctx.left), visit(ctx.right));
+      }
+    } else if (ctx.BETWEEN() != null) {
+      if (ctx.NOT() != null) {
+        return new RangeExpression(new GreaterThanExpression(visit(ctx.left), visit(ctx.upper)),
+            new LessThanExpression(visit(ctx.left), visit(ctx.lower)));
+      } else {
+        return new RangeExpression(
+            new GreaterThanEqualToExpression(visit(ctx.left), visit(ctx.lower)),
+            new LessThanEqualToExpression(visit(ctx.left), visit(ctx.upper)));
+      }
+    } else if (ctx.IN() != null) {
+      List<Expression> listExpression = new ArrayList<Expression>();
+      List<ExpressionParser.PrimaryExpressionContext> primaryExpressionContexts =
+          ctx.primaryExpression();
+      for (ExpressionParser.PrimaryExpressionContext primary : primaryExpressionContexts) {
+        if (ctx.left != primary) {
+          listExpression.add(visit(primary));
+        }
+      }
+      if (ctx.NOT() != null) {
+        return new NotInExpression(visit(ctx.left), new ListExpression(listExpression));
+      } else {
+        return new InExpression(visit(ctx.left), new ListExpression(listExpression));
+      }
+    } else if (ctx.NULL() != null) {
+      if (ctx.NOT() == null) {
+        return new EqualToExpression(
+            visit(ctx.left), new LiteralExpression(null, DataTypes.STRING), true);
+      } else {
+        return new NotEqualsExpression(visit(ctx.left),
+            new LiteralExpression(null, DataTypes.STRING), true);
+      }
+    }
+    return super.visitPredicate(ctx);
+  }
+
+  @Override
+  public Expression visitNullLiteral(ExpressionParser.NullLiteralContext ctx) {
+    return null;
+  }
+
+  @Override
+  public Expression visitDecimalLiteral(ExpressionParser.DecimalLiteralContext ctx) {
+    return new LiteralExpression(new BigDecimal(ctx.getText()),
+        DataTypes.createDefaultDecimalType());
+  }
+
+  @Override
+  public Expression visitIntegerLiteral(ExpressionParser.IntegerLiteralContext ctx) {
+    return new LiteralExpression(Integer.parseInt(ctx.getText()), DataTypes.INT);
+  }
+
+  @Override
+  public Expression visitBigIntLiteral(ExpressionParser.BigIntLiteralContext ctx) {
+    return new LiteralExpression(Long.parseLong(ctx.getText()), DataTypes.LONG);
+  }
+
+  @Override
+  public Expression visitSmallIntLiteral(ExpressionParser.SmallIntLiteralContext ctx) {
+    return new LiteralExpression(Short.parseShort(ctx.getText()), DataTypes.SHORT);
+  }
+
+  @Override
+  public Expression visitTinyIntLiteral(ExpressionParser.TinyIntLiteralContext ctx) {
+    return new LiteralExpression(Short.parseShort(ctx.getText()), DataTypes.SHORT);
+  }
+
+  @Override
+  public Expression visitDoubleLiteral(ExpressionParser.DoubleLiteralContext ctx) {
+    return new LiteralExpression(Double.parseDouble(ctx.getText()), DataTypes.DOUBLE);
+  }
+
+  @Override
+  public Expression visitBigDecimalLiteral(ExpressionParser.BigDecimalLiteralContext ctx) {
+    return new LiteralExpression(new BigDecimal(ctx.getText()),
+        DataTypes.createDefaultDecimalType());
+  }
+
+  @Override
+  public Expression visitBooleanLiteral(ExpressionParser.BooleanLiteralContext ctx) {
+    ExpressionParser.BooleanValueContext booleanValueContext = ctx.booleanValue();
+    if (booleanValueContext.FALSE() != null) {
+      return new LiteralExpression(false, DataTypes.BOOLEAN);
+    } else {
+      return new LiteralExpression(true, DataTypes.BOOLEAN);
+    }
+  }
+
+  @Override
+  public Expression visitStringLiteral(ExpressionParser.StringLiteralContext ctx) {
+    return new LiteralExpression(ctx.getText(), DataTypes.STRING);
+  }
+
+  @Override
+  public Expression visitUnquotedIdentifier(ExpressionParser.UnquotedIdentifierContext ctx) {
+    return getColumnExpression(ctx.getText());
+  }
+
+  private String identifier(String identifier) {
+    return identifier.replace("", "`");
+  }
+
+  @Override
+  public Expression visitBackQuotedIdentifier(ExpressionParser.BackQuotedIdentifierContext ctx) {
+    return getColumnExpression(identifier(ctx.getText()));
+  }
+
+  @Override
+  public Expression visitDereference(ExpressionParser.DereferenceContext ctx) {
+    String tableName = identifier(ctx.base.getText());
+    String columnName = identifier(ctx.fieldName.getText());
+    return getColumnExpression(tableName, columnName);
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/Expression.tokens
----------------------------------------------------------------------
diff --git a/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/Expression.tokens b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/Expression.tokens
new file mode 100644
index 0000000..0ba3c59
--- /dev/null
+++ b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/Expression.tokens
@@ -0,0 +1,51 @@
+T__0=1
+T__1=2
+T__2=3
+T__3=4
+AND=5
+BETWEEN=6
+FALSE=7
+IN=8
+IS=9
+NOT=10
+NULL=11
+OR=12
+TRUE=13
+EQ=14
+NEQ=15
+LT=16
+LTE=17
+GT=18
+GTE=19
+MINUS=20
+STRING=21
+BIGINT_LITERAL=22
+SMALLINT_LITERAL=23
+TINYINT_LITERAL=24
+INTEGER_VALUE=25
+DECIMAL_VALUE=26
+DOUBLE_LITERAL=27
+BIGDECIMAL_LITERAL=28
+IDENTIFIER=29
+BACKQUOTED_IDENTIFIER=30
+WS=31
+UNRECOGNIZED=32
+'('=1
+')'=2
+','=3
+'.'=4
+'AND'=5
+'BETWEEN'=6
+'FALSE'=7
+'IN'=8
+'IS'=9
+'NOT'=10
+'NULL'=11
+'OR'=12
+'TRUE'=13
+'='=14
+'<'=16
+'<='=17
+'>'=18
+'>='=19
+'-'=20

http://git-wip-us.apache.org/repos/asf/carbondata/blob/6fa86381/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/ExpressionBaseVisitor.java
----------------------------------------------------------------------
diff --git a/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/ExpressionBaseVisitor.java b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/ExpressionBaseVisitor.java
new file mode 100644
index 0000000..42ef9c8
--- /dev/null
+++ b/store/horizon/src/main/java/org/apache/carbondata/horizon/antlr/gen/ExpressionBaseVisitor.java
@@ -0,0 +1,168 @@
+// Generated from /home/david/Documents/code/carbondata/store/horizon/src/main/anltr/Expression.g4 by ANTLR 4.7
+package org.apache.carbondata.horizon.antlr.gen;
+import org.antlr.v4.runtime.tree.AbstractParseTreeVisitor;
+
+/**
+ * This class provides an empty implementation of {@link ExpressionVisitor},
+ * which can be extended to create a visitor which only needs to handle a subset
+ * of the available methods.
+ *
+ * @param <T> The return type of the visit operation. Use {@link Void} for
+ * operations with no return type.
+ */
+public class ExpressionBaseVisitor<T> extends AbstractParseTreeVisitor<T> implements ExpressionVisitor<T> {
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitParseFilter(ExpressionParser.ParseFilterContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBooleanExpression(ExpressionParser.BooleanExpressionContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitPredicate(ExpressionParser.PredicateContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitConstantDefault(ExpressionParser.ConstantDefaultContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitColumnReference(ExpressionParser.ColumnReferenceContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitDereference(ExpressionParser.DereferenceContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitParenthesizedExpression(ExpressionParser.ParenthesizedExpressionContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitNullLiteral(ExpressionParser.NullLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitNumericLiteral(ExpressionParser.NumericLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBooleanLiteral(ExpressionParser.BooleanLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitStringLiteral(ExpressionParser.StringLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitUnquotedIdentifier(ExpressionParser.UnquotedIdentifierContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBackQuotedIdentifier(ExpressionParser.BackQuotedIdentifierContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitComparisonOperator(ExpressionParser.ComparisonOperatorContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBooleanValue(ExpressionParser.BooleanValueContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitDecimalLiteral(ExpressionParser.DecimalLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitIntegerLiteral(ExpressionParser.IntegerLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBigIntLiteral(ExpressionParser.BigIntLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitSmallIntLiteral(ExpressionParser.SmallIntLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitTinyIntLiteral(ExpressionParser.TinyIntLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitDoubleLiteral(ExpressionParser.DoubleLiteralContext ctx) { return visitChildren(ctx); }
+	/**
+	 * {@inheritDoc}
+	 *
+	 * <p>The default implementation returns the result of calling
+	 * {@link #visitChildren} on {@code ctx}.</p>
+	 */
+	@Override public T visitBigDecimalLiteral(ExpressionParser.BigDecimalLiteralContext ctx) { return visitChildren(ctx); }
+}
\ No newline at end of file


Mime
View raw message