001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *      http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.oozie.executor.jpa;
020
021import java.sql.Timestamp;
022import java.util.ArrayList;
023import java.util.Date;
024import java.util.List;
025
026import javax.persistence.EntityManager;
027import javax.persistence.Query;
028
029import org.apache.oozie.CoordinatorActionBean;
030import org.apache.oozie.ErrorCode;
031import org.apache.oozie.service.JPAService;
032import org.apache.oozie.service.Services;
033
034/**
035 * Query Executor that provides API to run query for Coordinator Action
036 */
037public class CoordActionQueryExecutor extends
038        QueryExecutor<CoordinatorActionBean, CoordActionQueryExecutor.CoordActionQuery> {
039
040    public enum CoordActionQuery {
041        UPDATE_COORD_ACTION,
042        UPDATE_COORD_ACTION_STATUS_PENDING_TIME,
043        UPDATE_COORD_ACTION_FOR_INPUTCHECK,
044        UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK,
045        UPDATE_COORD_ACTION_DEPENDENCIES,
046        UPDATE_COORD_ACTION_FOR_START,
047        UPDATE_COORD_ACTION_FOR_MODIFIED_DATE,
048        UPDATE_COORD_ACTION_RERUN,
049        GET_COORD_ACTION,
050        GET_COORD_ACTION_STATUS,
051        GET_COORD_ACTIVE_ACTIONS_COUNT_BY_JOBID,
052        GET_COORD_ACTIONS_BY_LAST_MODIFIED_TIME,
053        GET_COORD_ACTIONS_STATUS_UNIGNORED,
054        GET_COORD_ACTIONS_PENDING_COUNT
055    };
056
057    private static CoordActionQueryExecutor instance = new CoordActionQueryExecutor();
058
059    private CoordActionQueryExecutor() {
060    }
061
062    public static QueryExecutor<CoordinatorActionBean, CoordActionQueryExecutor.CoordActionQuery> getInstance() {
063        return CoordActionQueryExecutor.instance;
064    }
065
066    @Override
067    public Query getUpdateQuery(CoordActionQuery namedQuery, CoordinatorActionBean actionBean, EntityManager em)
068            throws JPAExecutorException {
069
070        Query query = em.createNamedQuery(namedQuery.name());
071        switch (namedQuery) {
072            case UPDATE_COORD_ACTION:
073                query.setParameter("actionNumber", actionBean.getActionNumber());
074                query.setParameter("actionXml", actionBean.getActionXmlBlob());
075                query.setParameter("consoleUrl", actionBean.getConsoleUrl());
076                query.setParameter("createdConf", actionBean.getCreatedConfBlob());
077                query.setParameter("errorCode", actionBean.getErrorCode());
078                query.setParameter("errorMessage", actionBean.getErrorMessage());
079                query.setParameter("externalStatus", actionBean.getExternalStatus());
080                query.setParameter("missingDependencies", actionBean.getMissingDependenciesBlob());
081                query.setParameter("runConf", actionBean.getRunConfBlob());
082                query.setParameter("timeOut", actionBean.getTimeOut());
083                query.setParameter("trackerUri", actionBean.getTrackerUri());
084                query.setParameter("type", actionBean.getType());
085                query.setParameter("createdTime", actionBean.getCreatedTimestamp());
086                query.setParameter("externalId", actionBean.getExternalId());
087                query.setParameter("jobId", actionBean.getJobId());
088                query.setParameter("lastModifiedTime", new Date());
089                query.setParameter("nominalTime", actionBean.getNominalTimestamp());
090                query.setParameter("slaXml", actionBean.getSlaXmlBlob());
091                query.setParameter("status", actionBean.getStatus().toString());
092                query.setParameter("id", actionBean.getId());
093                break;
094
095            case UPDATE_COORD_ACTION_STATUS_PENDING_TIME:
096                query.setParameter("status", actionBean.getStatus().toString());
097                query.setParameter("pending", actionBean.getPending());
098                query.setParameter("lastModifiedTime", new Date());
099                query.setParameter("id", actionBean.getId());
100                break;
101
102            case UPDATE_COORD_ACTION_FOR_INPUTCHECK:
103                query.setParameter("status", actionBean.getStatus().toString());
104                query.setParameter("lastModifiedTime", new Date());
105                query.setParameter("actionXml", actionBean.getActionXmlBlob());
106                query.setParameter("missingDependencies", actionBean.getMissingDependenciesBlob());
107                query.setParameter("id", actionBean.getId());
108                break;
109
110            case UPDATE_COORD_ACTION_FOR_PUSH_INPUTCHECK:
111                query.setParameter("status", actionBean.getStatus().toString());
112                query.setParameter("lastModifiedTime", new Date());
113                query.setParameter("actionXml", actionBean.getActionXmlBlob());
114                query.setParameter("pushMissingDependencies", actionBean.getPushMissingDependenciesBlob());
115                query.setParameter("id", actionBean.getId());
116                break;
117
118            case UPDATE_COORD_ACTION_DEPENDENCIES:
119                query.setParameter("missingDependencies", actionBean.getMissingDependenciesBlob());
120                query.setParameter("pushMissingDependencies", actionBean.getPushMissingDependenciesBlob());
121                query.setParameter("id", actionBean.getId());
122                break;
123
124            case UPDATE_COORD_ACTION_FOR_START:
125                query.setParameter("status", actionBean.getStatus().toString());
126                query.setParameter("lastModifiedTime", new Date());
127                query.setParameter("runConf", actionBean.getRunConfBlob());
128                query.setParameter("externalId", actionBean.getExternalId());
129                query.setParameter("pending", actionBean.getPending());
130                query.setParameter("errorCode", actionBean.getErrorCode());
131                query.setParameter("errorMessage", actionBean.getErrorMessage());
132                query.setParameter("id", actionBean.getId());
133                break;
134
135            case UPDATE_COORD_ACTION_FOR_MODIFIED_DATE:
136                query.setParameter("lastModifiedTime", actionBean.getLastModifiedTimestamp());
137                query.setParameter("id", actionBean.getId());
138                break;
139
140            case UPDATE_COORD_ACTION_RERUN:
141                query.setParameter("actionXml", actionBean.getActionXmlBlob());
142                query.setParameter("status", actionBean.getStatusStr());
143                query.setParameter("externalId", actionBean.getExternalId());
144                query.setParameter("externalStatus", actionBean.getExternalStatus());
145                query.setParameter("rerunTime", actionBean.getRerunTimestamp());
146                query.setParameter("lastModifiedTime", actionBean.getLastModifiedTimestamp());
147                query.setParameter("createdTime", actionBean.getCreatedTimestamp());
148                query.setParameter("createdConf", actionBean.getCreatedConfBlob());
149                query.setParameter("runConf", actionBean.getRunConfBlob());
150                query.setParameter("missingDependencies", actionBean.getMissingDependenciesBlob());
151                query.setParameter("pushMissingDependencies", actionBean.getPushMissingDependenciesBlob());
152                query.setParameter("errorCode", actionBean.getErrorCode());
153                query.setParameter("errorMessage", actionBean.getErrorMessage());
154                query.setParameter("id", actionBean.getId());
155                break;
156
157            default:
158                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for "
159                        + namedQuery.name());
160        }
161        return query;
162    }
163
164    @Override
165    public Query getSelectQuery(CoordActionQuery namedQuery, EntityManager em, Object... parameters)
166            throws JPAExecutorException {
167        Query query = em.createNamedQuery(namedQuery.name());
168        CoordActionQuery caQuery = (CoordActionQuery) namedQuery;
169        switch (caQuery) {
170            case GET_COORD_ACTION:
171            case GET_COORD_ACTION_STATUS:
172                query.setParameter("id", parameters[0]);
173                break;
174            case GET_COORD_ACTIONS_BY_LAST_MODIFIED_TIME:
175                query.setParameter("lastModifiedTime", new Timestamp(((Date) parameters[0]).getTime()));
176                break;
177            case GET_COORD_ACTIONS_STATUS_UNIGNORED:
178                query.setParameter("jobId", parameters[0]);
179                break;
180            case GET_COORD_ACTIONS_PENDING_COUNT:
181                query.setParameter("jobId", parameters[0]);
182                break;
183
184            default:
185                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for "
186                        + caQuery.name());
187        }
188        return query;
189    }
190
191    @Override
192    public int executeUpdate(CoordActionQuery namedQuery, CoordinatorActionBean jobBean) throws JPAExecutorException {
193        JPAService jpaService = Services.get().get(JPAService.class);
194        EntityManager em = jpaService.getEntityManager();
195        Query query = getUpdateQuery(namedQuery, jobBean, em);
196        int ret = jpaService.executeUpdate(namedQuery.name(), query, em);
197        return ret;
198    }
199
200    @Override
201    public CoordinatorActionBean get(CoordActionQuery namedQuery, Object... parameters) throws JPAExecutorException {
202        JPAService jpaService = Services.get().get(JPAService.class);
203        EntityManager em = jpaService.getEntityManager();
204        Query query = getSelectQuery(namedQuery, em, parameters);
205        Object ret = jpaService.executeGet(namedQuery.name(), query, em);
206        if (ret == null) {
207            throw new JPAExecutorException(ErrorCode.E0605, query.toString());
208        }
209        CoordinatorActionBean bean = constructBean(namedQuery, ret);
210        return bean;
211    }
212
213    @Override
214    public List<CoordinatorActionBean> getList(CoordActionQuery namedQuery, Object... parameters)
215            throws JPAExecutorException {
216        JPAService jpaService = Services.get().get(JPAService.class);
217        EntityManager em = jpaService.getEntityManager();
218        Query query = getSelectQuery(namedQuery, em, parameters);
219        List<?> retList = (List<?>) jpaService.executeGetList(namedQuery.name(), query, em);
220        List<CoordinatorActionBean> beanList = new ArrayList<CoordinatorActionBean>();
221        if (retList != null) {
222            for (Object ret : retList) {
223                beanList.add(constructBean(namedQuery, ret));
224            }
225        }
226        return beanList;
227    }
228
229    private CoordinatorActionBean constructBean(CoordActionQuery namedQuery, Object ret) throws JPAExecutorException {
230        CoordinatorActionBean bean;
231        Object[] arr;
232        switch (namedQuery) {
233            case GET_COORD_ACTION:
234                bean = (CoordinatorActionBean) ret;
235                break;
236            case GET_COORD_ACTIONS_BY_LAST_MODIFIED_TIME:
237                bean = new CoordinatorActionBean();
238                bean.setJobId((String) ret);
239                break;
240            case GET_COORD_ACTION_STATUS:
241                bean = new CoordinatorActionBean();
242                bean.setStatusStr((String)ret);
243                break;
244            case GET_COORD_ACTIONS_STATUS_UNIGNORED:
245                arr = (Object[]) ret;
246                bean = new CoordinatorActionBean();
247                bean.setStatusStr((String)arr[0]);
248                bean.setPending((Integer)arr[1]);
249                break;
250
251            default:
252                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot construct action bean for "
253                        + namedQuery.name());
254        }
255        return bean;
256    }
257
258    @Override
259    public Object getSingleValue(CoordActionQuery namedQuery, Object... parameters) throws JPAExecutorException {
260        JPAService jpaService = Services.get().get(JPAService.class);
261        EntityManager em = jpaService.getEntityManager();
262        Query query = getSelectQuery(namedQuery, em, parameters);
263        Object ret = jpaService.executeGet(namedQuery.name(), query, em);
264        if (ret == null) {
265            throw new JPAExecutorException(ErrorCode.E0604, query.toString());
266        }
267        return ret;
268    }
269}