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.List;
024import java.util.Map;
025
026import javax.persistence.EntityManager;
027import javax.persistence.Query;
028
029import org.apache.oozie.CoordinatorJobBean;
030import org.apache.oozie.CoordinatorJobInfo;
031import org.apache.oozie.client.Job.Status;
032import org.apache.oozie.client.CoordinatorJob.Timeunit;
033import org.apache.oozie.store.StoreStatusFilter;
034import org.apache.oozie.util.ParamChecker;
035import org.apache.openjpa.persistence.OpenJPAPersistence;
036import org.apache.openjpa.persistence.OpenJPAQuery;
037import org.apache.openjpa.persistence.jdbc.FetchDirection;
038import org.apache.openjpa.persistence.jdbc.JDBCFetchPlan;
039import org.apache.openjpa.persistence.jdbc.LRSSizeAlgorithm;
040import org.apache.openjpa.persistence.jdbc.ResultSetType;
041
042/**
043 * Load the CoordinatorInfo and return it.
044 */
045public class CoordJobInfoGetJPAExecutor implements JPAExecutor<CoordinatorJobInfo> {
046
047    private Map<String, List<String>> filter;
048    private int start = 1;
049    private int len = 50;
050
051    public CoordJobInfoGetJPAExecutor(Map<String, List<String>> filter, int start, int len) {
052        ParamChecker.notNull(filter, "filter");
053        this.filter = filter;
054        this.start = start;
055        this.len = len;
056    }
057
058    @Override
059    public String getName() {
060        return "CoordJobInfoGetJPAExecutor";
061    }
062
063    @Override
064    @SuppressWarnings("unchecked")
065    public CoordinatorJobInfo execute(EntityManager em) throws JPAExecutorException {
066        List<String> orArray = new ArrayList<String>();
067        List<String> colArray = new ArrayList<String>();
068        List<String> valArray = new ArrayList<String>();
069        StringBuilder sb = new StringBuilder("");
070
071        StoreStatusFilter.filter(filter, orArray, colArray, valArray, sb, StoreStatusFilter.coordSeletStr,
072                                 StoreStatusFilter.coordCountStr);
073
074        int realLen = 0;
075
076        Query q = null;
077        Query qTotal = null;
078        if (orArray.size() == 0) {
079            q = em.createNamedQuery("GET_COORD_JOBS_COLUMNS");
080            q.setFirstResult(start - 1);
081            q.setMaxResults(len);
082            qTotal = em.createNamedQuery("GET_COORD_JOBS_COUNT");
083        }
084        else {
085            StringBuilder sbTotal = new StringBuilder(sb);
086            sb.append(" order by w.createdTimestamp desc ");
087            q = em.createQuery(sb.toString());
088            q.setFirstResult(start - 1);
089            q.setMaxResults(len);
090            qTotal = em.createQuery(sbTotal.toString().replace(StoreStatusFilter.coordSeletStr,
091                                                                          StoreStatusFilter.coordCountStr));
092        }
093
094        for (int i = 0; i < orArray.size(); i++) {
095            q.setParameter(colArray.get(i), valArray.get(i));
096            qTotal.setParameter(colArray.get(i), valArray.get(i));
097        }
098
099        OpenJPAQuery kq = OpenJPAPersistence.cast(q);
100        JDBCFetchPlan fetch = (JDBCFetchPlan) kq.getFetchPlan();
101        fetch.setFetchBatchSize(20);
102        fetch.setResultSetType(ResultSetType.SCROLL_INSENSITIVE);
103        fetch.setFetchDirection(FetchDirection.FORWARD);
104        fetch.setLRSSizeAlgorithm(LRSSizeAlgorithm.LAST);
105        List<?> resultList = q.getResultList();
106        List<Object[]> objectArrList = (List<Object[]>) resultList;
107        List<CoordinatorJobBean> coordBeansList = new ArrayList<CoordinatorJobBean>();
108
109        for (Object[] arr : objectArrList) {
110            CoordinatorJobBean ww = getBeanForCoordinatorJobFromArray(arr);
111            coordBeansList.add(ww);
112        }
113
114        realLen = ((Long) qTotal.getSingleResult()).intValue();
115
116        return new CoordinatorJobInfo(coordBeansList, start, len, realLen);
117    }
118
119    private CoordinatorJobBean getBeanForCoordinatorJobFromArray(Object[] arr) {
120        CoordinatorJobBean bean = new CoordinatorJobBean();
121        bean.setId((String) arr[0]);
122        if (arr[1] != null) {
123            bean.setAppName((String) arr[1]);
124        }
125        if (arr[2] != null) {
126            bean.setStatus(Status.valueOf((String) arr[2]));
127        }
128        if (arr[3] != null) {
129            bean.setUser((String) arr[3]);
130        }
131        if (arr[4] != null) {
132            bean.setGroup((String) arr[4]);
133        }
134        if (arr[5] != null) {
135            bean.setStartTime((Timestamp) arr[5]);
136        }
137        if (arr[6] != null) {
138            bean.setEndTime((Timestamp) arr[6]);
139        }
140        if (arr[7] != null) {
141            bean.setAppPath((String) arr[7]);
142        }
143        if (arr[8] != null) {
144            bean.setConcurrency(((Integer) arr[8]).intValue());
145        }
146        if (arr[9] != null) {
147            bean.setFrequency((String) arr[9]);
148        }
149        if (arr[10] != null) {
150            bean.setLastActionTime((Timestamp) arr[10]);
151        }
152        if (arr[11] != null) {
153            bean.setNextMaterializedTime((Timestamp) arr[11]);
154        }
155        if (arr[12] != null) {
156            bean.setCreatedTime((Timestamp) arr[12]);
157        }
158        if (arr[13] != null) {
159            bean.setTimeUnit(Timeunit.valueOf((String) arr[13]));
160        }
161        if (arr[14] != null) {
162            bean.setTimeZone((String) arr[14]);
163        }
164        if (arr[15] != null) {
165            bean.setTimeout((Integer) arr[15]);
166        }
167        return bean;
168    }
169}