diff --git a/execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java b/execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java
index c75c2a2a4..82017e3ca 100644
--- a/execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java
+++ b/execution/src/main/java/org/springframework/batch/execution/repository/dao/SqlJobDao.java
@@ -31,6 +31,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * SQL implementation of {@link JobDao}. Uses sequences (via Spring's
@@ -47,28 +48,35 @@ import org.springframework.util.Assert;
  */
 public class SqlJobDao implements JobDao, InitializingBean {
 
+	/**
+	 * Default value for the table prefix property.
+	 */
+	public static final String DEFAULT_TABLE_PREFIX = "BATCH_";
+
+	private static String tablePrefix = DEFAULT_TABLE_PREFIX;
+
 	// Job SQL statements
-	private static final String CREATE_JOB = "INSERT into BATCH_JOB(ID, JOB_NAME, JOB_STREAM, SCHEDULE_DATE, JOB_RUN)"
+	private static final String CREATE_JOB = "INSERT into %PREFIX%JOB(ID, JOB_NAME, JOB_STREAM, SCHEDULE_DATE, JOB_RUN)"
 			+ " values (?, ?, ?, ?, ?)";
 
-	private static final String FIND_JOBS = "SELECT ID, STATUS from BATCH_JOB where JOB_NAME = ? and "
+	private static final String FIND_JOBS = "SELECT ID, STATUS from %PREFIX%JOB where JOB_NAME = ? and "
 			+ "JOB_STREAM = ? and SCHEDULE_DATE = ? and JOB_RUN = ?";
 
-	private static final String UPDATE_JOB = "UPDATE BATCH_JOB set STATUS = ? where ID = ?";
+	private static final String UPDATE_JOB = "UPDATE %PREFIX%JOB set STATUS = ? where ID = ?";
 
-	private static final String GET_JOB_EXECUTION_COUNT = "SELECT count(ID) from BATCH_JOB_EXECUTION "
+	private static final String GET_JOB_EXECUTION_COUNT = "SELECT count(ID) from %PREFIX%JOB_EXECUTION "
 			+ "where JOB_ID = ?";
 
 	// Job Execution SqlStatements
-	private static final String UPDATE_JOB_EXECUTION = "UPDATE BATCH_JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
+	private static final String UPDATE_JOB_EXECUTION = "UPDATE %PREFIX%JOB_EXECUTION set START_TIME = ?, END_TIME = ?, "
 			+ " STATUS = ? where ID = ?";
 
-	private static final String SAVE_JOB_EXECUTION = "INSERT into BATCH_JOB_EXECUTION(ID, JOB_ID, START_TIME, END_TIME, STATUS)"
+	private static final String SAVE_JOB_EXECUTION = "INSERT into %PREFIX%JOB_EXECUTION(ID, JOB_ID, START_TIME, END_TIME, STATUS)"
 			+ " values (?, ?, ?, ?, ?)";
 
-	private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM BATCH_JOB_EXECUTION WHERE ID=?";
+	private static final String CHECK_JOB_EXECUTION_EXISTS = "SELECT COUNT(*) FROM %PREFIX%JOB_EXECUTION WHERE ID=?";
 
-	private static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from BATCH_JOB_EXECUTION"
+	private static final String FIND_JOB_EXECUTIONS = "SELECT ID, START_TIME, END_TIME, STATUS from %PREFIX%JOB_EXECUTION"
 			+ " where JOB_ID = ?";
 
 	private JdbcTemplate jdbcTemplate;
@@ -77,6 +85,18 @@ public class SqlJobDao implements JobDao, InitializingBean {
 
 	private DataFieldMaxValueIncrementer jobExecutionIncrementer;
 
+	/**
+	 * Public setter for the table prefix property. This will be prefixed to all
+	 * the table names before queries are executed. Defaults to
+	 * {@value #DEFAULT_TABLE_PREFIX}.
+	 * 
+	 * @param tablePrefix
+	 *            the tablePrefix to set
+	 */
+	public void setTablePrefix(String tablePrefix) {
+		SqlJobDao.tablePrefix = tablePrefix;
+	}
+
 	/**
 	 * In this sql implementation a job id is obtained by asking the
 	 * jobIncrementer (which is likely a sequence) for the nextLong, and then
@@ -94,12 +114,10 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		ScheduledJobIdentifier defaultJobId = getScheduledJobIdentifier(jobIdentifier);
 
 		Long jobId = new Long(jobIncrementer.nextLongValue());
-		Object[] parameters = new Object[] { jobId,
-				defaultJobId.getName(),
-				defaultJobId.getJobStream(),
-				defaultJobId.getScheduleDate(),
+		Object[] parameters = new Object[] { jobId, defaultJobId.getName(),
+				defaultJobId.getJobStream(), defaultJobId.getScheduleDate(),
 				new Long(defaultJobId.getJobRun()) };
-		jdbcTemplate.update(CREATE_JOB, parameters);
+		jdbcTemplate.update(getCreateJobQuery(), parameters);
 
 		JobInstance job = new JobInstance(jobId);
 		job.setIdentifier(jobIdentifier);
@@ -107,8 +125,8 @@ public class SqlJobDao implements JobDao, InitializingBean {
 	}
 
 	/**
-	 * The BATCH_JOB table is queried for <strong>any</strong> jobs that match
-	 * the given identifier, adding them to a list via the RowMapper callback.
+	 * The job table is queried for <strong>any</strong> jobs that match the
+	 * given identifier, adding them to a list via the RowMapper callback.
 	 * 
 	 * @see JobDao#findJobs(JobIdentifier)
 	 * @throws IllegalArgumentException
@@ -135,7 +153,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
 			}
 		};
 
-		return jdbcTemplate.query(FIND_JOBS, parameters, rowMapper);
+		return jdbcTemplate.query(getFindJobsQuery(), parameters, rowMapper);
 	}
 
 	/**
@@ -151,7 +169,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
 
 		Object[] parameters = new Object[] { job.getStatus().toString(),
 				job.getId() };
-		jdbcTemplate.update(UPDATE_JOB, parameters);
+		jdbcTemplate.update(getUpdateJobQuery(), parameters);
 	}
 
 	/**
@@ -173,7 +191,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		Object[] parameters = new Object[] { jobExecution.getId(),
 				jobExecution.getJobId(), jobExecution.getStartTime(),
 				jobExecution.getEndTime(), jobExecution.getStatus().toString() };
-		jdbcTemplate.update(SAVE_JOB_EXECUTION, parameters);
+		jdbcTemplate.update(getSaveJobExecutionQuery(), parameters);
 	}
 
 	/**
@@ -201,14 +219,14 @@ public class SqlJobDao implements JobDao, InitializingBean {
 		// Check if given JobExecution's Id already exists, if none is found it
 		// is invalid and
 		// an exception should be thrown.
-		if (jdbcTemplate.queryForInt(CHECK_JOB_EXECUTION_EXISTS,
+		if (jdbcTemplate.queryForInt(getCheckJobExecutionExistsQuery(),
 				new Object[] { jobExecution.getId() }) != 1) {
 			throw new NoSuchBatchDomainObjectException(
 					"Invalid JobExecution, ID " + jobExecution.getId()
 							+ " not found.");
 		}
 
-		jdbcTemplate.update(UPDATE_JOB_EXECUTION, parameters);
+		jdbcTemplate.update(getUpdateJobExecutionQuery(), parameters);
 	}
 
 	/**
@@ -222,7 +240,7 @@ public class SqlJobDao implements JobDao, InitializingBean {
 
 		Object[] parameters = new Object[] { jobId };
 
-		return jdbcTemplate.queryForInt(GET_JOB_EXECUTION_COUNT, parameters);
+		return jdbcTemplate.queryForInt(getJobExecutionCountQuery(), parameters);
 	}
 
 	public List findJobExecutions(JobInstance job) {
@@ -247,10 +265,46 @@ public class SqlJobDao implements JobDao, InitializingBean {
 
 		};
 
-		return jdbcTemplate.query(FIND_JOB_EXECUTIONS, new Object[] { jobId },
+		return jdbcTemplate.query(getFindJobExecutionsQuery(), new Object[] { jobId },
 				rowMapper);
 	}
 
+	private String getQuery(String base) {
+		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
+	}
+	
+	private String getCreateJobQuery() {
+		return getQuery(CREATE_JOB);
+	}
+	
+	private String getFindJobsQuery() {
+		return getQuery(FIND_JOBS);
+	}
+	
+	private String getUpdateJobQuery() {
+		return getQuery(UPDATE_JOB);
+	}
+	
+	private String getSaveJobExecutionQuery() {
+		return getQuery(SAVE_JOB_EXECUTION);
+	}
+	
+	private String getUpdateJobExecutionQuery() {
+		return getQuery(UPDATE_JOB_EXECUTION);
+	}
+	
+	private String getCheckJobExecutionExistsQuery() {
+		return getQuery(CHECK_JOB_EXECUTION_EXISTS);
+	}
+	
+	private String getJobExecutionCountQuery() {
+		return getQuery(GET_JOB_EXECUTION_COUNT);
+	}
+
+	private String getFindJobExecutionsQuery() {
+		return getQuery(FIND_JOB_EXECUTIONS);
+	}
+
 	public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
 		this.jdbcTemplate = jdbcTemplate;
 	}
diff --git a/execution/src/test/java/org/springframework/batch/execution/repository/dao/BaseJobDaoTests.java b/execution/src/test/java/org/springframework/batch/execution/repository/dao/BaseJobDaoTests.java
new file mode 100644
index 000000000..3a9c04d4d
--- /dev/null
+++ b/execution/src/test/java/org/springframework/batch/execution/repository/dao/BaseJobDaoTests.java
@@ -0,0 +1,251 @@
+/*
+ * Copyright 2006-2007 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.execution.repository.dao;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.text.SimpleDateFormat;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.batch.core.domain.BatchStatus;
+import org.springframework.batch.core.domain.JobExecution;
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.repository.NoSuchBatchDomainObjectException;
+import org.springframework.batch.core.runtime.SimpleJobIdentifier;
+import org.springframework.batch.execution.runtime.ScheduledJobIdentifier;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
+import org.springframework.util.ClassUtils;
+
+public class BaseJobDaoTests extends
+		AbstractTransactionalDataSourceSpringContextTests {
+
+	private static final String GET_JOB_EXECUTION = "SELECT JOB_ID, START_TIME, END_TIME, STATUS from "
+			+ "BATCH_JOB_EXECUTION where ID = ?";
+
+	protected JobDao jobDao;
+
+	protected ScheduledJobIdentifier jobRuntimeInformation;
+
+	protected JobInstance job;
+
+	protected JobExecution jobExecution;
+
+	protected Timestamp jobExecutionStartTime = new Timestamp(System
+			.currentTimeMillis());
+
+	protected String[] getConfigLocations() {
+		return new String[] { ClassUtils.addResourcePathToPackagePath(
+				getClass(), "sql-dao-test.xml") };
+	}
+
+	/*
+	 * Because AbstractTransactionalSpringContextTests is used, this method will
+	 * be called by Spring to set the JobRepository.
+	 */
+	public void setJobRepositoryDao(JobDao jobRepositoryDao) {
+		this.jobDao = jobRepositoryDao;
+	}
+
+	protected void onSetUpInTransaction() throws Exception {
+		jobRuntimeInformation = new ScheduledJobIdentifier("Job1");
+		jobRuntimeInformation.setName("Job1");
+		jobRuntimeInformation.setJobStream("TestStream");
+		jobRuntimeInformation.setJobRun(1);
+		jobRuntimeInformation.setScheduleDate(new SimpleDateFormat("yyyyMMdd")
+				.parse("20070505"));
+
+		// Create job.
+		job = jobDao.createJob(jobRuntimeInformation);
+
+		// Create an execution
+		jobExecutionStartTime = new Timestamp(System.currentTimeMillis());
+		jobExecution = new JobExecution(job.getId());
+		jobExecution.setStartTime(jobExecutionStartTime);
+		jobExecution.setStatus(BatchStatus.STARTED);
+		jobDao.save(jobExecution);
+	}
+
+	public void testVersionIsNotNullForJob() throws Exception {
+		int version = jdbcTemplate
+				.queryForInt("select version from BATCH_JOB where ID="
+						+ job.getId());
+		assertEquals(0, version);
+	}
+
+	public void testVersionIsNotNullForJobExecution() throws Exception {
+		int version = jdbcTemplate
+				.queryForInt("select version from BATCH_JOB_EXECUTION where ID="
+						+ jobExecution.getId());
+		assertEquals(0, version);
+	}
+
+	public void testFindNonExistentJob() {
+		// No job should be found since it hasn't been created.
+		List jobs = jobDao.findJobs(new ScheduledJobIdentifier("Job2"));
+		assertTrue(jobs.size() == 0);
+	}
+
+	public void testFindJob() {
+
+		List jobs = jobDao.findJobs(jobRuntimeInformation);
+		assertTrue(jobs.size() == 1);
+		JobInstance tempJob = (JobInstance) jobs.get(0);
+		assertTrue(job.equals(tempJob));
+		assertEquals(jobRuntimeInformation, tempJob.getIdentifier());
+	}
+
+	public void testFindJobWithNullRuntime() {
+
+		ScheduledJobIdentifier runtimeInformation = null;
+
+		try {
+			jobDao.findJobs(runtimeInformation);
+			fail();
+		} catch (IllegalArgumentException ex) {
+			// expected
+		}
+	}
+
+	public void testUpdateJob() {
+		// Update the returned job with a new status
+		job.setStatus(BatchStatus.COMPLETED);
+		jobDao.update(job);
+
+		// The job just updated should be found, with the saved status.
+		List jobs = jobDao.findJobs(jobRuntimeInformation);
+		assertTrue(jobs.size() == 1);
+		JobInstance tempJob = (JobInstance) jobs.get(0);
+		assertTrue(job.equals(tempJob));
+		assertEquals(tempJob.getStatus(), BatchStatus.COMPLETED);
+	}
+
+	public void testUpdateJobWithNullId() {
+
+		JobInstance testJob = new JobInstance(null);
+		try {
+			jobDao.update(testJob);
+			fail();
+		} catch (IllegalArgumentException ex) {
+			// expected
+		}
+	}
+
+	public void testUpdateNullJob() {
+
+		JobInstance testJob = null;
+		try {
+			jobDao.update(testJob);
+		} catch (IllegalArgumentException ex) {
+			// expected
+		}
+	}
+
+	public void testUpdateJobExecution() {
+
+		jobExecution.setStatus(BatchStatus.COMPLETED);
+		jobExecution.setEndTime(new Timestamp(System.currentTimeMillis()));
+		jobDao.update(jobExecution);
+
+		List executions = retrieveJobExecution(jobExecution.getId());
+		assertEquals(executions.size(), 1);
+		assertEquals(jobExecution, ((JobExecution) executions.get(0)));
+	}
+
+	public void testUpdateInvalidJobExecution() {
+
+		JobExecution execution = new JobExecution(job.getId());
+		// id is invalid
+		execution.setId(new Long(29432));
+		try {
+			jobDao.update(execution);
+			fail();
+		} catch (NoSuchBatchDomainObjectException ex) {
+			// expected
+		}
+	}
+
+	public void testUpdateNullIdJobExection() {
+
+		JobExecution execution = new JobExecution(job.getId());
+		try {
+			jobDao.update(execution);
+			fail();
+		} catch (IllegalArgumentException ex) {
+			// expected
+		}
+	}
+
+	public void testIncrementExecutionCount() {
+
+		// 1 JobExection already added in setup
+		assertEquals(jobDao.getJobExecutionCount(job.getId()), 1);
+
+		// Save new JobExecution for same job
+		JobExecution testJobExecution = new JobExecution(job.getId());
+		jobDao.save(testJobExecution);
+		// JobExecutionCount should be incremented by 1
+		assertEquals(jobDao.getJobExecutionCount(job.getId()), 2);
+	}
+
+	public void testZeroExecutionCount() {
+
+		JobInstance testJob = jobDao.createJob(new ScheduledJobIdentifier(
+				"TestJob"));
+		// no jobExecutions saved for new job, count should be 0
+		assertEquals(jobDao.getJobExecutionCount(testJob.getId()), 0);
+	}
+
+	public void testJobWithSimpleJobIdentifier() throws Exception {
+		SimpleJobIdentifier jobIdentifier = new SimpleJobIdentifier("Job1");
+
+		// Create job.
+		job = jobDao.createJob(jobIdentifier);
+
+		// sessionFactory.getCurrentSession().flush();
+
+		List jobs = jdbcTemplate.queryForList(
+				"SELECT * FROM BATCH_JOB where ID=?", new Object[] { job
+						.getId() });
+		assertEquals(1, jobs.size());
+		assertEquals(job.getName(), ((Map) jobs.get(0)).get("JOB_NAME"));
+
+	}
+
+	private List retrieveJobExecution(final Long id) {
+
+		RowMapper rowMapper = new RowMapper() {
+			public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
+
+				JobExecution execution = new JobExecution(new Long(rs
+						.getLong(1)));
+				execution.setStartTime(rs.getTimestamp(2));
+				execution.setEndTime(rs.getTimestamp(3));
+				execution.setStatus(BatchStatus.getStatus(rs.getString(4)));
+				execution.setId(id);
+
+				return execution;
+			}
+		};
+
+		return jdbcTemplate.query(GET_JOB_EXECUTION, new Object[] { id },
+				rowMapper);
+	}
+
+}
diff --git a/execution/src/test/java/org/springframework/batch/execution/repository/dao/HibernateJobDaoTests.java b/execution/src/test/java/org/springframework/batch/execution/repository/dao/HibernateJobDaoTests.java
index a8e29ae63..30391e4ac 100644
--- a/execution/src/test/java/org/springframework/batch/execution/repository/dao/HibernateJobDaoTests.java
+++ b/execution/src/test/java/org/springframework/batch/execution/repository/dao/HibernateJobDaoTests.java
@@ -24,7 +24,7 @@ import org.hibernate.SessionFactory;
 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.util.ClassUtils;
 
-public class HibernateJobDaoTests extends SqlJobDaoTests {
+public class HibernateJobDaoTests extends BaseJobDaoTests {
 	
 	private SessionFactory sessionFactory;