@Test public void executeQuery() { UsePlan.setCurrentSchema("booktest"); String sql = "select customer_id,quantity from orders where quantity<8"; Statement stmt = parser.createStatement(sql); PardPlanner planner = new PardPlanner(); Plan plan = planner.plan(stmt); plan.setJobId("customer_k"); QueryPlan qPlan = (QueryPlan) plan; System.out.println(qPlan.getPlan()); List<Task> task = TaskScheduler.INSTANCE().generateTasks(plan); PardServlet.planList.add((QueryPlan) plan); for (Task t : task) { System.out.println(t.getTaskId()); if (t instanceof SendDataTask) { QueryPlan p = new QueryTestPlan(((SendDataTask) t).getNode(), "send_Data_" + t.getTaskId()); System.out.println(p.getPlan()); PardServlet.planList.add(p); } else if (t instanceof JoinTask) { QueryPlan p = new QueryTestPlan(((JoinTask) t).getNode(), "Join_" + t.getTaskId()); System.out.println(p.getPlan()); PardServlet.planList.add(p); } else if (t instanceof QueryTask) { QueryTask tt = (QueryTask) t; QueryPlan p = new QueryTestPlan(tt.getPlanNode(), "table_" + t.getTaskId().replace('-', '_')); System.out.println(p.getPlan()); PardServlet.planList.add(p); } } System.out.println(PardServlet.planList.size()); Job job = JobScheduler.INSTANCE().newJob(); task.forEach(job::addTask); job.setJobState(JobState.EXECUTING); job.setPlan(plan); PardResultSet rs = TaskScheduler.INSTANCE().executeJob(job); PardWebServer.main(new String[0]); }
public PardResultSet executeQuery(String sql) { logger.info("Accepted query: " + sql); long timerStart = System.currentTimeMillis(); Job job = jobScheduler.newJob(); if (job == null) { logger.log(Level.WARNING, "Cannot create job for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.BEGIN_ERR); } job.setSql(sql); jobScheduler.updateJob(job.getJobId()); logger.info("Created job for query, job id: " + job.getJobId()); Statement statement; try { statement = sqlParser.createStatement(sql); } catch (ParsingException e) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } catch (NullPointerException e1) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR, e1.getMessage()); } if (statement == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create statement for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } job.setStatement(statement); jobScheduler.updateJob(job.getJobId()); logger.info("Created statement for job[" + job.getJobId() + "], job state: " + job.getJobState()); Plan plan = null; ErrorMessage msg = ErrorMessage.getOKMessage(); try { plan = planner.plan(statement); } catch (SemanticException e) { logger.log(Level.WARNING, e.getSemanticErrorMessage().toString()); msg = e.getSemanticErrorMessage(); if (msg == null) { msg = ErrorMessage.getOKMessage(); } } if (plan == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create plan for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PLANNING_ERR, msg.getErrmsg()); } job.setPlan(plan); plan.setJobId(job.getJobId()); jobScheduler.updateJob(job.getJobId()); logger.info("Created plan for job[" + job.getJobId() + "], job state: " + job.getJobState()); List<Task> tasks = null; String taskMsg = null; try { tasks = taskScheduler.generateTasks(plan); } catch (TaskSchedulerException e) { taskMsg = e.getPardErrorMessage().toString(); } if (tasks == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create tasks for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.SCHEDULING_ERR, taskMsg); } if (!tasks.isEmpty()) { tasks.forEach(job::addTask); } jobScheduler.updateJob(job.getJobId()); logger.info("Generated tasks for job[" + job.getJobId() + "], job state: " + job.getJobState()); PardResultSet resultSet = taskScheduler.executeJob(job); if (plan.getMsg() != null) { resultSet.setSemanticErrmsg(plan.getMsg().toString()); } if (resultSet.getStatus() != PardResultSet.ResultStatus.OK) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Failed to execute job for sql: " + sql); } jobScheduler.updateJob(job.getJobId()); long timerStop = System.currentTimeMillis(); logger.info("Done executing job[" + job.getJobId() + "], job state: " + job.getJobState() + ", execution time: " + ((double) (timerStop - timerStart)) / 1000 + "s"); resultSet.setExecutionTime(timerStop - timerStart); return resultSet; }
PardQueryHandler extends Thread { public PardResultSet executeQuery(String sql) { logger.info("Accepted query: " + sql); long timerStart = System.currentTimeMillis(); Job job = jobScheduler.newJob(); if (job == null) { logger.log(Level.WARNING, "Cannot create job for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.BEGIN_ERR); } job.setSql(sql); jobScheduler.updateJob(job.getJobId()); logger.info("Created job for query, job id: " + job.getJobId()); Statement statement; try { statement = sqlParser.createStatement(sql); } catch (ParsingException e) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } catch (NullPointerException e1) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR, e1.getMessage()); } if (statement == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create statement for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } job.setStatement(statement); jobScheduler.updateJob(job.getJobId()); logger.info("Created statement for job[" + job.getJobId() + "], job state: " + job.getJobState()); Plan plan = null; ErrorMessage msg = ErrorMessage.getOKMessage(); try { plan = planner.plan(statement); } catch (SemanticException e) { logger.log(Level.WARNING, e.getSemanticErrorMessage().toString()); msg = e.getSemanticErrorMessage(); if (msg == null) { msg = ErrorMessage.getOKMessage(); } } if (plan == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create plan for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PLANNING_ERR, msg.getErrmsg()); } job.setPlan(plan); plan.setJobId(job.getJobId()); jobScheduler.updateJob(job.getJobId()); logger.info("Created plan for job[" + job.getJobId() + "], job state: " + job.getJobState()); List<Task> tasks = null; String taskMsg = null; try { tasks = taskScheduler.generateTasks(plan); } catch (TaskSchedulerException e) { taskMsg = e.getPardErrorMessage().toString(); } if (tasks == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create tasks for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.SCHEDULING_ERR, taskMsg); } if (!tasks.isEmpty()) { tasks.forEach(job::addTask); } jobScheduler.updateJob(job.getJobId()); logger.info("Generated tasks for job[" + job.getJobId() + "], job state: " + job.getJobState()); PardResultSet resultSet = taskScheduler.executeJob(job); if (plan.getMsg() != null) { resultSet.setSemanticErrmsg(plan.getMsg().toString()); } if (resultSet.getStatus() != PardResultSet.ResultStatus.OK) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Failed to execute job for sql: " + sql); } jobScheduler.updateJob(job.getJobId()); long timerStop = System.currentTimeMillis(); logger.info("Done executing job[" + job.getJobId() + "], job state: " + job.getJobState() + ", execution time: " + ((double) (timerStop - timerStart)) / 1000 + "s"); resultSet.setExecutionTime(timerStop - timerStart); return resultSet; } }
PardQueryHandler extends Thread { public PardResultSet executeQuery(String sql) { logger.info("Accepted query: " + sql); long timerStart = System.currentTimeMillis(); Job job = jobScheduler.newJob(); if (job == null) { logger.log(Level.WARNING, "Cannot create job for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.BEGIN_ERR); } job.setSql(sql); jobScheduler.updateJob(job.getJobId()); logger.info("Created job for query, job id: " + job.getJobId()); Statement statement; try { statement = sqlParser.createStatement(sql); } catch (ParsingException e) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } catch (NullPointerException e1) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR, e1.getMessage()); } if (statement == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create statement for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } job.setStatement(statement); jobScheduler.updateJob(job.getJobId()); logger.info("Created statement for job[" + job.getJobId() + "], job state: " + job.getJobState()); Plan plan = null; ErrorMessage msg = ErrorMessage.getOKMessage(); try { plan = planner.plan(statement); } catch (SemanticException e) { logger.log(Level.WARNING, e.getSemanticErrorMessage().toString()); msg = e.getSemanticErrorMessage(); if (msg == null) { msg = ErrorMessage.getOKMessage(); } } if (plan == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create plan for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PLANNING_ERR, msg.getErrmsg()); } job.setPlan(plan); plan.setJobId(job.getJobId()); jobScheduler.updateJob(job.getJobId()); logger.info("Created plan for job[" + job.getJobId() + "], job state: " + job.getJobState()); List<Task> tasks = null; String taskMsg = null; try { tasks = taskScheduler.generateTasks(plan); } catch (TaskSchedulerException e) { taskMsg = e.getPardErrorMessage().toString(); } if (tasks == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create tasks for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.SCHEDULING_ERR, taskMsg); } if (!tasks.isEmpty()) { tasks.forEach(job::addTask); } jobScheduler.updateJob(job.getJobId()); logger.info("Generated tasks for job[" + job.getJobId() + "], job state: " + job.getJobState()); PardResultSet resultSet = taskScheduler.executeJob(job); if (plan.getMsg() != null) { resultSet.setSemanticErrmsg(plan.getMsg().toString()); } if (resultSet.getStatus() != PardResultSet.ResultStatus.OK) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Failed to execute job for sql: " + sql); } jobScheduler.updateJob(job.getJobId()); long timerStop = System.currentTimeMillis(); logger.info("Done executing job[" + job.getJobId() + "], job state: " + job.getJobState() + ", execution time: " + ((double) (timerStop - timerStart)) / 1000 + "s"); resultSet.setExecutionTime(timerStop - timerStart); return resultSet; } PardQueryHandler(
            Socket socket,
            JobScheduler jobScheduler,
            TaskScheduler taskScheduler); }
