Source code

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.io.IOException;
022import java.sql.Timestamp;
023import java.text.ParseException;
024import java.util.ArrayList;
025import java.util.Date;
026import java.util.List;
027import java.util.Map;
028
029import javax.persistence.EntityManager;
030import javax.persistence.Query;
031
032import org.apache.commons.lang.StringUtils;
033import org.apache.oozie.ErrorCode;
034import org.apache.oozie.WorkflowJobBean;
035import org.apache.oozie.WorkflowsInfo;
036import org.apache.oozie.client.OozieClient;
037import org.apache.oozie.client.WorkflowJob.Status;
038import org.apache.oozie.util.DateUtils;
039import org.apache.oozie.util.XLog;
040import org.apache.openjpa.persistence.OpenJPAPersistence;
041import org.apache.openjpa.persistence.OpenJPAQuery;
042import org.apache.openjpa.persistence.jdbc.FetchDirection;
043import org.apache.openjpa.persistence.jdbc.JDBCFetchPlan;
044import org.apache.openjpa.persistence.jdbc.LRSSizeAlgorithm;
045import org.apache.openjpa.persistence.jdbc.ResultSetType;
046
047public class WorkflowsJobGetJPAExecutor implements JPAExecutor<WorkflowsInfo> {
048
049    private static final String seletStr = "Select w.id, w.appName, w.statusStr, w.run, w.user, w.group, w.createdTimestamp, "
050            + "w.startTimestamp, w.lastModifiedTimestamp, w.endTimestamp, w.externalId, w.parentId from WorkflowJobBean w";
051    private static final String countStr = "Select count(w) from WorkflowJobBean w";
052
053    private final Map<String, List<String>> filter;
054    private final int start;
055    private final int len;
056
057    /**
058     * This JPA Executor gets the workflows info for the range.
059     *
060     * @param filter
061     * @param start
062     * @param len
063     */
064    public WorkflowsJobGetJPAExecutor(Map<String, List<String>> filter, int start, int len) {
065        this.filter = filter;
066        this.start = start;
067        this.len = len;
068    }
069
070    /* (non-Javadoc)
071     * @see org.apache.oozie.executor.jpa.JPAExecutor#execute(javax.persistence.EntityManager)
072     */
073    @SuppressWarnings("unchecked")
074    @Override
075    public WorkflowsInfo execute(EntityManager em) throws JPAExecutorException {
076        List<String> orArray = new ArrayList<String>();
077        List<String> colArray = new ArrayList<String>();
078        List<Object> valArray = new ArrayList<Object>();
079        StringBuilder sb = new StringBuilder("");
080        boolean isStatus = false;
081        boolean isAppName = false;
082        boolean isUser = false;
083        boolean isEnabled = false;
084        boolean isId = false;
085        int index = 0;
086        for (Map.Entry<String, List<String>> entry : filter.entrySet()) {
087            String colName = null;
088            String colVar = null;
089            if (entry.getKey().equals(OozieClient.FILTER_GROUP)) {
090                XLog.getLog(getClass()).warn("Filter by 'group' is not supported anymore");
091            } else {
092                if (entry.getKey().equals(OozieClient.FILTER_STATUS)) {
093                    List<String> values = filter.get(OozieClient.FILTER_STATUS);
094                    colName = "status";
095                    for (int i = 0; i < values.size(); i++) {
096                        colVar = "status";
097                        colVar = colVar + index;
098                        if (!isEnabled && !isStatus) {
099                            sb.append(seletStr).append(" where w.statusStr IN (:status" + index);
100                            isStatus = true;
101                            isEnabled = true;
102                        }
103                        else {
104                            if (isEnabled && !isStatus) {
105                                sb.append(" and w.statusStr IN (:status" + index);
106                                isStatus = true;
107                            }
108                            else {
109                                if (isStatus) {
110                                    sb.append(", :status" + index);
111                                }
112                            }
113                        }
114                        if (i == values.size() - 1) {
115                            sb.append(")");
116                        }
117                        index++;
118                        valArray.add(values.get(i));
119                        orArray.add(colName);
120                        colArray.add(colVar);
121                    }
122                }
123                else {
124                    if (entry.getKey().equals(OozieClient.FILTER_NAME)) {
125                        List<String> values = filter.get(OozieClient.FILTER_NAME);
126                        colName = "appName";
127                        for (int i = 0; i < values.size(); i++) {
128                            colVar = "appName";
129                            colVar = colVar + index;
130                            if (!isEnabled && !isAppName) {
131                                sb.append(seletStr).append(" where w.appName IN (:appName" + index);
132                                isAppName = true;
133                                isEnabled = true;
134                            }
135                            else {
136                                if (isEnabled && !isAppName) {
137                                    sb.append(" and w.appName IN (:appName" + index);
138                                    isAppName = true;
139                                }
140                                else {
141                                    if (isAppName) {
142                                        sb.append(", :appName" + index);
143                                    }
144                                }
145                            }
146                            if (i == values.size() - 1) {
147                                sb.append(")");
148                            }
149                            index++;
150                            valArray.add(values.get(i));
151                            orArray.add(colName);
152                            colArray.add(colVar);
153                        }
154                    }
155                    else {
156                        if (entry.getKey().equals(OozieClient.FILTER_USER)) {
157                            List<String> values = filter.get(OozieClient.FILTER_USER);
158                            colName = "user";
159                            for (int i = 0; i < values.size(); i++) {
160                                colVar = "user";
161                                colVar = colVar + index;
162                                if (!isEnabled && !isUser) {
163                                    sb.append(seletStr).append(" where w.user IN (:user" + index);
164                                    isUser = true;
165                                    isEnabled = true;
166                                }
167                                else {
168                                    if (isEnabled && !isUser) {
169                                        sb.append(" and w.user IN (:user" + index);
170                                        isUser = true;
171                                    }
172                                    else {
173                                        if (isUser) {
174                                            sb.append(", :user" + index);
175                                        }
176                                    }
177                                }
178                                if (i == values.size() - 1) {
179                                    sb.append(")");
180                                }
181                                index++;
182                                valArray.add(values.get(i));
183                                orArray.add(colName);
184                                colArray.add(colVar);
185                            }
186                        }
187                    }
188                    if (entry.getKey().equals(OozieClient.FILTER_ID)) {
189                        List<String> values = filter.get(OozieClient.FILTER_ID);
190                        colName = "id";
191                        for (int i = 0; i < values.size(); i++) {
192                            colVar = "id";
193                            colVar = colVar + index;
194                            if (!isEnabled && !isId) {
195                                sb.append(seletStr).append(" where w.id IN (:id" + index);
196                                isId = true;
197                                isEnabled = true;
198                            }
199                            else {
200                                if (isEnabled && !isId) {
201                                    sb.append(" and w.id IN (:id" + index);
202                                    isId = true;
203                                }
204                                else {
205                                    if (isId) {
206                                        sb.append(", :id" + index);
207                                    }
208                                }
209                            }
210                            if (i == values.size() - 1) {
211                                sb.append(")");
212                            }
213                            index++;
214                            valArray.add(values.get(i));
215                            orArray.add(colName);
216                            colArray.add(colVar);
217                        }
218                    }
219                    if (entry.getKey().equalsIgnoreCase(OozieClient.FILTER_CREATED_TIME_START)) {
220                        List<String> values = filter.get(OozieClient.FILTER_CREATED_TIME_START);
221                        colName = "createdTimestampStart";
222                        if (values.size() > 1) {
223                            throw new JPAExecutorException(ErrorCode.E0302,
224                                    "cannot specify multiple startcreatedtime");
225                        }
226                        colVar = colName;
227                        colVar = colVar + index;
228                        if (!isEnabled) {
229                            sb.append(seletStr).append(" where w.createdTimestamp >= :" + colVar);
230                            isEnabled = true;
231                        }
232                        else {
233                            sb.append(" and w.createdTimestamp >= :" + colVar);
234                        }
235                        index++;
236                        Date createdTime = null;
237                        try {
238                            createdTime = parseCreatedTimeString(values.get(0));
239                        }
240                        catch (Exception e) {
241                            throw new JPAExecutorException(ErrorCode.E0302, e.getMessage());
242                        }
243                        Timestamp createdTimeStamp = new Timestamp(createdTime.getTime());
244                        valArray.add(createdTimeStamp);
245                        orArray.add(colName);
246                        colArray.add(colVar);
247
248                    }
249                    if (entry.getKey().equalsIgnoreCase(OozieClient.FILTER_CREATED_TIME_END)) {
250                        List<String> values = filter.get(OozieClient.FILTER_CREATED_TIME_END);
251                        colName = "createdTimestampEnd";
252                        if (values.size() > 1) {
253                            throw new JPAExecutorException(ErrorCode.E0302,
254                                    "cannot specify multiple endcreatedtime");
255                        }
256                        colVar = colName;
257                        colVar = colVar + index;
258                        if (!isEnabled) {
259                            sb.append(seletStr).append(" where w.createdTimestamp <= :" + colVar);
260                            isEnabled = true;
261                        }
262                        else {
263                            sb.append(" and w.createdTimestamp <= :" + colVar);
264                        }
265                        index++;
266                        Date createdTime = null;
267                        try {
268                            createdTime = parseCreatedTimeString(values.get(0));
269                        }
270                        catch (Exception e) {
271                            throw new JPAExecutorException(ErrorCode.E0302, e.getMessage());
272                        }
273                        Timestamp createdTimeStamp = new Timestamp(createdTime.getTime());
274                        valArray.add(createdTimeStamp);
275                        orArray.add(colName);
276                        colArray.add(colVar);
277                    }
278                }
279            }
280        }
281        int realLen = 0;
282
283        Query q = null;
284        Query qTotal = null;
285        if (orArray.size() == 0) {
286            q = em.createNamedQuery("GET_WORKFLOWS_COLUMNS");
287            q.setFirstResult(start - 1);
288            q.setMaxResults(len);
289            qTotal = em.createNamedQuery("GET_WORKFLOWS_COUNT");
290        }
291        else {
292            if (orArray.size() > 0) {
293                StringBuilder sbTotal = new StringBuilder(sb);
294                sb.append(" order by w.createdTimestamp desc ");
295                q = em.createQuery(sb.toString());
296                q.setFirstResult(start - 1);
297                q.setMaxResults(len);
298                qTotal = em.createQuery(sbTotal.toString().replace(seletStr, countStr));
299
300                for (int i = 0; i < orArray.size(); i++) {
301                    q.setParameter(colArray.get(i), valArray.get(i));
302                    qTotal.setParameter(colArray.get(i), valArray.get(i));
303                }
304            }
305        }
306
307        OpenJPAQuery kq = OpenJPAPersistence.cast(q);
308        JDBCFetchPlan fetch = (JDBCFetchPlan) kq.getFetchPlan();
309        fetch.setFetchBatchSize(20);
310        fetch.setResultSetType(ResultSetType.SCROLL_INSENSITIVE);
311        fetch.setFetchDirection(FetchDirection.FORWARD);
312        fetch.setLRSSizeAlgorithm(LRSSizeAlgorithm.LAST);
313        List<?> resultList = q.getResultList();
314        List<Object[]> objectArrList = (List<Object[]>) resultList;
315        List<WorkflowJobBean> wfBeansList = new ArrayList<WorkflowJobBean>();
316
317        for (Object[] arr : objectArrList) {
318            WorkflowJobBean ww = getBeanForWorkflowFromArray(arr);
319            wfBeansList.add(ww);
320        }
321
322        realLen = ((Long) qTotal.getSingleResult()).intValue();
323
324        return new WorkflowsInfo(wfBeansList, start, len, realLen);
325    }
326
327    /* (non-Javadoc)
328     * @see org.apache.oozie.executor.jpa.JPAExecutor#getName()
329     */
330    @Override
331    public String getName() {
332        return "WorkflowsJobGetJPAExecutor";
333    }
334
335    private Date parseCreatedTimeString(String time) throws Exception{
336        Date createdTime = null;
337        int offset = 0;
338        if (Character.isLetter(time.charAt(time.length() - 1))) {
339            switch (time.charAt(time.length() - 1)) {
340                case 'd':
341                    offset = Integer.parseInt(time.substring(0, time.length() - 1));
342                    if(offset > 0) {
343                        throw new IllegalArgumentException("offset must be minus from currentTime");
344                    }
345                    createdTime = org.apache.commons.lang.time.DateUtils.addDays(new Date(), offset);
346                    break;
347                case 'h':
348                    offset =  Integer.parseInt(time.substring(0, time.length() - 1));
349                    if(offset > 0) {
350                        throw new IllegalArgumentException("offset must be minus from currentTime");
351                    }
352                    createdTime = org.apache.commons.lang.time.DateUtils.addHours(new Date(), offset);
353                    break;
354                case 'm':
355                    offset =  Integer.parseInt(time.substring(0, time.length() - 1));
356                    if(offset > 0) {
357                        throw new IllegalArgumentException("offset must be minus from currentTime");
358                    }
359                    createdTime = org.apache.commons.lang.time.DateUtils.addMinutes(new Date(), offset);
360                    break;
361                case 'Z':
362                    createdTime = DateUtils.parseDateUTC(time);
363                    break;
364                default:
365                    throw new IllegalArgumentException("Unsupported time format " + time);
366            }
367        } else {
368            throw new IllegalArgumentException("the format of createdTime is wrong: " + time);
369        }
370        return createdTime;
371    }
372    private WorkflowJobBean getBeanForWorkflowFromArray(Object[] arr) {
373
374        WorkflowJobBean wfBean = new WorkflowJobBean();
375        wfBean.setId((String) arr[0]);
376        if (arr[1] != null) {
377            wfBean.setAppName((String) arr[1]);
378        }
379        if (arr[2] != null) {
380            wfBean.setStatus(Status.valueOf((String) arr[2]));
381        }
382        if (arr[3] != null) {
383            wfBean.setRun((Integer) arr[3]);
384        }
385        if (arr[4] != null) {
386            wfBean.setUser((String) arr[4]);
387        }
388        if (arr[5] != null) {
389            wfBean.setGroup((String) arr[5]);
390        }
391        if (arr[6] != null) {
392            wfBean.setCreatedTime((Timestamp) arr[6]);
393        }
394        if (arr[7] != null) {
395            wfBean.setStartTime((Timestamp) arr[7]);
396        }
397        if (arr[8] != null) {
398            wfBean.setLastModifiedTime((Timestamp) arr[8]);
399        }
400        if (arr[9] != null) {
401            wfBean.setEndTime((Timestamp) arr[9]);
402        }
403        if (arr[10] != null) {
404            wfBean.setExternalId((String) arr[10]);
405        }
406        if (arr[11] != null) {
407            wfBean.setParentId((String) arr[11]);
408        }
409        return wfBean;
410    }
411}