http://git-wip-us.apache.org/repos/asf/carbondata/blob/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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/4437920a/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