PardQueryHandler extends Thread { public PardResultSet executeQuery(String sql) { logger.info("Accepted query: " + sql); long timerStart = System.currentTimeMillis(); Job job = jobScheduler.newJob(); if (job == null) { logger.log(Level.WARNING, "Cannot create job for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.BEGIN_ERR); } job.setSql(sql); jobScheduler.updateJob(job.getJobId()); logger.info("Created job for query, job id: " + job.getJobId()); Statement statement; try { statement = sqlParser.createStatement(sql); } catch (ParsingException e) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } catch (NullPointerException e1) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR, e1.getMessage()); } if (statement == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create statement for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } job.setStatement(statement); jobScheduler.updateJob(job.getJobId()); logger.info("Created statement for job[" + job.getJobId() + "], job state: " + job.getJobState()); Plan plan = null; ErrorMessage msg = ErrorMessage.getOKMessage(); try { plan = planner.plan(statement); } catch (SemanticException e) { logger.log(Level.WARNING, e.getSemanticErrorMessage().toString()); msg = e.getSemanticErrorMessage(); if (msg == null) { msg = ErrorMessage.getOKMessage(); } } if (plan == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create plan for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PLANNING_ERR, msg.getErrmsg()); } job.setPlan(plan); plan.setJobId(job.getJobId()); jobScheduler.updateJob(job.getJobId()); logger.info("Created plan for job[" + job.getJobId() + "], job state: " + job.getJobState()); List<Task> tasks = null; String taskMsg = null; try { tasks = taskScheduler.generateTasks(plan); } catch (TaskSchedulerException e) { taskMsg = e.getPardErrorMessage().toString(); } if (tasks == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create tasks for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.SCHEDULING_ERR, taskMsg); } if (!tasks.isEmpty()) { tasks.forEach(job::addTask); } jobScheduler.updateJob(job.getJobId()); logger.info("Generated tasks for job[" + job.getJobId() + "], job state: " + job.getJobState()); PardResultSet resultSet = taskScheduler.executeJob(job); if (plan.getMsg() != null) { resultSet.setSemanticErrmsg(plan.getMsg().toString()); } if (resultSet.getStatus() != PardResultSet.ResultStatus.OK) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Failed to execute job for sql: " + sql); } jobScheduler.updateJob(job.getJobId()); long timerStop = System.currentTimeMillis(); logger.info("Done executing job[" + job.getJobId() + "], job state: " + job.getJobState() + ", execution time: " + ((double) (timerStop - timerStart)) / 1000 + "s"); resultSet.setExecutionTime(timerStop - timerStart); return resultSet; } PardQueryHandler(
            Socket socket,
            JobScheduler jobScheduler,
            TaskScheduler taskScheduler); @Override void run(); PardResultSet executeQuery(String sql); }
