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