RESOLVED - issue BATCH-1340: Add Support for Native Queries to Jpa/Hibernate readers

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateCursorItemReader.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.batch.item.database;
 
+import java.util.Map;
+
 import org.hibernate.Query;
 import org.hibernate.ScrollMode;
 import org.hibernate.ScrollableResults;
@@ -25,6 +27,7 @@ import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.database.support.AbstractHibernateQueryProvider;
 import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
@@ -68,6 +71,10 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 
 	private String queryName = "";
 
+	private Map<String, Object> parameterValues;
+
+	private HibernateQueryProvider queryProvider;
+
 	private boolean useStatelessSession = true;
 
 	private boolean initialized = false;
@@ -77,29 +84,53 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 	public HibernateCursorItemReader() {
 		setName(ClassUtils.getShortName(HibernateCursorItemReader.class));
 	}
+	
+	/**
+	 * The parameter values to apply to a query (map of name:value).
+	 * 
+	 * @param parameterValues the parameter values to set
+	 */
+	public void setParameterValues(Map<String, Object> parameterValues) {
+		this.parameterValues = parameterValues;
+	}
 
 	/**
 	 * Open appropriate type of hibernate session and create the query.
 	 */
 	private Query createQuery() {
+
 		if (useStatelessSession) {
 			statelessSession = sessionFactory.openStatelessSession();
-			if (StringUtils.hasText(queryName)) {
-				return statelessSession.getNamedQuery(queryName);
+			if (queryProvider != null) {
+				queryProvider.setStatelessSession(statelessSession);
 			}
 			else {
-				return statelessSession.createQuery(queryString);
+				if (StringUtils.hasText(queryName)) {
+					return statelessSession.getNamedQuery(queryName);
+				}
+				else {
+					return statelessSession.createQuery(queryString);
+				}
 			}
 		}
 		else {
 			statefulSession = sessionFactory.openSession();
-			if (StringUtils.hasText(queryName)) {
-				return statefulSession.getNamedQuery(queryName);
+			if (queryProvider != null) {
+				queryProvider.setSession(statefulSession);
 			}
 			else {
-				return statefulSession.createQuery(queryString);
+				if (StringUtils.hasText(queryName)) {
+					return statefulSession.getNamedQuery(queryName);
+				}
+				else {
+					return statefulSession.createQuery(queryString);
+				}
 			}
 		}
+
+		// if queryProvider is set use it to create a query
+		return queryProvider.createQuery();
+
 	}
 
 	/**
@@ -110,10 +141,21 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 	}
 
 	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(sessionFactory, "session factory must be set");
-		Assert.isTrue(fetchSize >= 0, "fetchSize must not be negative");
-		Assert.isTrue(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName),
-				"exactly one of queryString or queryName must be set");
+
+		Assert.state(sessionFactory!=null, "A SessionFactory must be provided");
+		Assert.state(fetchSize >= 0, "fetchSize must not be negative");
+
+		if (queryProvider == null) {
+			Assert.notNull(sessionFactory, "session factory must be set");
+			Assert.state(StringUtils.hasText(queryString) ^ StringUtils.hasText(queryName),
+					"queryString or queryName must be set");
+		}
+		// making sure that the appropriate (Hibernate) query provider is set
+		else {
+			Assert.state(queryProvider instanceof AbstractHibernateQueryProvider,
+					"Hibernate query provider must be set");
+		}
+
 	}
 
 	/**
@@ -130,6 +172,13 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 		this.queryString = queryString;
 	}
 
+	/**
+	 * @param queryProvider Hibernate query provider
+	 */
+	public void setQueryProvider(HibernateQueryProvider queryProvider) {
+		this.queryProvider = queryProvider;
+	}
+
 	/**
 	 * Can be set only in uninitialized state.
 	 * 
@@ -193,24 +242,27 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 
 	/**
 	 * Open hibernate session and create a forward-only cursor for the
-	 * {@link #setQueryString(String)}.
+	 * query.
 	 */
 	protected void doOpen() throws Exception {
 		Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first");
 
-		cursor = createQuery().setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
+		Query query = createQuery();
+		if (parameterValues!=null) {
+			query.setProperties(parameterValues);
+		}
+		cursor = query.setFetchSize(fetchSize).scroll(ScrollMode.FORWARD_ONLY);
 
 		initialized = true;
-
 	}
-	
+
 	@Override
 	protected void jumpToItem(int itemIndex) throws Exception {
-		for(int i = 0; i < itemIndex; i++){
+		for (int i = 0; i < itemIndex; i++) {
 			cursor.next();
 		}
 	}
-	
+
 	/**
 	 * Close the cursor and hibernate session.
 	 */
@@ -220,6 +272,7 @@ public class HibernateCursorItemReader<T> extends AbstractItemCountingItemStream
 		if (cursor != null) {
 			cursor.close();
 		}
+
 		if (useStatelessSession) {
 			if (statelessSession != null) {
 				statelessSession.close();

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/HibernateQueryProvider.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database;
+
+import org.hibernate.Query;
+import org.hibernate.Session;
+import org.hibernate.StatelessSession;
+import org.springframework.batch.item.ItemReader;
+
+/**
+ * <p>
+ * Interface defining the functionality to be provided for generating queries
+ * for use with Hibernate {@link ItemReader}s or other custom built artifacts.
+ * </p>
+ * 
+ * @author Anatoly Polinsky
+ * @author Dave Syer
+ * 
+ * @since 2.1
+ * 
+ */
+public interface HibernateQueryProvider {
+
+	/**
+	 * <p>
+	 * Create the query object which type will be determined by the underline
+	 * implementation (e.g. Hibernate, JPA, etc.)
+	 * </p>
+	 * 
+	 * @return created query
+	 */
+	Query createQuery();
+
+	/**
+	 * <p>
+	 * Inject a {@link Session} that can be used as a factory for queries. The
+	 * state of the session is controlled by the caller (i.e. it should be
+	 * closed if necessary).
+	 * </p>
+	 * 
+	 * <p>
+	 * Use either this method or {@link #setStatelessSession(StatelessSession)}
+	 * </p>
+	 * 
+	 * @param session the {@link Session} to set
+	 */
+	void setSession(Session session);
+
+	/**
+	 * <p>
+	 * Inject a {@link StatelessSession} that can be used as a factory for
+	 * queries. The state of the session is controlled by the caller (i.e. it
+	 * should be closed if necessary).
+	 * </p>
+	 * 
+	 * <p>
+	 * Use either this method or {@link #setSession(Session)}
+	 * </p>
+	 * 
+	 * @param session the {@link StatelessSession} to set
+	 */
+	void setStatelessSession(StatelessSession session);
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaPagingItemReader.java

@@ -26,6 +26,7 @@ import javax.persistence.EntityTransaction;
 import javax.persistence.Query;
 
 import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.database.support.AbstractJpaQueryProvider;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -90,12 +91,27 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
 
 	private String queryString;
 
+	private JpaQueryProvider queryProvider;
+
 	private Map<String, Object> parameterValues;
 
 	public JpaPagingItemReader() {
 		setName(ClassUtils.getShortName(JpaPagingItemReader.class));
 	}
 
+	/**
+	 * Create a query using an appropriate query provider (entityManager OR
+	 * queryProvider).
+	 */
+	private Query createQuery() {
+		if (queryProvider == null) {
+			return entityManager.createQuery(queryString);
+		}
+		else {
+			return queryProvider.createQuery();
+		}
+	}
+
 	public void setEntityManagerFactory(EntityManagerFactory entityManagerFactory) {
 		this.entityManagerFactory = entityManagerFactory;
 	}
@@ -112,8 +128,15 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
 
 	public void afterPropertiesSet() throws Exception {
 		super.afterPropertiesSet();
-		Assert.notNull(entityManagerFactory);
-		Assert.hasLength(queryString);
+
+		if (queryProvider == null) {
+			Assert.notNull(entityManagerFactory);
+			Assert.hasLength(queryString);
+		}
+		// making sure that the appropriate (JPA) query provider is set
+		else {
+			Assert.isTrue(queryProvider instanceof AbstractJpaQueryProvider, "JPA query provider must be set");
+		}
 	}
 
 	/**
@@ -123,13 +146,27 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
 		this.queryString = queryString;
 	}
 
+	/**
+	 * @param queryProvider JPA query provider
+	 */
+	public void setQueryProvider(JpaQueryProvider queryProvider) {
+		this.queryProvider = queryProvider;
+	}
+
 	@Override
 	protected void doOpen() throws Exception {
 		super.doOpen();
+
 		entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap);
 		if (entityManager == null) {
 			throw new DataAccessResourceFailureException("Unable to obtain an EntityManager");
 		}
+		// set entityManager to queryProvider, so it participates
+		// in JpaPagingItemReader's managed transaction
+		if (queryProvider != null) {
+			queryProvider.setEntityManager(entityManager);
+		}
+
 	}
 
 	@Override
@@ -142,8 +179,7 @@ public class JpaPagingItemReader<T> extends AbstractPagingItemReader<T> {
 		entityManager.flush();
 		entityManager.clear();
 
-		Query query = entityManager.createQuery(queryString).setFirstResult(getPage() * getPageSize()).setMaxResults(
-				getPageSize());
+		Query query = createQuery().setFirstResult(getPage() * getPageSize()).setMaxResults(getPageSize());
 
 		if (parameterValues != null) {
 			for (Map.Entry<String, Object> me : parameterValues.entrySet()) {

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JpaQueryProvider.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database;
+
+import javax.persistence.EntityManager;
+import javax.persistence.Query;
+
+import org.springframework.batch.item.ItemReader;
+
+/**
+ * <p>Interface defining the functionality to be provided for generating queries 
+ * for use with JPA {@link ItemReader}s or other custom built artifacts.</p>
+ * 
+ * @author Anatoly Polinsky
+ * @author Dave Syer
+ * @since 2.1
+ * 
+ */
+public interface JpaQueryProvider {
+
+    /**
+     * <p>Create the query object.</p>
+     * 
+     * @return created query
+     */
+	public Query createQuery();
+	
+    /**
+     * Provide an {@link EntityManager} for the query to be built.
+     * 
+     * @param entityManager
+     */
+    void setEntityManager(EntityManager entityManager);
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractHibernateQueryProvider.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database.support;
+
+import org.hibernate.Query;
+import org.hibernate.Session;
+import org.hibernate.StatelessSession;
+import org.springframework.batch.item.database.HibernateQueryProvider;
+
+/**
+ * <p>Abstract Hibernate Query Provider to serve as a base class for all 
+ * Hibernate {@link Query} providers.</p>
+ * 
+ * <p>The implementing provider can be configured to use either 
+ * {@link StatelessSession} sufficient for simple mappings without the need 
+ * to cascade to associated objects or standard Hibernate {@link Session} 
+ * for more advanced mappings or when caching is desired.</p>
+ * 
+ * @author Anatoly Polinsky
+ * @author Dave Syer
+ * 
+ * @since 2.1
+ *
+ */
+public abstract class AbstractHibernateQueryProvider implements HibernateQueryProvider {
+    
+    private StatelessSession statelessSession;
+    private Session statefulSession;
+    
+    public void setStatelessSession(StatelessSession statelessSession) {
+		this.statelessSession = statelessSession;
+	}
+
+	public void setSession(Session statefulSession) {
+		this.statefulSession = statefulSession;
+	}
+
+	public boolean isStatelessSession() {
+        return this.statefulSession==null && this.statelessSession!=null;
+    }
+
+	protected StatelessSession getStatelessSession() {
+	    return statelessSession;
+	}
+
+    protected Session getStatefulSession() {
+        return statefulSession;
+    }	
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/AbstractJpaQueryProvider.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database.support;
+
+import javax.persistence.EntityManager;
+import javax.persistence.Query;
+
+import org.springframework.batch.item.database.JpaQueryProvider;
+import org.springframework.batch.item.database.HibernateQueryProvider;
+import org.springframework.beans.factory.InitializingBean;
+
+/**
+ * <p>
+ * Abstract JPA Query Provider to serve as a base class for all JPA
+ * {@link Query} providers.
+ * </p>
+ * 
+ * @author Anatoly Polinsky
+ * @author Dave Syer
+ * 
+ * @since 2.1
+ */
+public abstract class AbstractJpaQueryProvider implements JpaQueryProvider, InitializingBean {
+
+	private EntityManager entityManager;
+
+	/**
+	 * <p>
+	 * Public setter to override the entityManager that was created by this
+	 * {@link HibernateQueryProvider}. This is currently needed to allow
+	 * {@link HibernateQueryProvider} to participate in a user's managed transaction.
+	 * </p>
+	 * 
+	 * @param entityManager
+	 */
+	public void setEntityManager(EntityManager entityManager) {
+		this.entityManager = entityManager;
+	}
+
+	/**
+	 * <p>
+	 * Getter for {@link EntityManager}
+	 * </p>
+	 * 
+	 * @return entityManager the injected {@link EntityManager}
+	 */
+	protected EntityManager getEntityManager() {
+		return entityManager;
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/HibernateNativeQueryProvider.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database.support;
+
+import org.hibernate.Query;
+import org.hibernate.SQLQuery;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>
+ * This query provider creates Hibernate {@link Query}s from injected native SQL
+ * queries. This is useful if there is a need to utilize database-specific
+ * features such as query hints, the CONNECT keyword in Oracle, etc.
+ * </p>
+ * 
+ * @author Anatoly Polinsky
+ * 
+ * @param <E> entity returned by executing the query
+ */
+public class HibernateNativeQueryProvider<E> extends AbstractHibernateQueryProvider {
+
+	private String sqlQuery;
+
+	private Class<E> entityClass;
+
+	/**
+	 * <p>
+	 * Create an {@link SQLQuery} from the session provided (preferring
+	 * stateless if both are available).
+	 * </p>
+	 */
+	public SQLQuery createQuery() {
+
+		if (isStatelessSession()) {
+			return getStatelessSession().createSQLQuery(sqlQuery).addEntity(entityClass);
+		}
+		else {
+			return getStatefulSession().createSQLQuery(sqlQuery).addEntity(entityClass);
+		}
+	}
+
+	public void setSqlQuery(String sqlQuery) {
+		this.sqlQuery = sqlQuery;
+	}
+
+	public void setEntityClass(Class<E> entityClazz) {
+		this.entityClass = entityClazz;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.isTrue(StringUtils.hasText(sqlQuery), "Native SQL query cannot be empty");
+		Assert.notNull(entityClass, "Entity class cannot be NULL");
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/JpaNativeQueryProvider.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.batch.item.database.support;
+
+import javax.persistence.Query;
+
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * <p>
+ * This query provider creates JPA {@link Query}s from injected native SQL
+ * queries. This is useful if there is a need to utilize database-specific
+ * features such as query hints, the CONNECT keyword in Oracle, etc.
+ * </p>
+ * 
+ * @author Anatoly Polinsky
+ * 
+ * @param <E> entity returned by executing the query
+ */
+public class JpaNativeQueryProvider<E> extends AbstractJpaQueryProvider {
+
+	private Class<E> entityClass;
+
+	private String sqlQuery;
+
+	public Query createQuery() {
+		return getEntityManager().createNativeQuery(sqlQuery, entityClass);
+	}
+
+	public void setSqlQuery(String sqlQuery) {
+		this.sqlQuery = sqlQuery;
+	}
+
+	public void setEntityClass(Class<E> entityClazz) {
+		this.entityClass = entityClazz;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		Assert.isTrue(StringUtils.hasText(sqlQuery), "Native SQL query cannot be empty");
+		Assert.notNull(entityClass, "Entity class cannot be NULL");
+	}
+}