From bf3ea7ab0aaa42c00811f1bd8642f91dcd9d1e8c Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Wed, 27 Nov 2019 11:29:49 +0100
Subject: [PATCH] DATACASS-704 - Add utility to initialize and cleanup the CQL
 keyspace.

Original pull request: #167.
---
 .../config/CassandraNamespaceHandler.java     |   1 +
 ...nitializeKeyspaceBeanDefinitionParser.java | 133 ++++
 .../config/SortedResourcesFactoryBean.java    |  81 +++
 .../init/CannotReadScriptException.java       |  38 +
 .../init/CompositeKeyspacePopulator.java      |  84 +++
 .../cql/session/init/KeyspacePopulator.java   |  46 ++
 .../init/ResourceKeyspacePopulator.java       | 266 +++++++
 .../cql/session/init/ScriptException.java     |  48 ++
 .../session/init/ScriptParseException.java    |  55 ++
 .../init/ScriptStatementFailedException.java  |  54 ++
 .../core/cql/session/init/ScriptUtils.java    | 572 +++++++++++++++
 .../init/SessionFactoryInitializer.java       | 110 +++
 .../init/UncategorizedScriptException.java    |  47 ++
 .../core/cql/session/init/package-info.java   |   7 +
 .../cassandra/config/spring-cassandra-3.0.xsd | 684 ++++++++++++++++++
 .../CompositeKeyspacePopulatorUnitTests.java  | 102 +++
 .../ResourceKeyspacePopulatorUnitTests.java   | 158 ++++
 .../session/init/ScriptUtilsUnitTests.java    | 213 ++++++
 .../init/SessionFactoryInitializerTests.java  |  92 +++
 19 files changed, 2791 insertions(+)
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/InitializeKeyspaceBeanDefinitionParser.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SortedResourcesFactoryBean.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CannotReadScriptException.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulator.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/KeyspacePopulator.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulator.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptException.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptParseException.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptStatementFailedException.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtils.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializer.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/UncategorizedScriptException.java
 create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/package-info.java
 create mode 100644 spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-3.0.xsd
 create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulatorUnitTests.java
 create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulatorUnitTests.java
 create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtilsUnitTests.java
 create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializerTests.java

diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraNamespaceHandler.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraNamespaceHandler.java
index 2ac77da01..10e25775f 100644
--- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraNamespaceHandler.java
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraNamespaceHandler.java
@@ -40,5 +40,6 @@ public class CassandraNamespaceHandler extends NamespaceHandlerSupport {
 		registerBeanDefinitionParser("auditing", new CassandraAuditingBeanDefinitionParser());
 		registerBeanDefinitionParser("converter", new CassandraMappingConverterParser());
 		registerBeanDefinitionParser("mapping", new CassandraMappingContextParser());
+		registerBeanDefinitionParser("initialize-keyspace", new InitializeKeyspaceBeanDefinitionParser());
 	}
 }
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/InitializeKeyspaceBeanDefinitionParser.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/InitializeKeyspaceBeanDefinitionParser.java
new file mode 100644
index 000000000..e565d8e7e
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/InitializeKeyspaceBeanDefinitionParser.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.config;
+
+import java.util.List;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.TypedStringValue;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedList;
+import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.data.cassandra.core.cql.session.init.CompositeKeyspacePopulator;
+import org.springframework.data.cassandra.core.cql.session.init.ResourceKeyspacePopulator;
+import org.springframework.data.cassandra.core.cql.session.init.SessionFactoryInitializer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.StringUtils;
+import org.springframework.util.xml.DomUtils;
+
+import org.w3c.dom.Element;
+
+/**
+ * {@link org.springframework.beans.factory.xml.BeanDefinitionParser} that parses an {@code initialize-keyspace} element
+ * and creates a {@link BeanDefinition} of type {@link DataSourceInitializer}. Picks up nested {@code script} elements
+ * and configures a {@link ResourceKeyspacePopulator} for them.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class InitializeKeyspaceBeanDefinitionParser extends AbstractBeanDefinitionParser {
+
+	@Override
+	protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SessionFactoryInitializer.class);
+		builder.addPropertyReference("sessionFactory", element.getAttribute("session-factory"));
+		builder.addPropertyValue("enabled", element.getAttribute("enabled"));
+
+		parseKeyspacePopulator(element, builder);
+
+		builder.getRawBeanDefinition().setSource(parserContext.extractSource(element));
+
+		return builder.getBeanDefinition();
+	}
+
+	@Override
+	protected boolean shouldGenerateId() {
+		return true;
+	}
+
+	public static void parseKeyspacePopulator(Element element, BeanDefinitionBuilder builder) {
+		List<Element> scripts = DomUtils.getChildElementsByTagName(element, "script");
+		if (!scripts.isEmpty()) {
+			builder.addPropertyValue("keyspacePopulator", createKeyspacePopulator(element, scripts, "INIT"));
+			builder.addPropertyValue("keyspaceCleaner", createKeyspacePopulator(element, scripts, "DESTROY"));
+		}
+	}
+
+	private static BeanDefinition createKeyspacePopulator(Element element, List<Element> scripts, String execution) {
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CompositeKeyspacePopulator.class);
+
+		boolean ignoreFailedDrops = element.getAttribute("ignore-failures").equals("DROPS");
+		boolean continueOnError = element.getAttribute("ignore-failures").equals("ALL");
+
+		ManagedList<BeanMetadataElement> delegates = new ManagedList<>();
+		for (Element scriptElement : scripts) {
+
+			String executionAttr = scriptElement.getAttribute("execution");
+
+			if (!StringUtils.hasText(executionAttr)) {
+				executionAttr = "INIT";
+			}
+			if (!execution.equals(executionAttr)) {
+				continue;
+			}
+
+			BeanDefinitionBuilder delegate = BeanDefinitionBuilder.genericBeanDefinition(ResourceKeyspacePopulator.class);
+			delegate.addPropertyValue("ignoreFailedDrops", ignoreFailedDrops);
+			delegate.addPropertyValue("continueOnError", continueOnError);
+
+			// Use a factory bean for the resources so they can be given an order if a pattern is used
+			BeanDefinitionBuilder resourcesFactory = BeanDefinitionBuilder
+					.genericBeanDefinition(SortedResourcesFactoryBean.class);
+			resourcesFactory.addConstructorArgValue(new TypedStringValue(scriptElement.getAttribute("location")));
+			delegate.addPropertyValue("scripts", resourcesFactory.getBeanDefinition());
+
+			if (StringUtils.hasLength(scriptElement.getAttribute("encoding"))) {
+				delegate.addPropertyValue("cqlScriptEncoding", new TypedStringValue(scriptElement.getAttribute("encoding")));
+			}
+
+			String separator = getSeparator(element, scriptElement);
+			if (separator != null) {
+				delegate.addPropertyValue("separator", new TypedStringValue(separator));
+			}
+			delegates.add(delegate.getBeanDefinition());
+		}
+		builder.addPropertyValue("populators", delegates);
+
+		return builder.getBeanDefinition();
+	}
+
+	@Nullable
+	private static String getSeparator(Element element, Element scriptElement) {
+
+		String scriptSeparator = scriptElement.getAttribute("separator");
+		if (StringUtils.hasLength(scriptSeparator)) {
+			return scriptSeparator;
+		}
+
+		String elementSeparator = element.getAttribute("separator");
+		if (StringUtils.hasLength(elementSeparator)) {
+			return elementSeparator;
+		}
+
+		return null;
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SortedResourcesFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SortedResourcesFactoryBean.java
new file mode 100644
index 000000000..e1d3fb1d0
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SortedResourcesFactoryBean.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.config;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.context.ResourceLoaderAware;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.ResourceLoader;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternUtils;
+
+/**
+ * {@link org.springframework.beans.factory.FactoryBean} implementation that takes a list of location Strings and
+ * creates a sorted array of {@link Resource} instances.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class SortedResourcesFactoryBean extends AbstractFactoryBean<Resource[]> implements ResourceLoaderAware {
+
+	private final List<String> locations;
+
+	private ResourcePatternResolver resourcePatternResolver;
+
+	public SortedResourcesFactoryBean(List<String> locations) {
+		this.locations = locations;
+		this.resourcePatternResolver = new PathMatchingResourcePatternResolver();
+	}
+
+	public SortedResourcesFactoryBean(ResourceLoader resourceLoader, List<String> locations) {
+		this.locations = locations;
+		this.resourcePatternResolver = ResourcePatternUtils.getResourcePatternResolver(resourceLoader);
+	}
+
+	@Override
+	public void setResourceLoader(ResourceLoader resourceLoader) {
+		this.resourcePatternResolver = ResourcePatternUtils.getResourcePatternResolver(resourceLoader);
+	}
+
+	@Override
+	public Class<? extends Resource[]> getObjectType() {
+		return Resource[].class;
+	}
+
+	@Override
+	protected Resource[] createInstance() throws Exception {
+		List<Resource> scripts = new ArrayList<>();
+		for (String location : this.locations) {
+			List<Resource> resources = new ArrayList<>(Arrays.asList(this.resourcePatternResolver.getResources(location)));
+			resources.sort((r1, r2) -> {
+				try {
+					return r1.getURL().toString().compareTo(r2.getURL().toString());
+				} catch (IOException ex) {
+					return 0;
+				}
+			});
+			scripts.addAll(resources);
+		}
+		return scripts.toArray(new Resource[0]);
+	}
+
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CannotReadScriptException.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CannotReadScriptException.java
new file mode 100644
index 000000000..070f17fe0
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CannotReadScriptException.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import org.springframework.core.io.support.EncodedResource;
+
+/**
+ * Thrown by {@link ScriptUtils} if a CQL script cannot be read.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class CannotReadScriptException extends ScriptException {
+
+	/**
+	 * Construct a new {@link CannotReadScriptException}.
+	 *
+	 * @param resource the resource that cannot be read from.
+	 * @param cause the root cause.
+	 */
+	public CannotReadScriptException(EncodedResource resource, Throwable cause) {
+		super("Cannot read CQL script from " + resource, cause);
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulator.java
new file mode 100644
index 000000000..8f236bd68
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulator.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Composite {@link KeyspacePopulator} that delegates to a list of given {@link KeyspacePopulator} implementations,
+ * executing all scripts.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+public class CompositeKeyspacePopulator implements KeyspacePopulator {
+
+	private final List<KeyspacePopulator> populators = new ArrayList<>(4);
+
+	/**
+	 * Create an empty {@link CompositeKeyspacePopulator}.
+	 *
+	 * @see #setPopulators
+	 * @see #addPopulators
+	 */
+	public CompositeKeyspacePopulator() {}
+
+	/**
+	 * Create a {@link CompositeKeyspacePopulator} with the given populators.
+	 *
+	 * @param populators one or more populators to delegate to.
+	 */
+	public CompositeKeyspacePopulator(Collection<KeyspacePopulator> populators) {
+		this.populators.addAll(populators);
+	}
+
+	/**
+	 * Create a {@link CompositeKeyspacePopulator} with the given populators.
+	 *
+	 * @param populators one or more populators to delegate to.
+	 */
+	public CompositeKeyspacePopulator(KeyspacePopulator... populators) {
+		this.populators.addAll(Arrays.asList(populators));
+	}
+
+	/**
+	 * Specify one or more populators to delegate to.
+	 */
+	public void setPopulators(KeyspacePopulator... populators) {
+		this.populators.clear();
+		this.populators.addAll(Arrays.asList(populators));
+	}
+
+	/**
+	 * Add one or more populators to the list of delegates.
+	 */
+	public void addPopulators(KeyspacePopulator... populators) {
+		this.populators.addAll(Arrays.asList(populators));
+	}
+
+	@Override
+	public void populate(Session session) throws ScriptException {
+
+		for (KeyspacePopulator populator : this.populators) {
+			populator.populate(session);
+		}
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/KeyspacePopulator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/KeyspacePopulator.java
new file mode 100644
index 000000000..20bff5245
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/KeyspacePopulator.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Strategy used to populate, initialize, or clean up a Cassandra keyspace.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ * @see ResourceKeyspacePopulator
+ * @see KeyspacePopulatorUtils
+ * @see SessionFactoryInitializer
+ */
+@FunctionalInterface
+public interface KeyspacePopulator {
+
+	/**
+	 * Populate, initialize, or clean up the database using the provided JDBC connection.
+	 * <p>
+	 * Concrete implementations <em>may</em> throw a {@link RuntimeException} if an error is encountered but are
+	 * <em>strongly encouraged</em> to throw a specific {@link ScriptException} instead. For example, Spring's
+	 * {@link ResourceKeyspacePopulator} and {@link KeyspacePopulatorUtils} wrap all exceptions in
+	 * {@code ScriptExceptions}.
+	 *
+	 * @param session the CQL {@link Session} to use to populate the keyspace; already configured and ready to use; never
+	 *          {@literal null}
+	 * @throws ScriptException in all other error cases
+	 * @see KeyspacePopulatorUtils#execute
+	 */
+	void populate(Session session) throws ScriptException;
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulator.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulator.java
new file mode 100644
index 000000000..5cb961fff
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulator.java
@@ -0,0 +1,266 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.EncodedResource;
+import org.springframework.data.cassandra.SessionFactory;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Populates, initializes, or cleans up a keyspace using CQL scripts defined in external resources.
+ * <ul>
+ * <li>Call {@link #addScript} to add a single CQL script location.
+ * <li>Call {@link #addScripts} to add multiple CQL script locations.
+ * <li>Consult the setter methods in this class for further configuration options.
+ * <li>Call {@link #populate} or {@link #execute} to initialize or clean up the database using the configured scripts.
+ * </ul>
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ * @see ScriptUtils
+ */
+public class ResourceKeyspacePopulator implements KeyspacePopulator {
+
+	List<Resource> scripts = new ArrayList<>();
+
+	@Nullable private String cqlScriptEncoding;
+
+	private String separator = ScriptUtils.DEFAULT_STATEMENT_SEPARATOR;
+
+	private String[] commentPrefixes = ScriptUtils.DEFAULT_COMMENT_PREFIXES;
+
+	private String blockCommentStartDelimiter = ScriptUtils.DEFAULT_BLOCK_COMMENT_START_DELIMITER;
+
+	private String blockCommentEndDelimiter = ScriptUtils.DEFAULT_BLOCK_COMMENT_END_DELIMITER;
+
+	private boolean continueOnError = false;
+
+	private boolean ignoreFailedDrops = false;
+
+	/**
+	 * Construct a new {@link ResourceKeyspacePopulator} with default settings.
+	 */
+	public ResourceKeyspacePopulator() {}
+
+	/**
+	 * Construct a new {@link ResourceKeyspacePopulator} with default settings for the supplied scripts.
+	 *
+	 * @param scripts the scripts to execute to initialize or clean up the database (never {@literal null})
+	 */
+	public ResourceKeyspacePopulator(Resource... scripts) {
+		setScripts(scripts);
+	}
+
+	/**
+	 * Construct a new {@link ResourceKeyspacePopulator} with the supplied values.
+	 *
+	 * @param continueOnError flag to indicate that all failures in CQL should be logged but not cause a failure
+	 * @param ignoreFailedDrops flag to indicate that a failed CQL {@code DROP} statement can be ignored
+	 * @param cqlScriptEncoding the encoding for the supplied CQL scripts (may be {@literal null} or <em>empty</em> to
+	 *          indicate platform encoding)
+	 * @param scripts the scripts to execute to initialize or clean up the database (never {@literal null})
+	 */
+	public ResourceKeyspacePopulator(boolean continueOnError, boolean ignoreFailedDrops,
+			@Nullable String cqlScriptEncoding, Resource... scripts) {
+
+		this.continueOnError = continueOnError;
+		this.ignoreFailedDrops = ignoreFailedDrops;
+		setCqlScriptEncoding(cqlScriptEncoding);
+		setScripts(scripts);
+	}
+
+	/**
+	 * Add a script to execute to initialize or clean up the database.
+	 *
+	 * @param script the path to an CQL script (never {@literal null}).
+	 */
+	public void addScript(Resource script) {
+
+		Assert.notNull(script, "'script' must not be null");
+
+		this.scripts.add(script);
+	}
+
+	/**
+	 * Add multiple scripts to execute to initialize or clean up the database.
+	 *
+	 * @param scripts the scripts to execute (never {@literal null}).
+	 */
+	public void addScripts(Resource... scripts) {
+
+		assertContentsOfScriptArray(scripts);
+
+		this.scripts.addAll(Arrays.asList(scripts));
+	}
+
+	/**
+	 * Set the scripts to execute to initialize or clean up the database, replacing any previously added scripts.
+	 *
+	 * @param scripts the scripts to execute (never {@literal null}).
+	 */
+	public void setScripts(Resource... scripts) {
+		assertContentsOfScriptArray(scripts);
+		// Ensure that the list is modifiable
+		this.scripts = new ArrayList<>(Arrays.asList(scripts));
+	}
+
+	private void assertContentsOfScriptArray(Resource... scripts) {
+		Assert.notNull(scripts, "'scripts' must not be null");
+		Assert.noNullElements(scripts, "'scripts' must not contain null elements");
+	}
+
+	/**
+	 * Specify the encoding for the configured CQL scripts, if different from the platform encoding.
+	 *
+	 * @param cqlScriptEncoding the encoding used in scripts (may be {@literal null} or empty to indicate platform
+	 *          encoding).
+	 * @see #addScript(Resource)
+	 */
+	public void setCqlScriptEncoding(@Nullable String cqlScriptEncoding) {
+		this.cqlScriptEncoding = (StringUtils.hasText(cqlScriptEncoding) ? cqlScriptEncoding : null);
+	}
+
+	/**
+	 * Specify the statement separator, if a custom one.
+	 * <p>
+	 * Defaults to {@code ";"} if not specified and falls back to {@code "\n"} as a last resort; may be set to
+	 * {@link ScriptUtils#EOF_STATEMENT_SEPARATOR} to signal that each script contains a single statement without a
+	 * separator.
+	 *
+	 * @param separator the script statement separator.
+	 */
+	public void setSeparator(String separator) {
+		this.separator = separator;
+	}
+
+	/**
+	 * Set the prefix that identifies single-line comments within the CQL scripts.
+	 * <p>
+	 * Defaults to {@code "--"}.
+	 *
+	 * @param commentPrefix the prefix for single-line comments.
+	 * @see #setCommentPrefixes(String...)
+	 */
+	public void setCommentPrefix(String commentPrefix) {
+
+		Assert.hasText(commentPrefix, "'commentPrefix' must not be null or empty");
+
+		this.commentPrefixes = new String[] { commentPrefix };
+	}
+
+	/**
+	 * Set the prefixes that identify single-line comments within the CQL scripts.
+	 * <p>
+	 * Defaults to {@code ["--"]}.
+	 *
+	 * @param commentPrefixes the prefixes for single-line comments.
+	 */
+	public void setCommentPrefixes(String... commentPrefixes) {
+
+		Assert.notEmpty(commentPrefixes, "'commentPrefixes' must not be null or empty");
+		Assert.noNullElements(commentPrefixes, "'commentPrefixes' must not contain null elements");
+
+		this.commentPrefixes = commentPrefixes;
+	}
+
+	/**
+	 * Set the start delimiter that identifies block comments within the CQL scripts.
+	 * <p>
+	 * Defaults to {@code "/*"}.
+	 *
+	 * @param blockCommentStartDelimiter the start delimiter for block comments (never {@literal null} or empty).
+	 * @see #setBlockCommentEndDelimiter
+	 */
+	public void setBlockCommentStartDelimiter(String blockCommentStartDelimiter) {
+
+		Assert.hasText(blockCommentStartDelimiter, "'blockCommentStartDelimiter' must not be null or empty");
+
+		this.blockCommentStartDelimiter = blockCommentStartDelimiter;
+	}
+
+	/**
+	 * Set the end delimiter that identifies block comments within the CQL scripts.
+	 * <p>
+	 * Defaults to <code>"*&#47;"</code>.
+	 *
+	 * @param blockCommentEndDelimiter the end delimiter for block comments (never {@literal null} or empty).
+	 * @see #setBlockCommentStartDelimiter
+	 */
+	public void setBlockCommentEndDelimiter(String blockCommentEndDelimiter) {
+
+		Assert.hasText(blockCommentEndDelimiter, "'blockCommentEndDelimiter' must not be null or empty");
+
+		this.blockCommentEndDelimiter = blockCommentEndDelimiter;
+	}
+
+	/**
+	 * Flag to indicate that all failures in CQL should be logged but not cause a failure.
+	 * <p>
+	 * Defaults to {@literal false}.
+	 *
+	 * @param continueOnError {@literal true} if script execution should continue on error
+	 */
+	public void setContinueOnError(boolean continueOnError) {
+		this.continueOnError = continueOnError;
+	}
+
+	/**
+	 * Flag to indicate that a failed CQL {@code DROP} statement can be ignored.
+	 * <p>
+	 * This is useful for a non-embedded database whose CQL dialect does not support an {@code IF EXISTS} clause in a
+	 * {@code DROP} statement.
+	 * <p>
+	 * The default is {@literal false} so that if the populator runs accidentally, it will fail fast if a script starts
+	 * with a {@code DROP} statement.
+	 *
+	 * @param ignoreFailedDrops {@literal true} if failed drop statements should be ignored.
+	 */
+	public void setIgnoreFailedDrops(boolean ignoreFailedDrops) {
+		this.ignoreFailedDrops = ignoreFailedDrops;
+	}
+
+	@Override
+	public void populate(Session session) throws ScriptException {
+
+		Assert.notNull(session, "session must not be null");
+
+		for (Resource script : this.scripts) {
+			EncodedResource encodedScript = new EncodedResource(script, this.cqlScriptEncoding);
+			ScriptUtils.executeCqlScript(session, encodedScript, this.continueOnError, this.ignoreFailedDrops,
+					this.commentPrefixes, this.separator, this.blockCommentStartDelimiter, this.blockCommentEndDelimiter);
+		}
+	}
+
+	/**
+	 * Execute this {@link ResourceKeyspacePopulator} against the given {@link SessionFactory}.
+	 *
+	 * @param sessionFactory the {@link SessionFactory} to execute against (never {@literal null})
+	 * @throws ScriptException if an error occurs
+	 * @see #populate(Session)
+	 */
+	public void execute(SessionFactory sessionFactory) throws ScriptException {
+		populate(sessionFactory.getSession());
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptException.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptException.java
new file mode 100644
index 000000000..67e9de183
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptException.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.lang.Nullable;
+
+/**
+ * Root of the hierarchy of data access exceptions that are related to processing of CQL scripts.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public abstract class ScriptException extends DataAccessException {
+
+	/**
+	 * Constructor for {@link ScriptException}.
+	 *
+	 * @param message the detail message.
+	 */
+	public ScriptException(String message) {
+		super(message);
+	}
+
+	/**
+	 * Constructor for {@link ScriptException}.
+	 *
+	 * @param message the detail message.
+	 * @param cause the root cause.
+	 */
+	public ScriptException(String message, @Nullable Throwable cause) {
+		super(message, cause);
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptParseException.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptParseException.java
new file mode 100644
index 000000000..a7e115003
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptParseException.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import org.springframework.core.io.support.EncodedResource;
+import org.springframework.lang.Nullable;
+
+/**
+ * Thrown by {@link ScriptUtils} if a CQL script cannot be properly parsed.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class ScriptParseException extends ScriptException {
+
+	/**
+	 * Construct a new {@link ScriptParseException}.
+	 *
+	 * @param message detailed message.
+	 * @param resource the resource from which the CQL script was read.
+	 */
+	public ScriptParseException(String message, @Nullable EncodedResource resource) {
+		super(buildMessage(message, resource));
+	}
+
+	/**
+	 * Construct a new {@link ScriptParseException}.
+	 *
+	 * @param message detailed message.
+	 * @param resource the resource from which the CQL script was read.
+	 * @param cause the root cause.
+	 */
+	public ScriptParseException(String message, @Nullable EncodedResource resource, @Nullable Throwable cause) {
+		super(buildMessage(message, resource), cause);
+	}
+
+	private static String buildMessage(String message, @Nullable EncodedResource resource) {
+		return String.format("Failed to parse CQL script from resource [%s]: %s",
+				(resource == null ? "<unknown>" : resource), message);
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptStatementFailedException.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptStatementFailedException.java
new file mode 100644
index 000000000..c9180e825
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptStatementFailedException.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import org.springframework.core.io.support.EncodedResource;
+
+/**
+ * Thrown by {@link ScriptUtils} if a statement in an SQL script failed when executing it against the target database.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class ScriptStatementFailedException extends ScriptException {
+
+	/**
+	 * Construct a new {@link ScriptStatementFailedException}.
+	 *
+	 * @param stmt the actual CQL statement that failed.
+	 * @param stmtNumber the statement number in the CQL script (i.e., the n<sup>th</sup> statement present in the
+	 *          resource).
+	 * @param encodedResource the resource from which the CQL statement was read.
+	 * @param cause the root cause.
+	 */
+	public ScriptStatementFailedException(String stmt, int stmtNumber, EncodedResource encodedResource, Throwable cause) {
+		super(buildErrorMessage(stmt, stmtNumber, encodedResource), cause);
+	}
+
+	/**
+	 * Build an error message for an CQL script execution failure, based on the supplied arguments.
+	 *
+	 * @param stmt the actual CQL statement that failed.
+	 * @param stmtNumber the statement number in the CQL script (i.e., the n<sup>th</sup> statement present in the
+	 *          resource).
+	 * @param encodedResource the resource from which the CQL statement was read
+	 * @return an error message suitable for an exception's <em>detail message</em> or logging.
+	 */
+	public static String buildErrorMessage(String stmt, int stmtNumber, EncodedResource encodedResource) {
+		return String.format("Failed to execute CQL script statement #%s of %s: %s", stmtNumber, encodedResource, stmt);
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtils.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtils.java
new file mode 100644
index 000000000..709c6d733
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtils.java
@@ -0,0 +1,572 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import java.io.IOException;
+import java.io.LineNumberReader;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.EncodedResource;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import com.datastax.driver.core.ExecutionInfo;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+
+/**
+ * Generic utility methods for working with CQL scripts.
+ * <p>
+ * Mainly for internal use within the framework.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+public abstract class ScriptUtils {
+
+	/**
+	 * Default statement separator within CQL scripts: {@code ";"}.
+	 */
+	public static final String DEFAULT_STATEMENT_SEPARATOR = ";";
+
+	/**
+	 * Fallback statement separator within CQL scripts: {@code "\n"}.
+	 * <p>
+	 * Used if neither a custom separator nor the {@link #DEFAULT_STATEMENT_SEPARATOR} is present in a given script.
+	 */
+	public static final String FALLBACK_STATEMENT_SEPARATOR = "\n";
+
+	/**
+	 * End of file (EOF) CQL statement separator: {@code "^^^ END OF SCRIPT ^^^"}.
+	 * <p>
+	 * This value may be supplied as the {@code separator} to
+	 * {@link #executeCqlScript(Session, EncodedResource, boolean, boolean, String, String, String, String)} to denote
+	 * that an CQL script contains a single statement (potentially spanning multiple lines) with no explicit statement
+	 * separator. Note that such a script should not actually contain this value; it is merely a <em>virtual</em>
+	 * statement separator.
+	 */
+	public static final String EOF_STATEMENT_SEPARATOR = "^^^ END OF SCRIPT ^^^";
+
+	/**
+	 * Default prefix for single-line comments within CQL scripts: {@code "--"}.
+	 */
+	public static final String DEFAULT_COMMENT_PREFIX = "--";
+
+	/**
+	 * Default prefixes for single-line comments within CQL scripts: {@code ["--"]}.
+	 *
+	 * @since 5.2
+	 */
+	public static final String[] DEFAULT_COMMENT_PREFIXES = { DEFAULT_COMMENT_PREFIX };
+
+	/**
+	 * Default start delimiter for block comments within CQL scripts: {@code "/*"}.
+	 */
+	public static final String DEFAULT_BLOCK_COMMENT_START_DELIMITER = "/*";
+
+	/**
+	 * Default end delimiter for block comments within CQL scripts: <code>"*&#47;"</code>.
+	 */
+	public static final String DEFAULT_BLOCK_COMMENT_END_DELIMITER = "*/";
+
+	private static final Log logger = LogFactory.getLog(ScriptUtils.class);
+
+	/**
+	 * Split an CQL script into separate statements delimited by the provided separator character. Each individual
+	 * statement will be added to the provided {@code List}.
+	 * <p>
+	 * Within the script, {@value #DEFAULT_COMMENT_PREFIX} will be used as the comment prefix; any text beginning with the
+	 * comment prefix and extending to the end of the line will be omitted from the output. Similarly,
+	 * {@value #DEFAULT_BLOCK_COMMENT_START_DELIMITER} and {@value #DEFAULT_BLOCK_COMMENT_END_DELIMITER} will be used as
+	 * the <em>start</em> and <em>end</em> block comment delimiters: any text enclosed in a block comment will be omitted
+	 * from the output. In addition, multiple adjacent whitespace characters will be collapsed into a single space.
+	 *
+	 * @param script the CQL script.
+	 * @param separator character separating each statement (typically a ';').
+	 * @param statements the list that will contain the individual statements.
+	 * @throws ScriptException if an error occurred while splitting the CQL script.
+	 * @see #splitCqlScript(String, String, List)
+	 * @see #splitCqlScript(EncodedResource, String, String, String, String, String, List)
+	 */
+	public static void splitCqlScript(String script, char separator, List<String> statements) throws ScriptException {
+		splitCqlScript(script, String.valueOf(separator), statements);
+	}
+
+	/**
+	 * Split an CQL script into separate statements delimited by the provided separator string. Each individual statement
+	 * will be added to the provided {@code List}.
+	 * <p>
+	 * Within the script, {@value #DEFAULT_COMMENT_PREFIX} will be used as the comment prefix; any text beginning with the
+	 * comment prefix and extending to the end of the line will be omitted from the output. Similarly,
+	 * {@value #DEFAULT_BLOCK_COMMENT_START_DELIMITER} and {@value #DEFAULT_BLOCK_COMMENT_END_DELIMITER} will be used as
+	 * the <em>start</em> and <em>end</em> block comment delimiters: any text enclosed in a block comment will be omitted
+	 * from the output. In addition, multiple adjacent whitespace characters will be collapsed into a single space.
+	 *
+	 * @param script the CQL script.
+	 * @param separator text separating each statement (typically a ';' or newline character).
+	 * @param statements the list that will contain the individual statements.
+	 * @throws ScriptException if an error occurred while splitting the CQL script.
+	 * @see #splitCqlScript(String, char, List)
+	 * @see #splitCqlScript(EncodedResource, String, String, String, String, String, List)
+	 */
+	public static void splitCqlScript(String script, String separator, List<String> statements) throws ScriptException {
+		splitCqlScript(null, script, separator, DEFAULT_COMMENT_PREFIX, DEFAULT_BLOCK_COMMENT_START_DELIMITER,
+				DEFAULT_BLOCK_COMMENT_END_DELIMITER, statements);
+	}
+
+	/**
+	 * Split an CQL script into separate statements delimited by the provided separator string. Each individual statement
+	 * will be added to the provided {@code List}.
+	 * <p>
+	 * Within the script, the provided {@code commentPrefix} will be honored: any text beginning with the comment prefix
+	 * and extending to the end of the line will be omitted from the output. Similarly, the provided
+	 * {@code blockCommentStartDelimiter} and {@code blockCommentEndDelimiter} delimiters will be honored: any text
+	 * enclosed in a block comment will be omitted from the output. In addition, multiple adjacent whitespace characters
+	 * will be collapsed into a single space.
+	 *
+	 * @param resource the resource from which the script was read.
+	 * @param script the CQL script.
+	 * @param separator text separating each statement (typically a ';' or newline character).
+	 * @param commentPrefix the prefix that identifies CQL line comments (typically "--").
+	 * @param blockCommentStartDelimiter the <em>start</em> block comment delimiter; never {@literal null} or empty.
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter; never {@literal null} or empty.
+	 * @param statements the list that will contain the individual statements
+	 * @throws ScriptException if an error occurred while splitting the CQL script
+	 */
+	public static void splitCqlScript(@Nullable EncodedResource resource, String script, String separator,
+			String commentPrefix, String blockCommentStartDelimiter, String blockCommentEndDelimiter, List<String> statements)
+			throws ScriptException {
+
+		Assert.hasText(commentPrefix, "'commentPrefix' must not be null or empty");
+		splitCqlScript(resource, script, separator, new String[] { commentPrefix }, blockCommentStartDelimiter,
+				blockCommentEndDelimiter, statements);
+	}
+
+	/**
+	 * Split an CQL script into separate statements delimited by the provided separator string. Each individual statement
+	 * will be added to the provided {@code List}.
+	 * <p>
+	 * Within the script, the provided {@code commentPrefixes} will be honored: any text beginning with one of the comment
+	 * prefixes and extending to the end of the line will be omitted from the output. Similarly, the provided
+	 * {@code blockCommentStartDelimiter} and {@code blockCommentEndDelimiter} delimiters will be honored: any text
+	 * enclosed in a block comment will be omitted from the output. In addition, multiple adjacent whitespace characters
+	 * will be collapsed into a single space.
+	 *
+	 * @param resource the resource from which the script was read.
+	 * @param script the CQL script.
+	 * @param separator text separating each statement (typically a ';' or newline character).
+	 * @param commentPrefixes the prefixes that identify CQL line comments (typically "--").
+	 * @param blockCommentStartDelimiter the <em>start</em> block comment delimiter; never {@literal null} or empty.
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter; never {@literal null} or empty.
+	 * @param statements the list that will contain the individual statements.
+	 * @throws ScriptException if an error occurred while splitting the CQL script
+	 */
+	public static void splitCqlScript(@Nullable EncodedResource resource, String script, String separator,
+			String[] commentPrefixes, String blockCommentStartDelimiter, String blockCommentEndDelimiter,
+			List<String> statements) throws ScriptException {
+
+		Assert.hasText(script, "'script' must not be null or empty");
+		Assert.notNull(separator, "'separator' must not be null");
+		Assert.notEmpty(commentPrefixes, "'commentPrefixes' must not be null or empty");
+		for (String commentPrefix : commentPrefixes) {
+			Assert.hasText(commentPrefix, "'commentPrefixes' must not contain null or empty elements");
+		}
+		Assert.hasText(blockCommentStartDelimiter, "'blockCommentStartDelimiter' must not be null or empty");
+		Assert.hasText(blockCommentEndDelimiter, "'blockCommentEndDelimiter' must not be null or empty");
+
+		StringBuilder sb = new StringBuilder();
+		boolean inSingleQuote = false;
+		boolean inDoubleQuote = false;
+		boolean inEscape = false;
+
+		for (int i = 0; i < script.length(); i++) {
+			char c = script.charAt(i);
+			if (inEscape) {
+				inEscape = false;
+				sb.append(c);
+				continue;
+			}
+			// MyCQL style escapes
+			if (c == '\\') {
+				inEscape = true;
+				sb.append(c);
+				continue;
+			}
+			if (!inDoubleQuote && (c == '\'')) {
+				inSingleQuote = !inSingleQuote;
+			} else if (!inSingleQuote && (c == '"')) {
+				inDoubleQuote = !inDoubleQuote;
+			}
+			if (!inSingleQuote && !inDoubleQuote) {
+				if (script.startsWith(separator, i)) {
+					// We've reached the end of the current statement
+					if (sb.length() > 0) {
+						statements.add(sb.toString());
+						sb = new StringBuilder();
+					}
+					i += separator.length() - 1;
+					continue;
+				} else if (startsWithAny(script, commentPrefixes, i)) {
+					// Skip over any content from the start of the comment to the EOL
+					int indexOfNextNewline = script.indexOf('\n', i);
+					if (indexOfNextNewline > i) {
+						i = indexOfNextNewline;
+						continue;
+					} else {
+						// If there's no EOL, we must be at the end of the script, so stop here.
+						break;
+					}
+				} else if (script.startsWith(blockCommentStartDelimiter, i)) {
+					// Skip over any block comments
+					int indexOfCommentEnd = script.indexOf(blockCommentEndDelimiter, i);
+					if (indexOfCommentEnd > i) {
+						i = indexOfCommentEnd + blockCommentEndDelimiter.length() - 1;
+						continue;
+					} else {
+						throw new ScriptParseException("Missing block comment end delimiter: " + blockCommentEndDelimiter,
+								resource);
+					}
+				} else if (c == ' ' || c == '\r' || c == '\n' || c == '\t') {
+					// Avoid multiple adjacent whitespace characters
+					if (sb.length() > 0 && sb.charAt(sb.length() - 1) != ' ') {
+						c = ' ';
+					} else {
+						continue;
+					}
+				}
+			}
+			sb.append(c);
+		}
+
+		if (StringUtils.hasText(sb)) {
+			statements.add(sb.toString());
+		}
+	}
+
+	/**
+	 * Read a script from the given resource, using "{@code --}" as the comment prefix and "{@code ;}" as the statement
+	 * separator, and build a String containing the lines.
+	 *
+	 * @param resource the {@code EncodedResource} to be read.
+	 * @return {@code String} containing the script lines.
+	 * @throws IOException in case of I/O errors
+	 */
+	static String readScript(EncodedResource resource) throws IOException {
+		return readScript(resource, DEFAULT_COMMENT_PREFIXES, DEFAULT_STATEMENT_SEPARATOR,
+				DEFAULT_BLOCK_COMMENT_END_DELIMITER);
+	}
+
+	/**
+	 * Read a script from the provided resource, using the supplied comment prefixes and statement separator, and build a
+	 * {@code String} containing the lines.
+	 * <p>
+	 * Lines <em>beginning</em> with one of the comment prefixes are excluded from the results; however, line comments
+	 * anywhere else &mdash; for example, within a statement &mdash; will be included in the results.
+	 *
+	 * @param resource the {@code EncodedResource} containing the script to be processed.
+	 * @param commentPrefixes the prefixes that identify comments in the CQL script (typically "--").
+	 * @param separator the statement separator in the CQL script (typically ";").
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter.
+	 * @return a {@code String} containing the script lines
+	 * @throws IOException in case of I/O errors
+	 */
+	private static String readScript(EncodedResource resource, @Nullable String[] commentPrefixes,
+			@Nullable String separator, @Nullable String blockCommentEndDelimiter) throws IOException {
+
+		try (LineNumberReader lnr = new LineNumberReader(resource.getReader())) {
+			return readScript(lnr, commentPrefixes, separator, blockCommentEndDelimiter);
+		}
+	}
+
+	/**
+	 * Read a script from the provided {@code LineNumberReader}, using the supplied comment prefix and statement
+	 * separator, and build a {@code String} containing the lines.
+	 * <p>
+	 * Lines <em>beginning</em> with the comment prefix are excluded from the results; however, line comments anywhere
+	 * else &mdash; for example, within a statement &mdash; will be included in the results.
+	 *
+	 * @param lineNumberReader the {@code LineNumberReader} containing the script to be processed.
+	 * @param lineCommentPrefix the prefix that identifies comments in the CQL script (typically "--").
+	 * @param separator the statement separator in the CQL script (typically ";").
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter.
+	 * @return a {@code String} containing the script lines
+	 * @throws IOException in case of I/O errors
+	 */
+	public static String readScript(LineNumberReader lineNumberReader, @Nullable String lineCommentPrefix,
+			@Nullable String separator, @Nullable String blockCommentEndDelimiter) throws IOException {
+
+		String[] lineCommentPrefixes = (lineCommentPrefix != null) ? new String[] { lineCommentPrefix } : null;
+		return readScript(lineNumberReader, lineCommentPrefixes, separator, blockCommentEndDelimiter);
+	}
+
+	/**
+	 * Read a script from the provided {@code LineNumberReader}, using the supplied comment prefixes and statement
+	 * separator, and build a {@code String} containing the lines.
+	 * <p>
+	 * Lines <em>beginning</em> with one of the comment prefixes are excluded from the results; however, line comments
+	 * anywhere else &mdash; for example, within a statement &mdash; will be included in the results.
+	 *
+	 * @param lineNumberReader the {@code LineNumberReader} containing the script to be processed.
+	 * @param lineCommentPrefixes the prefixes that identify comments in the CQL script (typically "--").
+	 * @param separator the statement separator in the CQL script (typically ";").
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter.
+	 * @return a {@code String} containing the script lines
+	 * @throws IOException in case of I/O errors
+	 */
+	public static String readScript(LineNumberReader lineNumberReader, @Nullable String[] lineCommentPrefixes,
+			@Nullable String separator, @Nullable String blockCommentEndDelimiter) throws IOException {
+
+		String currentStatement = lineNumberReader.readLine();
+		StringBuilder scriptBuilder = new StringBuilder();
+		while (currentStatement != null) {
+			if ((blockCommentEndDelimiter != null && currentStatement.contains(blockCommentEndDelimiter))
+					|| (lineCommentPrefixes != null && !startsWithAny(currentStatement, lineCommentPrefixes, 0))) {
+				if (scriptBuilder.length() > 0) {
+					scriptBuilder.append('\n');
+				}
+				scriptBuilder.append(currentStatement);
+			}
+			currentStatement = lineNumberReader.readLine();
+		}
+		appendSeparatorToScriptIfNecessary(scriptBuilder, separator);
+		return scriptBuilder.toString();
+	}
+
+	private static void appendSeparatorToScriptIfNecessary(StringBuilder scriptBuilder, @Nullable String separator) {
+		if (separator == null) {
+			return;
+		}
+		String trimmed = separator.trim();
+		if (trimmed.length() == separator.length()) {
+			return;
+		}
+		// separator ends in whitespace, so we might want to see if the script is trying
+		// to end the same way
+		if (scriptBuilder.lastIndexOf(trimmed) == scriptBuilder.length() - trimmed.length()) {
+			scriptBuilder.append(separator.substring(trimmed.length()));
+		}
+	}
+
+	private static boolean startsWithAny(String script, String[] prefixes, int offset) {
+		for (String prefix : prefixes) {
+			if (script.startsWith(prefix, offset)) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Does the provided CQL script contain the specified delimiter?
+	 *
+	 * @param script the CQL script.
+	 * @param separator the string delimiting each statement - typically a ';' character.
+	 */
+	public static boolean containsCqlScriptDelimiters(String script, String separator) {
+
+		boolean inLiteral = false;
+		boolean inEscape = false;
+
+		for (int i = 0; i < script.length(); i++) {
+			char c = script.charAt(i);
+			if (inEscape) {
+				inEscape = false;
+				continue;
+			}
+			if (c == '\\') {
+				inEscape = true;
+				continue;
+			}
+			if (c == '\'') {
+				inLiteral = !inLiteral;
+			}
+			if (!inLiteral && script.startsWith(separator, i)) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
+	/**
+	 * Execute the given CQL script using default settings for statement separators, comment delimiters, and exception
+	 * handling flags.
+	 * <p>
+	 * Statement separators and comments will be removed before executing individual statements within the supplied
+	 * script.
+	 *
+	 * @param session the CQL {@link Session} to use to execute the script; already configured and ready to use.
+	 * @param resource the resource to load the CQL script from; encoded with the current platform's default encoding.
+	 * @throws ScriptException if an error occurred while executing the CQL script
+	 * @see #executeCqlScript(Session, EncodedResource, boolean, boolean, String, String, String, String)
+	 * @see #DEFAULT_STATEMENT_SEPARATOR
+	 * @see #DEFAULT_COMMENT_PREFIX
+	 * @see #DEFAULT_BLOCK_COMMENT_START_DELIMITER
+	 * @see #DEFAULT_BLOCK_COMMENT_END_DELIMITER
+	 */
+	public static void executeCqlScript(Session session, Resource resource) throws ScriptException {
+		executeCqlScript(session, new EncodedResource(resource));
+	}
+
+	/**
+	 * Execute the given CQL script using default settings for statement separators, comment delimiters, and exception
+	 * handling flags.
+	 * <p>
+	 * Statement separators and comments will be removed before executing individual statements within the supplied
+	 * script.
+	 *
+	 * @param session the CQL {@link Session} to use to execute the script; already configured and ready to use.
+	 * @param resource the resource (potentially associated with a specific encoding) to load the CQL script from.
+	 * @throws ScriptException if an error occurred while executing the CQL script
+	 * @see #executeCqlScript(Session, EncodedResource, boolean, boolean, String, String, String, String)
+	 * @see #DEFAULT_STATEMENT_SEPARATOR
+	 * @see #DEFAULT_COMMENT_PREFIX
+	 * @see #DEFAULT_BLOCK_COMMENT_START_DELIMITER
+	 * @see #DEFAULT_BLOCK_COMMENT_END_DELIMITER
+	 */
+	public static void executeCqlScript(Session session, EncodedResource resource) throws ScriptException {
+		executeCqlScript(session, resource, false, false, DEFAULT_COMMENT_PREFIX, DEFAULT_STATEMENT_SEPARATOR,
+				DEFAULT_BLOCK_COMMENT_START_DELIMITER, DEFAULT_BLOCK_COMMENT_END_DELIMITER);
+	}
+
+	/**
+	 * Execute the given CQL script.
+	 * <p>
+	 * Statement separators and comments will be removed before executing individual statements within the supplied
+	 * script.
+	 *
+	 * @param session the CQL {@link Session} to use to execute the script; already configured and ready to use.
+	 * @param resource the resource (potentially associated with a specific encoding) to load the CQL script from.
+	 * @param continueOnError whether or not to continue without throwing an exception in the event of an error.
+	 * @param ignoreFailedDrops whether or not to continue in the event of specifically an error on a {@code DROP}
+	 *          statement.
+	 * @param commentPrefix the prefix that identifies single-line comments in the CQL script (typically "--")-
+	 * @param separator the script statement separator; defaults to {@value #DEFAULT_STATEMENT_SEPARATOR} if not specified
+	 *          and falls back to {@value #FALLBACK_STATEMENT_SEPARATOR} as a last resort; may be set to
+	 *          {@value #EOF_STATEMENT_SEPARATOR} to signal that the script contains a single statement without a
+	 *          separator.
+	 * @param blockCommentStartDelimiter the <em>start</em> block comment delimiter
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter
+	 * @throws ScriptException if an error occurred while executing the CQL script
+	 * @see #DEFAULT_STATEMENT_SEPARATOR
+	 * @see #FALLBACK_STATEMENT_SEPARATOR
+	 * @see #EOF_STATEMENT_SEPARATOR
+	 */
+	public static void executeCqlScript(Session session, EncodedResource resource, boolean continueOnError,
+			boolean ignoreFailedDrops, String commentPrefix, @Nullable String separator, String blockCommentStartDelimiter,
+			String blockCommentEndDelimiter) throws ScriptException {
+
+		executeCqlScript(session, resource, continueOnError, ignoreFailedDrops, new String[] { commentPrefix }, separator,
+				blockCommentStartDelimiter, blockCommentEndDelimiter);
+	}
+
+	/**
+	 * Execute the given CQL script.
+	 * <p>
+	 * Statement separators and comments will be removed before executing individual statements within the supplied
+	 * script.
+	 *
+	 * @param session the CQL {@link Session} to use to execute the script; already configured and ready to use.
+	 * @param resource the resource (potentially associated with a specific encoding) to load the CQL script from.
+	 * @param continueOnError whether or not to continue without throwing an exception in the event of an error.
+	 * @param ignoreFailedDrops whether or not to continue in the event of specifically an error on a {@code DROP}
+	 *          statement.
+	 * @param commentPrefixes the prefixes that identify single-line comments in the CQL script (typically "--").
+	 * @param separator the script statement separator; defaults to {@value #DEFAULT_STATEMENT_SEPARATOR} if not specified
+	 *          and falls back to {@value #FALLBACK_STATEMENT_SEPARATOR} as a last resort; may be set to
+	 *          {@value #EOF_STATEMENT_SEPARATOR} to signal that the script contains a single statement without a
+	 *          separator.
+	 * @param blockCommentStartDelimiter the <em>start</em> block comment delimiter
+	 * @param blockCommentEndDelimiter the <em>end</em> block comment delimiter
+	 * @throws ScriptException if an error occurred while executing the CQL script
+	 * @see #DEFAULT_STATEMENT_SEPARATOR
+	 * @see #FALLBACK_STATEMENT_SEPARATOR
+	 * @see #EOF_STATEMENT_SEPARATOR
+	 */
+	public static void executeCqlScript(Session session, EncodedResource resource, boolean continueOnError,
+			boolean ignoreFailedDrops, String[] commentPrefixes, @Nullable String separator,
+			String blockCommentStartDelimiter, String blockCommentEndDelimiter) throws ScriptException {
+
+		try {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Executing CQL script from " + resource);
+			}
+
+			long startTime = System.currentTimeMillis();
+
+			String script;
+			try {
+				script = readScript(resource, commentPrefixes, separator, blockCommentEndDelimiter);
+			} catch (IOException ex) {
+				throw new CannotReadScriptException(resource, ex);
+			}
+
+			if (separator == null) {
+				separator = DEFAULT_STATEMENT_SEPARATOR;
+			}
+			if (!EOF_STATEMENT_SEPARATOR.equals(separator) && !containsCqlScriptDelimiters(script, separator)) {
+				separator = FALLBACK_STATEMENT_SEPARATOR;
+			}
+
+			List<String> statements = new ArrayList<>();
+			splitCqlScript(resource, script, separator, commentPrefixes, blockCommentStartDelimiter, blockCommentEndDelimiter,
+					statements);
+
+			int stmtNumber = 0;
+			for (String statement : statements) {
+				stmtNumber++;
+				try {
+					ResultSet result = session.execute(statement);
+					if (logger.isDebugEnabled()) {
+
+						ExecutionInfo executionInfo = result.getExecutionInfo();
+						if (executionInfo != null) {
+							for (String warning : executionInfo.getWarnings()) {
+								logger.debug(String.format("CQL warning ignored: [%s]", warning));
+							}
+						}
+					}
+				} catch (RuntimeException ex) {
+					boolean dropStatement = StringUtils.startsWithIgnoreCase(statement.trim(), "drop");
+					if (continueOnError || (dropStatement && ignoreFailedDrops)) {
+						if (logger.isDebugEnabled()) {
+							logger.debug(ScriptStatementFailedException.buildErrorMessage(statement, stmtNumber, resource), ex);
+						}
+					} else {
+						throw new ScriptStatementFailedException(statement, stmtNumber, resource, ex);
+					}
+				}
+			}
+
+			long elapsedTime = System.currentTimeMillis() - startTime;
+			if (logger.isDebugEnabled()) {
+				logger.debug("Executed CQL script from " + resource + " in " + elapsedTime + " ms.");
+			}
+		} catch (Exception ex) {
+			if (ex instanceof ScriptException) {
+				throw (ScriptException) ex;
+			}
+			throw new UncategorizedScriptException("Failed to execute database script from resource [" + resource + "]", ex);
+		}
+	}
+
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializer.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializer.java
new file mode 100644
index 000000000..7784563c6
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializer.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.data.cassandra.SessionFactory;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Used to {@linkplain #setKeyspacePopulator set up} a keyspace during initialization and {@link #setKeyspaceCleaner
+ * clean up} a keyspace during destruction.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ * @see KeyspacePopulator
+ */
+public class SessionFactoryInitializer implements InitializingBean, DisposableBean {
+
+	@Nullable private SessionFactory sessionFactory;
+
+	@Nullable private KeyspacePopulator keyspacePopulator;
+
+	@Nullable private KeyspacePopulator keyspaceCleaner;
+
+	private boolean enabled = true;
+
+	/**
+	 * The {@link SessionFactory} for the keyspace to populate when this component is initialized and to clean up when
+	 * this component is shut down.
+	 * <p>
+	 * This property is mandatory with no default provided.
+	 *
+	 * @param sessionFactory the SessionFactory.
+	 */
+	public void setSessionFactory(SessionFactory sessionFactory) {
+		this.sessionFactory = sessionFactory;
+	}
+
+	/**
+	 * Set the {@link KeyspacePopulator} to execute during the bean initialization phase.
+	 *
+	 * @param keyspacePopulator the {@link KeyspacePopulator} to use during initialization.
+	 * @see #setKeyspaceCleaner
+	 */
+	public void setKeyspacePopulator(KeyspacePopulator keyspacePopulator) {
+		this.keyspacePopulator = keyspacePopulator;
+	}
+
+	/**
+	 * Set the {@link KeyspacePopulator} to execute during the bean destruction phase, cleaning up the keyspace and
+	 * leaving it in a known state for others.
+	 *
+	 * @param keyspaceCleaner the {@link KeyspacePopulator} to use during destruction.
+	 * @see #setKeyspacePopulator
+	 */
+	public void setKeyspaceCleaner(KeyspacePopulator keyspaceCleaner) {
+		this.keyspaceCleaner = keyspaceCleaner;
+	}
+
+	/**
+	 * Flag to explicitly enable or disable the {@linkplain #setKeyspacePopulator keyspace populator} and
+	 * {@linkplain #setKeyspaceCleaner keyspace cleaner}.
+	 *
+	 * @param enabled {@literal true} if the keyspace populator and keyspace cleaner should be called on startup and
+	 *          shutdown, respectively.
+	 */
+	public void setEnabled(boolean enabled) {
+		this.enabled = enabled;
+	}
+
+	/**
+	 * Use the {@linkplain #setKeyspacePopulator keyspace populator} to set up the keyspace.
+	 */
+	@Override
+	public void afterPropertiesSet() {
+		execute(this.keyspacePopulator);
+	}
+
+	/**
+	 * Use the {@linkplain #setKeyspaceCleaner keyspace cleaner} to clean up the keyspace.
+	 */
+	@Override
+	public void destroy() {
+		execute(this.keyspaceCleaner);
+	}
+
+	private void execute(@Nullable KeyspacePopulator populator) {
+
+		Assert.state(this.sessionFactory != null, "SessionFactory must be set");
+
+		if (this.enabled && populator != null) {
+			populator.populate(this.sessionFactory.getSession());
+		}
+	}
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/UncategorizedScriptException.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/UncategorizedScriptException.java
new file mode 100644
index 000000000..5571a566a
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/UncategorizedScriptException.java
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+/**
+ * Thrown when we cannot determine anything more specific than "something went wrong while processing a CQL script": for
+ * example, a {@link Exception} from Cassandra that we cannot pinpoint more precisely.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class UncategorizedScriptException extends ScriptException {
+
+	/**
+	 * Construct a new {@link UncategorizedScriptException}.
+	 *
+	 * @param message detailed message.
+	 */
+	public UncategorizedScriptException(String message) {
+		super(message);
+	}
+
+	/**
+	 * Construct a new {@link UncategorizedScriptException}.
+	 *
+	 * @param message detailed message.
+	 * @param cause the root cause.
+	 */
+	public UncategorizedScriptException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+}
diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/package-info.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/package-info.java
new file mode 100644
index 000000000..1e2981585
--- /dev/null
+++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/init/package-info.java
@@ -0,0 +1,7 @@
+/**
+ * Provides extensible support for initializing databases through scripts.
+ */
+@NonNullApi
+package org.springframework.data.cassandra.core.cql.session.init;
+
+import org.springframework.lang.NonNullApi;
diff --git a/spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-3.0.xsd b/spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-3.0.xsd
new file mode 100644
index 000000000..e7c113ac4
--- /dev/null
+++ b/spring-data-cassandra/src/main/resources/org/springframework/data/cassandra/config/spring-cassandra-3.0.xsd
@@ -0,0 +1,684 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/data/cassandra"
+			xmlns:repository="http://www.springframework.org/schema/data/repository"
+			xmlns:tool="http://www.springframework.org/schema/tool"
+			xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+			targetNamespace="http://www.springframework.org/schema/data/cassandra"
+			elementFormDefault="qualified" attributeFormDefault="unqualified">
+
+	<xsd:import namespace="http://www.springframework.org/schema/beans"
+				schemaLocation="https://www.springframework.org/schema/beans/spring-beans.xsd"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"
+				schemaLocation="https://www.springframework.org/schema/tool/spring-tool.xsd"/>
+	<xsd:import namespace="http://www.springframework.org/schema/context"
+				schemaLocation="https://www.springframework.org/schema/context/spring-context.xsd"/>
+	<xsd:import namespace="http://www.springframework.org/schema/data/repository"
+				schemaLocation="https://www.springframework.org/schema/data/repository/spring-repository.xsd"/>
+
+	<xsd:annotation>
+		<xsd:documentation><![CDATA[
+Defines configuration elements in the XML namespace for Spring Data for Apache Cassandra.
+		]]></xsd:documentation>
+	</xsd:annotation>
+
+	<xsd:element name="auditing">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation>
+					<tool:exports
+							type="org.springframework.data.mongodb.core.mapping.event.AuditingEntityCallback"/>
+					<tool:exports
+							type="org.springframework.data.auditing.IsNewAwareAuditingHandler"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:attributeGroup ref="repository:auditing-attributes"/>
+			<xsd:attribute name="mapping-context-ref" type="mappingContextRef"/>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="cql-template" type="cqlTemplateType">
+		<xsd:annotation>
+			<xsd:documentation
+					source="org.springframework.data.cql.config.CassandraCqlTemplateFactoryBean">
+				<![CDATA[
+Defines a CassandraTemplate.
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:annotation>
+					<tool:exports
+							type="org.springframework.data.cassandra.core.cql.CqlTemplate"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+	</xsd:element>
+
+	<xsd:element name="initialize-keyspace" type="initializeKeyspaceType">
+		<xsd:annotation>
+			<xsd:documentation
+					source="org.springframework.data.cassandra.core.cql.session.init.SessionFactoryInitializer">
+				<![CDATA[
+	Initializes a keyspace with CQL scripts provided in nested <script/> elements.
+			]]></xsd:documentation>
+		</xsd:annotation>
+	</xsd:element>
+
+	<xsd:element name="session" type="sessionType">
+		<xsd:annotation>
+			<xsd:documentation
+					source="org.springframework.data.cassandra.config.CqlSessionFactoryBean">
+				<![CDATA[
+Defines a Cassandra session.
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:annotation>
+					<tool:exports type="com.datastax.driver.core.Session"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+	</xsd:element>
+
+	<xsd:element name="template" type="templateType">
+		<xsd:annotation>
+			<xsd:documentation
+					source="org.springframework.data.cassandra.config.CassandraTemplateFactoryBean">
+				<![CDATA[
+Defines a CassandraTemplate.
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:annotation>
+					<tool:exports
+							type="org.springframework.data.cassandra.CassandraTemplate"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+	</xsd:element>
+
+	<xsd:simpleType name="sessionRef" final="union">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation kind="ref">
+					<tool:assignable-to type="com.datastax.driver.core.Session"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:union memberTypes="xsd:string"/>
+	</xsd:simpleType>
+
+	<xsd:complexType name="datacenterType">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+Provides the ability to specify replication factors by data center.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:attribute name="name" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the data center.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="replication-factor" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The replication factor for the data center.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="initializeKeyspaceType">
+		<xsd:sequence>
+			<xsd:element name="script" type="scriptType" minOccurs="1"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+			An CQL script to execute to populate, initialize, or clean up a keyspace.
+						]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+		</xsd:sequence>
+		<xsd:attribute name="session-factory" type="xsd:string" default="sessionFactory">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+			A reference to a data source that should be initialized. Defaults to "sessionFactory".
+					]]></xsd:documentation>
+				<xsd:appinfo>
+					<tool:annotation kind="ref"/>
+					<tool:expected-type
+							type="org.springframework.data.cassandra.SessionFactory"/>
+				</xsd:appinfo>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="enabled" type="xsd:string" default="true">
+			<xsd:annotation>
+				<xsd:documentation>
+					Is this bean "enabled", meaning the scripts will be executed?
+					Defaults to true but can be used to switch on and off script execution
+					depending on the environment.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="ignore-failures" default="NONE">
+			<xsd:annotation>
+				<xsd:documentation>
+					Should failed CQL statements be ignored during execution?
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:string">
+					<xsd:enumeration value="NONE">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+						Do not ignore failures (the default)
+								]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="DROPS">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+						Ignore failed DROP statements
+								]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+					<xsd:enumeration value="ALL">
+						<xsd:annotation>
+							<xsd:documentation><![CDATA[
+						Ignore all failures
+								]]></xsd:documentation>
+						</xsd:annotation>
+					</xsd:enumeration>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+		<xsd:attribute name="separator" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The default statement separator to use (the default is to use ';' if it is present
+	in the script, or '\n' otherwise).
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="keyspaceType">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+Provides the ability to define keyspaces.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:sequence>
+			<xsd:element name="replication" type="replicationType" minOccurs="0"
+						 maxOccurs="1">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Provides the ability to configure the keyspace's replication settings.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+		</xsd:sequence>
+		<xsd:attribute name="action" use="required" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The keyspace action to take at startup and possibly shutdown.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="durable-writes" type="xsd:string" use="optional"
+					   default="false">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Whether or not the keyspace supports durable writes.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="name" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of this keyspace.  Required.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="replicationType">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+Provides the ability to configure the keyspace's replication settings.
+					]]></xsd:documentation>
+		</xsd:annotation>
+		<xsd:sequence>
+			<xsd:element name="data-center" type="datacenterType" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Provides the ability to specify replication factors by data center.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+		</xsd:sequence>
+		<xsd:attribute name="class" type="xsd:string"
+					   default="SimpleStrategy">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the replication class; default is "SIMPLE_STRATEGY".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="replication-factor" type="xsd:string"
+					   default="1">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The replication factor; default is 1.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="sessionType">
+		<xsd:sequence>
+			<xsd:element name="keyspace" type="keyspaceType" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Provides the ability to define a keyspace.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+			<xsd:element name="keyspace-startup-cql" type="xsd:string" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Arbitrary CQL script to be executed against the system keyspace during bean initialization.  Multiple elements will be executed in document order.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+			<xsd:element name="keyspace-shutdown-cql" type="xsd:string" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Arbitrary CQL script to be executed against the system keyspace during bean destruction.  Multiple elements will be executed in document order.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+			<xsd:element name="startup-cql" type="xsd:string" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Arbitrary CQL script to be executed against the session's keyspace during bean initialization.  Multiple elements will be executed in document order.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+			<xsd:element name="shutdown-cql" type="xsd:string" minOccurs="0"
+						 maxOccurs="unbounded">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+Arbitrary CQL script to be executed against the session's keyspace during bean destruction.  Multiple elements will be executed in document order.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:element>
+		</xsd:sequence>
+		<xsd:attribute name="id" type="xsd:ID">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the session definition; default is "cassandraSession".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="cassandra-converter-ref" type="cassandraConverterRef"
+		>
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a CassandraConverter; default is "cassandraConverter".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="contact-points" type="xsd:string" default="localhost"
+		>
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The comma separated list of Cassandra servers.  Default is "localhost".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="keyspace-name" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of a Cassandra Keyspace.  No default; for the system keyspace, use the empty string.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="password" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+When Authentication is enabled, the password to use when connecting to the Cluster.
+				]]>
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="port" type="xsd:string" default="9042">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The native CQL port to connect to.  Default is 9042.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="schema-action" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The schema action to perform; default is NONE.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="username" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+When Authentication is enabled, the username to use when connecting to the Cluster.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="cqlTemplateType">
+		<xsd:attribute name="id" type="xsd:ID">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the template; default is "cqlTemplate".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="session-ref" type="sessionRef">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a Cassandra session; default is "cassandraSession".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="templateType">
+		<xsd:attribute name="id" type="xsd:ID">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the template; default is "cassandraTemplate".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="cassandra-converter-ref" type="cassandraConverterRef"
+		>
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a CassandraConverter; default is "cassandraConverter".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="cql-template-ref" type="cqlTemplateRef">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a CqlTemplate; default is none. Providing a CqlTemplate reference overrides session references.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="session-ref" type="sessionRef">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a Cassandra session; default is "cassandraSession". Session reference is omitted if a CqlTemplate reference is provided.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<!-- Spring Data Repository and Mapping (Persistence) Schema Elements -->
+
+	<xsd:element name="converter">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+Defines a CassandraConverter for getting rich mapping functionality.
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:exports
+						type="org.springframework.data.cassandra.core.convert.CassandraConverter"/>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:attribute name="id" type="xsd:ID">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+	The name of the converter ; default is "cassandraConverter".
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="mapping-ref" type="mappingContextRef">
+				<xsd:annotation>
+					<xsd:documentation
+							source="org.springframework.data.cassandra.core.mapping.CassandraMappingContext">
+						<![CDATA[
+The reference to a CassandraMappingContext. Will default to 'cassandraMapping'.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="mapping">
+		<xsd:annotation>
+			<xsd:documentation><![CDATA[
+Defines a CassandraMappingContext for holding rich entity mapping information.
+			]]></xsd:documentation>
+			<xsd:appinfo>
+				<tool:exports
+						type="org.springframework.data.cassandra.core.mapping.CassandraMappingContext"/>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:sequence>
+				<xsd:element name="entity" type="entityType" minOccurs="0"
+							 maxOccurs="unbounded"/>
+				<xsd:element name="user-type-resolver" type="userTypeResolverType"
+							 minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+			<xsd:attribute name="id" type="xsd:ID">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+	The name of the mapping context; default is "cassandraMapping".
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="entity-base-packages" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+The comma-delimited base packages in which to scan for entities and their mapping information.
+					]]></xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="user-type-resolver-ref" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+The reference to a UserTypeResolver. UserTypeResolver is required when working with User-defined types.
+					]]></xsd:documentation>
+					<xsd:appinfo>
+						<tool:exports
+								type="org.springframework.data.cassandra.core.mapping.UserTypeResolver"/>
+					</xsd:appinfo>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="repositories">
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="repository:repositories">
+					<xsd:attributeGroup ref="cassandra-repository-attributes"/>
+					<xsd:attributeGroup ref="repository:repository-attributes"/>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:attributeGroup name="cassandra-repository-attributes">
+		<xsd:attribute name="cassandra-template-ref" type="cassandraTemplateRef">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a CassandraTemplate. Will default to 'cqlTemplate'.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
+	<xsd:simpleType name="cassandraConverterRef" final="union">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation kind="ref">
+					<tool:assignable-to
+							type="org.springframework.data.cassandra.core.convert.CassandraConverter"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:union memberTypes="xsd:string"/>
+	</xsd:simpleType>
+
+	<xsd:simpleType name="cassandraTemplateRef" final="union">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation kind="ref">
+					<tool:assignable-to
+							type="org.springframework.data.cassandra.core.CassandraTemplate"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:union memberTypes="xsd:string"/>
+	</xsd:simpleType>
+
+	<xsd:simpleType name="cqlTemplateRef" final="union">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation kind="ref">
+					<tool:assignable-to
+							type="org.springframework.data.cassandra.core.cql.CqlTemplate"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:union memberTypes="xsd:string"/>
+	</xsd:simpleType>
+
+	<xsd:simpleType name="mappingContextRef">
+		<xsd:annotation>
+			<xsd:appinfo>
+				<tool:annotation kind="ref">
+					<tool:assignable-to
+							type="org.springframework.data.cassandra.core.mapping.CassandraMappingContext"/>
+				</tool:annotation>
+			</xsd:appinfo>
+		</xsd:annotation>
+		<xsd:union memberTypes="xsd:string"/>
+	</xsd:simpleType>
+
+	<xsd:complexType name="entityType">
+		<xsd:sequence>
+			<xsd:element name="table" type="tableType" minOccurs="0" maxOccurs="1"/>
+			<xsd:element name="property" type="propertyType" minOccurs="0"
+						 maxOccurs="unbounded"/>
+		</xsd:sequence>
+		<xsd:attribute name="class" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Entity class name.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="propertyType">
+		<xsd:attribute name="column-name" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The column-name that the property should be mapped to.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="force-quote" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Whether the column name should be force-quoted.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="name" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The name of the property.  Required.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="scriptType">
+		<xsd:attribute name="location" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The resource location of an CQL script to execute. Can be a single script location
+	or a pattern (e.g. classpath:/com/foo/cql/*-data.cql).
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="encoding" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The encoding for CQL scripts, if different from the platform encoding.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="separator" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	The statement separator in the script (the default is to use ';' if it is present
+	in the script, or '\n' otherwise).
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="execution">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+	Indicate the execution phase of this script. Use INIT to execute on startup (as a
+	bean initialization) or DESTROY to execute on shutdown (as a bean destruction callback).
+				]]></xsd:documentation>
+			</xsd:annotation>
+			<xsd:simpleType>
+				<xsd:restriction base="xsd:string">
+					<xsd:enumeration value="INIT"/>
+					<xsd:enumeration value="DESTROY"/>
+				</xsd:restriction>
+			</xsd:simpleType>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="tableType">
+		<xsd:attribute name="force-quote" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Whether to force-quote the table name.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="name" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+Table name override.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+	<xsd:complexType name="userTypeResolverType">
+		<xsd:attribute name="session-ref" type="sessionRef" use="required">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+The reference to a Cassandra Session; default is "cassandraSession".
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:complexType>
+
+</xsd:schema>
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulatorUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulatorUnitTests.java
new file mode 100644
index 000000000..02ec3b036
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/CompositeKeyspacePopulatorUnitTests.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import static org.mockito.Mockito.*;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import org.junit.Test;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Unit tests for {@link CompositeKeyspacePopulator}.
+ *
+ * @author Mark Paluch
+ */
+public class CompositeKeyspacePopulatorUnitTests {
+
+	private final Session mockedConnection = mock(Session.class);
+
+	private final KeyspacePopulator mockedKeyspacePopulator1 = mock(KeyspacePopulator.class);
+
+	private final KeyspacePopulator mockedKeyspacePopulator2 = mock(KeyspacePopulator.class);
+
+	@Test // DATACASS-704
+	public void addPopulators() {
+
+		CompositeKeyspacePopulator populator = new CompositeKeyspacePopulator();
+		populator.addPopulators(mockedKeyspacePopulator1, mockedKeyspacePopulator2);
+
+		populator.populate(mockedConnection);
+
+		verify(mockedKeyspacePopulator1, times(1)).populate(mockedConnection);
+		verify(mockedKeyspacePopulator2, times(1)).populate(mockedConnection);
+	}
+
+	@Test // DATACASS-704
+	public void setPopulatorsWithMultiple() {
+
+		CompositeKeyspacePopulator populator = new CompositeKeyspacePopulator();
+		populator.setPopulators(mockedKeyspacePopulator1, mockedKeyspacePopulator2); // multiple
+
+		populator.populate(mockedConnection);
+
+		verify(mockedKeyspacePopulator1, times(1)).populate(mockedConnection);
+		verify(mockedKeyspacePopulator2, times(1)).populate(mockedConnection);
+	}
+
+	@Test // DATACASS-704
+	public void setPopulatorsForOverride() {
+
+		CompositeKeyspacePopulator populator = new CompositeKeyspacePopulator();
+		populator.setPopulators(mockedKeyspacePopulator1);
+		populator.setPopulators(mockedKeyspacePopulator2); // override
+
+		populator.populate(mockedConnection);
+
+		verify(mockedKeyspacePopulator1, times(0)).populate(mockedConnection);
+		verify(mockedKeyspacePopulator2, times(1)).populate(mockedConnection);
+	}
+
+	@Test // DATACASS-704
+	public void constructWithVarargs() {
+
+		CompositeKeyspacePopulator populator = new CompositeKeyspacePopulator(mockedKeyspacePopulator1,
+				mockedKeyspacePopulator2);
+
+		populator.populate(mockedConnection);
+
+		verify(mockedKeyspacePopulator1, times(1)).populate(mockedConnection);
+		verify(mockedKeyspacePopulator2, times(1)).populate(mockedConnection);
+	}
+
+	@Test // DATACASS-704
+	public void constructWithCollection() {
+
+		Set<KeyspacePopulator> populators = new LinkedHashSet<>();
+		populators.add(mockedKeyspacePopulator1);
+		populators.add(mockedKeyspacePopulator2);
+		CompositeKeyspacePopulator populator = new CompositeKeyspacePopulator(populators);
+
+		populator.populate(mockedConnection);
+
+		verify(mockedKeyspacePopulator1, times(1)).populate(mockedConnection);
+		verify(mockedKeyspacePopulator2, times(1)).populate(mockedConnection);
+	}
+}
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulatorUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulatorUnitTests.java
new file mode 100644
index 000000000..321c283fa
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ResourceKeyspacePopulatorUnitTests.java
@@ -0,0 +1,158 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+
+/**
+ * Unit tests for {@link ResourceKeyspacePopulator}.
+ *
+ * @author Mark Paluch
+ */
+public class ResourceKeyspacePopulatorUnitTests {
+
+	private static final Resource script1 = mock(Resource.class);
+	private static final Resource script2 = mock(Resource.class);
+	private static final Resource script3 = mock(Resource.class);
+
+	@Test // DATACASS-704
+	public void constructWithNullResource() {
+
+		assertThatIllegalArgumentException().isThrownBy(() -> new ResourceKeyspacePopulator((Resource) null));
+	}
+
+	@Test // DATACASS-704
+	public void constructWithNullResourceArray() {
+
+		assertThatIllegalArgumentException().isThrownBy(() -> new ResourceKeyspacePopulator((Resource[]) null));
+	}
+
+	@Test // DATACASS-704
+	public void constructWithResource() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator(script1);
+
+		assertThat(keyspacePopulator.scripts).hasSize(1);
+	}
+
+	@Test // DATACASS-704
+	public void constructWithMultipleResources() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator(script1, script2);
+
+		assertThat(keyspacePopulator.scripts).hasSize(2);
+	}
+
+	@Test // DATACASS-704
+	public void constructWithMultipleResourcesAndThenAddScript() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator(script1, script2);
+
+		assertThat(keyspacePopulator.scripts).hasSize(2);
+
+		keyspacePopulator.addScript(script3);
+
+		assertThat(keyspacePopulator.scripts).hasSize(3);
+	}
+
+	@Test // DATACASS-704
+	public void addScriptsWithNullResource() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+
+		assertThatIllegalArgumentException().isThrownBy(() -> keyspacePopulator.addScripts((Resource) null));
+	}
+
+	@Test // DATACASS-704
+	public void addScriptsWithNullResourceArray() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+
+		assertThatIllegalArgumentException().isThrownBy(() -> keyspacePopulator.addScripts((Resource[]) null));
+	}
+
+	@Test // DATACASS-704
+	public void setScriptsWithNullResource() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+
+		assertThatIllegalArgumentException().isThrownBy(() -> keyspacePopulator.setScripts((Resource) null));
+	}
+
+	@Test // DATACASS-704
+	public void setScriptsWithNullResourceArray() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+
+		assertThatIllegalArgumentException().isThrownBy(() -> keyspacePopulator.setScripts((Resource[]) null));
+	}
+
+	@Test // DATACASS-704
+	public void shouldFailOnError() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+		keyspacePopulator.setScripts(new ByteArrayResource("drop table;create table;".getBytes()));
+
+		Session sessionMock = mock(Session.class);
+		when(sessionMock.execute("drop table")).thenThrow(new IllegalStateException("Boom!"));
+
+		assertThatExceptionOfType(ScriptStatementFailedException.class)
+				.isThrownBy(() -> keyspacePopulator.populate(sessionMock));
+
+		verify(sessionMock).execute("drop table");
+		verifyNoMoreInteractions(sessionMock);
+	}
+
+	@Test // DATACASS-704
+	public void shouldContinueOnError() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+		keyspacePopulator.setIgnoreFailedDrops(true);
+		keyspacePopulator.setScripts(new ByteArrayResource("drop table;create table;".getBytes()));
+
+		Session sessionMock = mock(Session.class);
+		when(sessionMock.execute("drop table")).thenThrow(new IllegalStateException("Boom!"));
+
+		when(sessionMock.execute("create table")).thenReturn(mock(ResultSet.class));
+
+		keyspacePopulator.populate(sessionMock);
+
+		verify(sessionMock).execute("drop table");
+		verify(sessionMock).execute("create table");
+	}
+
+	@Test
+	public void setScriptsAndThenAddScript() {
+
+		ResourceKeyspacePopulator keyspacePopulator = new ResourceKeyspacePopulator();
+		assertThat(keyspacePopulator.scripts).isEmpty();
+
+		keyspacePopulator.setScripts(script1, script2);
+		assertThat(keyspacePopulator.scripts).hasSize(2);
+
+		keyspacePopulator.addScript(script3);
+		assertThat(keyspacePopulator.scripts).hasSize(3);
+	}
+}
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtilsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtilsUnitTests.java
new file mode 100644
index 000000000..5c12ffff0
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/ScriptUtilsUnitTests.java
@@ -0,0 +1,213 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.support.EncodedResource;
+
+/**
+ * Unit tests for {@link ScriptUtils}.
+ *
+ * @author Mark Paluch
+ */
+public class ScriptUtilsUnitTests {
+
+	@Test // DATACASS-704
+	public void splitCqlScriptDelimitedWithSemicolon() {
+
+		String rawStatement1 = "insert into customer (id, name)\nvalues (1, 'Walter ; White'), (2, 'Hank \n Schrader')";
+		String cleanedStatement1 = "insert into customer (id, name) values (1, 'Walter ; White'), (2, 'Hank \n Schrader')";
+		String rawStatement2 = "insert into orders(id, order_date, customer_id)\nvalues (1, '2008-01-02', 2)";
+		String cleanedStatement2 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+		String rawStatement3 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+		String cleanedStatement3 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+
+		char delim = ';';
+
+		String script = rawStatement1 + delim + rawStatement2 + delim + rawStatement3 + delim;
+		List<String> statements = new ArrayList<>();
+		ScriptUtils.splitCqlScript(script, delim, statements);
+
+		assertThat(statements).containsExactly(cleanedStatement1, cleanedStatement2, cleanedStatement3);
+	}
+
+	@Test // DATACASS-704
+	public void splitCqlScriptDelimitedWithNewLine() {
+
+		String statement1 = "insert into customer (id, name) values (1, 'Walter ; White'), (2, 'Hank \n Schrader')";
+		String statement2 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+		String statement3 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+
+		char delim = '\n';
+
+		String script = statement1 + delim + statement2 + delim + statement3 + delim;
+		List<String> statements = new ArrayList<>();
+		ScriptUtils.splitCqlScript(script, delim, statements);
+
+		assertThat(statements).containsExactly(statement1, statement2, statement3);
+	}
+
+	@Test // DATACASS-704
+	public void splitCqlScriptDelimitedWithNewLineButDefaultDelimiterSpecified() {
+
+		String statement1 = "do something";
+		String statement2 = "do something else";
+
+		char delim = '\n';
+
+		String script = statement1 + delim + statement2 + delim;
+		List<String> statements = new ArrayList<>();
+		ScriptUtils.splitCqlScript(script, ScriptUtils.DEFAULT_STATEMENT_SEPARATOR, statements);
+
+		assertThat(statements).as("stripped but not split statements").containsExactly(script.replace('\n', ' '));
+	}
+
+	@Test // DATACASS-704
+	public void splitScriptWithSingleQuotesNestedInsideDoubleQuotes() {
+
+		String statement1 = "select '1' as \"Hank's owner's\" from dual";
+		String statement2 = "select '2' as \"Hank's\" from dual";
+		char delim = ';';
+		String script = statement1 + delim + statement2 + delim;
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(script, ';', statements);
+
+		assertThat(statements).containsExactly(statement1, statement2);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptWithMultipleNewlinesAsSeparator() throws IOException {
+
+		String script = readScript("db-test-data-multi-newline.cql");
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(script, "\n\n", statements);
+
+		String statement1 = "insert into T_TEST (NAME) values ('Hank')";
+		String statement2 = "insert into T_TEST (NAME) values ('Walter')";
+
+		assertThat(statements).containsExactly(statement1, statement2);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingComments() throws Exception {
+
+		String script = readScript("test-data-with-comments.cql");
+
+		splitScriptContainingComments(script, ScriptUtils.DEFAULT_COMMENT_PREFIXES);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingCommentsWithWindowsLineEnding() throws Exception {
+		String script = readScript("test-data-with-comments.cql").replaceAll("\n", "\r\n");
+		splitScriptContainingComments(script, ScriptUtils.DEFAULT_COMMENT_PREFIXES);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingCommentsWithMultiplePrefixes() throws Exception {
+		String script = readScript("test-data-with-multi-prefix-comments.cql");
+		splitScriptContainingComments(script, "--", "#", "^");
+	}
+
+	private void splitScriptContainingComments(String script, String... commentPrefixes) {
+
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(null, script, ";", commentPrefixes, ScriptUtils.DEFAULT_BLOCK_COMMENT_START_DELIMITER,
+				ScriptUtils.DEFAULT_BLOCK_COMMENT_END_DELIMITER, statements);
+
+		String statement1 = "insert into customer (id, name) values (1, 'Walter; White'), (2, 'Hank Schrader')";
+		String statement2 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+		String statement3 = "insert into orders(id, order_date, customer_id) values (1, '2008-01-02', 2)";
+		// Statement 4 addresses the error described in SPR-9982.
+		String statement4 = "INSERT INTO persons( person_id , name) VALUES( 1 , 'Name' )";
+
+		assertThat(statements).containsExactly(statement1, statement2, statement3, statement4);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingCommentsWithLeadingTabs() throws Exception {
+
+		String script = readScript("test-data-with-comments-and-leading-tabs.cql");
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(script, ';', statements);
+
+		String statement1 = "insert into customer (id, name) values (1, 'Hank Schrader')";
+		String statement2 = "insert into orders(id, order_date, customer_id) values (1, '2013-06-08', 1)";
+		String statement3 = "insert into orders(id, order_date, customer_id) values (2, '2013-06-08', 1)";
+
+		assertThat(statements).containsExactly(statement1, statement2, statement3);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingMultiLineComments() throws Exception {
+
+		String script = readScript("test-data-with-multi-line-comments.cql");
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(script, ';', statements);
+
+		String statement1 = "INSERT INTO users(first_name, last_name) VALUES('Walter', 'White')";
+		String statement2 = "INSERT INTO users(first_name, last_name) VALUES( 'Hank' , 'Schrader' )";
+
+		assertThat(statements).containsExactly(statement1, statement2);
+	}
+
+	@Test // DATACASS-704
+	public void readAndSplitScriptContainingMultiLineNestedComments() throws Exception {
+
+		String script = readScript("test-data-with-multi-line-nested-comments.cql");
+		List<String> statements = new ArrayList<>();
+
+		ScriptUtils.splitCqlScript(script, ';', statements);
+
+		String statement1 = "INSERT INTO users(first_name, last_name) VALUES('Walter', 'White')";
+		String statement2 = "INSERT INTO users(first_name, last_name) VALUES( 'Hank' , 'Schrader' )";
+
+		assertThat(statements).containsExactly(statement1, statement2);
+	}
+
+	@Test // DATACASS-704
+	public void containsDelimiters() {
+
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1\n select ';'", ";")).isFalse();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1; select 2", ";")).isTrue();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1; select '\\n\n';", "\n")).isFalse();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1\n select 2", "\n")).isTrue();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1\n select 2", "\n\n")).isFalse();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters("select 1\n\n select 2", "\n\n")).isTrue();
+		assertThat(
+				ScriptUtils.containsCqlScriptDelimiters("insert into users(first_name, last_name)\nvalues('a\\\\', 'b;')", ";"))
+						.isFalse();
+		assertThat(ScriptUtils.containsCqlScriptDelimiters(
+				"insert into users(first_name, last_name)\nvalues('Charles', 'd\\'Artagnan'); select 1;", ";")).isTrue();
+	}
+
+	private String readScript(String path) throws IOException {
+		EncodedResource resource = new EncodedResource(new ClassPathResource(path, getClass()));
+		return ScriptUtils.readScript(resource);
+	}
+}
diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializerTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializerTests.java
new file mode 100644
index 000000000..28172b227
--- /dev/null
+++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/session/init/SessionFactoryInitializerTests.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.cassandra.core.cql.session.init;
+
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.data.cassandra.SessionFactory;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+
+/**
+ * Unit tests for {@link org.springframework.data.cassandra.core.cql.session.init.SessionFactoryInitializer}.
+ *
+ * @author Mark Paluch
+ */
+public class SessionFactoryInitializerTests {
+
+	@Test // DATACASS-704
+	public void shouldInitializeKeyspace() {
+
+		Session session = initialize("initialize-keyspace.xml");
+
+		verify(session).execute("create table if not exists mytable1 (id uuid primary key, column1 text)");
+		verify(session).execute("create table if not exists mytable2 (id uuid primary key, column1 text)");
+		verifyNoMoreInteractions(session);
+	}
+
+	@Test // DATACASS-704
+	public void shouldInitializeAndCleanupKeyspace() {
+
+		Session session = initialize("initialize-and-cleanup-keyspace.xml");
+
+		verify(session).execute("create table if not exists mytable1 (id uuid primary key, column1 text)");
+		verify(session).execute("create table if not exists mytable2 (id uuid primary key, column1 text)");
+		verify(session).execute("drop table mytable1");
+		verify(session).execute("drop table mytable2");
+		verifyNoMoreInteractions(session);
+	}
+
+	private ClassPathXmlApplicationContext context(String file) {
+		return new ClassPathXmlApplicationContext(file, getClass());
+	}
+
+	private Session initialize(String file) {
+		ConfigurableApplicationContext context = context(file);
+		try {
+			return context.getBean(SessionFactory.class).getSession();
+		} finally {
+			context.close();
+		}
+	}
+
+	@SuppressWarnings("unused")
+	private static class MockSessionFactoryFactoryBean extends AbstractFactoryBean<SessionFactory> {
+
+		@Override
+		public Class<?> getObjectType() {
+			return SessionFactory.class;
+		}
+
+		@Override
+		protected SessionFactory createInstance() {
+
+			Session sessionMock = mock(Session.class);
+			SessionFactory sessionFactoryMock = mock(SessionFactory.class);
+
+			when(sessionFactoryMock.getSession()).thenReturn(sessionMock);
+			when(sessionMock.execute(anyString())).thenReturn(mock(ResultSet.class));
+
+			return sessionFactoryMock;
+		}
+	}
+}