PardQueryHandler extends Thread { public PardResultSet executeQuery(String sql) { logger.info("Accepted query: " + sql); long timerStart = System.currentTimeMillis(); Job job = jobScheduler.newJob(); if (job == null) { logger.log(Level.WARNING, "Cannot create job for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.BEGIN_ERR); } job.setSql(sql); jobScheduler.updateJob(job.getJobId()); logger.info("Created job for query, job id: " + job.getJobId()); Statement statement; try { statement = sqlParser.createStatement(sql); } catch (ParsingException e) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } catch (NullPointerException e1) { return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR, e1.getMessage()); } if (statement == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create statement for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PARSING_ERR); } job.setStatement(statement); jobScheduler.updateJob(job.getJobId()); logger.info("Created statement for job[" + job.getJobId() + "], job state: " + job.getJobState()); Plan plan = null; ErrorMessage msg = ErrorMessage.getOKMessage(); try { plan = planner.plan(statement); } catch (SemanticException e) { logger.log(Level.WARNING, e.getSemanticErrorMessage().toString()); msg = e.getSemanticErrorMessage(); if (msg == null) { msg = ErrorMessage.getOKMessage(); } } if (plan == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create plan for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.PLANNING_ERR, msg.getErrmsg()); } job.setPlan(plan); plan.setJobId(job.getJobId()); jobScheduler.updateJob(job.getJobId()); logger.info("Created plan for job[" + job.getJobId() + "], job state: " + job.getJobState()); List<Task> tasks = null; String taskMsg = null; try { tasks = taskScheduler.generateTasks(plan); } catch (TaskSchedulerException e) { taskMsg = e.getPardErrorMessage().toString(); } if (tasks == null) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Cannot create tasks for sql: " + sql); return new PardResultSet(PardResultSet.ResultStatus.SCHEDULING_ERR, taskMsg); } if (!tasks.isEmpty()) { tasks.forEach(job::addTask); } jobScheduler.updateJob(job.getJobId()); logger.info("Generated tasks for job[" + job.getJobId() + "], job state: " + job.getJobState()); PardResultSet resultSet = taskScheduler.executeJob(job); if (plan.getMsg() != null) { resultSet.setSemanticErrmsg(plan.getMsg().toString()); } if (resultSet.getStatus() != PardResultSet.ResultStatus.OK) { jobScheduler.failJob(job.getJobId()); logger.log(Level.WARNING, "Failed to execute job for sql: " + sql); } jobScheduler.updateJob(job.getJobId()); long timerStop = System.currentTimeMillis(); logger.info("Done executing job[" + job.getJobId() + "], job state: " + job.getJobState() + ", execution time: " + ((double) (timerStop - timerStart)) / 1000 + "s"); resultSet.setExecutionTime(timerStop - timerStart); return resultSet; } PardQueryHandler(
            Socket socket,
            JobScheduler jobScheduler,
            TaskScheduler taskScheduler); @Override void run(); PardResultSet executeQuery(String sql);  }