S-Store: A Streaming OLTP System for \\ Big Velocity Applications
\alignauthor John Meehan$^{1
\alignauthor Nesime Tatbul$^{1
2
conclusion.tex
\section{Summary \& Future Directions} \label{sec:summary}

This paper has defined a simple model of transactions for stream processing. We have presented the design and implementation of a novel system called S-Store that seemlessly combines OLTP transaction processing with our streaming transaction model. We have also shown how this symbiosis can be implemented in the context of a main-memory, OLTP database system called H-Store in a straight-forward way. S-Store is shown to outperform H-Store as well as Spark Streaming and Storm on a number of streaming workloads including a leaderboard maintenance application.

Future work includes extending S-Store to operate on multiple nodes. We plan to address a number of research issues including data and workload partitioning, distributed recovery, and distributed transaction scheduling. We also plan to investigate handling of dynamic and hybrid (OLTP+stream\-ing) workloads.

definitions.tex
\section{Definitions and Assumptions} \label{sec:definitions}

\begin{definition}
{\bf Database} is a collection of relations, streams, and windows.
\end{definition}

\begin{definition}
{\bf Relation} is an unordered, bounded collection of tuples. 
\end{definition}

\begin{definition}
{\bf Stream} is an ordered, unbounded collection of tuples. 
\end{definition}

\begin{definition}
{\bf Atomic Batch} is a finite, contiguous subsequence of a stream that must be processed atomically as an indivisible unit. Atomic batches do not overlap.
\end{definition}

\begin{definition}
{\bf Window} is a finite, contiguous subsequence of a stream.
\end{definition}

\begin{definition}
{\bf Sliding Window} is a window which has a fixed size and a fixed slide; slide $\leq$ size and specifies the distance between two consecutive windows.
\end{definition}

\begin{definition}
{\bf Time-based Sliding Window} is a sliding window for which size and slide are defined in terms of tuple timestamps.
\end{definition}

\begin{definition}
{\bf Tuple-based Sliding Window} is a sliding window for which size and slide are defined in terms of the number of tuples.
\end{definition}

\begin{definition}
{\bf Transaction} represents a finite unit of work (i.e., a finite sequence of read and write operations) performed over a given database. Each transaction has a definition and possibly many executions (i.e., instances).
\end{definition}

\begin{definition}
{\bf OLTP Transaction} is a transaction that involves operations only over relations.
\end{definition}

\begin{definition}
{\bf Streaming Transaction} is a transaction that involves operations over one or more streams (i.e., it can additionally involve operations over windows and/or relations).
\end{definition}

\begin{definition}
{\bf Workflow} is a directed acyclic graph (DAG) in which nodes represent streaming transactions and edges represent an execution ordering between them.
\end{definition}

\begin{definition}
{\bf Nested Transaction} is a group of transactions with a partial order defined among them. A nested transaction will commit if and only if all of its transactions commit, and will abort if one or more of its transactions abort.
\end{definition}

\begin{definition}
{\bf Trigger}, defined over a stream (or over a sliding window), specifies a finite computation that must be automatically executed in response to append of a new atomic batch to that stream (or append of at least a slide worth of tuples to that window signaling the availability of a new full window).
\end{definition}

\subsection{Assumptions}

\begin{itemize}

\item Stream order 

\item Single computational node, multiple processing sites per node.

\item Serial transaction execution per site.

\item Transactions are defined as stored procedures with input parameters.

\end{itemize}

experiments.tex
\section{Experiments} \label{sec:experiments}

\begin{figure}[t]
\begin{center}
\subfigure[EE Trigger Micro-benchmark]{
    \includegraphics[width=0.45\textwidth]{graphs/EE-trigger-experiment.png}
    \label{f:EE-trigger-mb}
} \\
\subfigure[EE Trigger Result]{
    \includegraphics[width=0.45\textwidth]{graphs/BTriggers.png}
    \label{f:EE-trigger-exp}
}
\caption{Execution Engine Triggers}
\label{f:EE-triggers}
\vspace{-0.3in}
\end{center}
\end{figure}

\begin{figure}[t]
\begin{center}
\subfigure[PE Trigger Micro-benchmark]{
    \includegraphics[width=0.41\textwidth]{graphs/PE-trigger-experiment.png}
    \label{f:PE-trigger-mb}
} \\
\subfigure[PE Trigger Result]{
    \includegraphics[width=0.45\textwidth]{graphs/FTriggers.png}
    \label{f:PE-trigger-exp}
}
\caption{Partition Engine Triggers}
\label{f:PE-triggers}
\vspace{-0.3in}
\end{center}
\end{figure}

%While S-Store is primarily designed to improve the functionality of streaming systems, we have also found that our added features often improve transactional throughput as well. Below we compare S-Store to the state-of-the-art, in both transaction processing and stream processing systems.
In this section, we present the results of our experimental study that evaluates S-Store with respect to existing  alternatives in OLTP and stream processing. More specifically, we compare S-Store to H-Store, Spark Streaming and Storm in terms of overall transaction throughput on a number of workloads. Sections \ref{sec:EE-triggers}-\ref{sec:recovery-exp} explore a number of micro-benchmarks that focus on evaluating specific architectural features of S-Store in comparison to its base system H-Store (e.g., triggers, windows, recovery modes).  The benchmarks described in Sections \ref{sec:voter}-\ref{sec:storm-spark} are based on our leaderboard maintenance application (described in Section \ref{sec:example}). This represents a more realistic use case, and we use it to compare S-Store to H-Store, Storm and Spark Streaming. The benchmark details are described below together with their corresponding results. We also provide early insight into S-Store's scalability in Section \ref{sec:multisite}.  Here we implement a subset of the Linear Road benchmark \cite{arasu2004linear} in order to demonstrate S-Store's ability to effectively parallelize streaming workflows in the case of highly partitionable workloads.

To properly evaluate streaming workloads, we record throughput in terms of ''workflows per second''.  This number represents the number of full workflows that complete end-to-end, regardless of the number of transactions required to complete those workflows.  In cases where workflows necessarily consist of only a single transaction, we simplify our unit to ''transactions per second''. All experiments were run on a cluster of Intel Xeon machines, each containing 64 cores and 264 GB of memory. Because we focus on single-node S-Store deployments in this paper and due to the partitioned architecture of S-Store, effectively only a single core is used for data access (with the exception of Section \ref{sec:multisite}). The experiments were run using a single non-blocking client which asynchronously sends requests to the system. Logging was disabled unless otherwise specified.

\subsection{Execution Engine Triggers} \label{sec:EE-triggers}

%In our microbenchmarks, we compare features native to S-Store to comparable implementations within its predecessor, H-Store.
In this experiment, we evaluate the benefit of S-Store's EE triggers. The micro-benchmark contains a single stored procedure that consists of a sequence of SQL statements that simulate an input stream flowing through a set of query stages (Figure \ref{f:EE-trigger-mb}). In S-Store, these SQL statements can be activated using EE triggers such that all execution takes place inside the EE layer. H-Store, on the other hand,
%contains all of these SQL statements in the PE, and
must submit the set of SQL statements (an insert and a delete) for each query stage as a separate execution batch from PE to EE. Figure \ref{f:EE-trigger-mb} illustrates the case for 3 streams and query stages. S-Store's EE triggers enable it to trade off trigger execution cost for reducing the number of PE-to-EE round-trips (e.g., 2 triggers instead of 2 additional round-trips). Note also that the deletion statements are not needed in S-Store, since garbage collection on streams is done automatically as part of our EE trigger implementation.

Figure \ref{f:EE-trigger-exp} shows how transaction throughput changes with varying number of EE triggers. S-Store outperforms H-Store in all cases and its relative performance further increases with increasing number of EE triggers, reaching up to a factor of 2.5x for 10 triggers.
%Our findings show that the additional overhead of submitting separate batches to the EE results in roughly a 100 percent increase in throughput for S-Store, as shown in Figure \ref{f:EE-trigger-exp}.  This benefit is maintained regardless of the number of triggers in the sequence.

%\newpage

\subsection{Partition Engine Triggers} \label{sec:PE-triggers}

\begin{comment}
\begin{figure*}[t]
\begin{center}
\subfigure[b][Window Micro-benchmarks]{
    \includegraphics[width=0.5\textwidth]{images/windowing.png}
    \label{f:window-mb}
}
\begin{minipage}[b]{0.45\textwidth}
\subfigure[b][Window Size Result]{
    \includegraphics[width=\linewidth]{graphs/WindowsSize.png}
    \label{f:window-size-exp}
} \\ [\baselineskip]
\subfigure[b][Window Slide Result]{
    \includegraphics[width=\linewidth]{graphs/WindowSlide.png}
    \label{f:window-slide-exp}
}
\end{minipage}
\caption{Windows}
\label{f:windows}
\end{center}
\end{figure*}
\end{comment}

This experiment compares the performance of S-Store's PE triggers to an equivalent implementation in H-Store, which has no such trigger support in its PE. As illustrated in Figure \ref{f:PE-trigger-mb}, the micro-benchmark consists of a workflow with a number of identical stored procedures that must execute one after the other. We assume that the workflow must execute in exact sequential order, as is the case with many streaming workflows that require the maintenance of consistent state across the workflow. In H-Store, the scheduling request of a new transaction must come from the client, and because the workflow order of these transactions must be maintained, transactions cannot be submitted asynchronously. Serializing transaction requests severely limits H-Store's performance, as the engine will be unable to perform meaningful work while it waits for a client request. In S-Store, a PE trigger can activate the next transaction directly within the PE and can prioritize these triggered transactions ahead of the current scheduling queue using its streaming scheduler. Thus, S-Store again trades off additional trigger execution overhead for avoiding unnecessary round-trips to the client layer.

Figure \ref{f:PE-trigger-exp} shows how transaction throughput (measured in workflows/sec and plotted in log-scale) changes with increasing workflow size (shown in number of PE triggers for S-Store). 
As can be seen, H-Store's throughput tapers early due to the PE's need to wait for the client to determine which transaction is next to schedule.  
%excessive communication between the PE and the client.
%as too many workflows are submitted for the system to handle.
S-Store is able to process roughly an order of magnitude more workflow instances per second thanks to its PE triggers. This performance benefit becomes even more pronounced as the number of transactions per workflow increases.


\vspace{2mm}
\subsection{Windows} \label{sec:windows-exp}

\begin{figure}[!t]
\begin{center}
\subfigure[b][Window Micro-benchmarks]{
   \includegraphics[width=0.45\textwidth]{graphs/windowing.png}
    \label{f:window-mb}
} \\
\vspace{-0.1in}
\subfigure[b][Window Result]{
   \includegraphics[width=0.5\textwidth]{graphs/WindowsSize.png}
    \label{f:window-size-exp}
} \\
%\begin{comment}
%\subfigure[b][Window Slide Result]{
%   \includegraphics[width=0.5\textwidth]{graphs/WindowSlide.png}
%    \label{f:window-slide-exp}
%}
\vspace{-0.1in}
\caption{Windows}
\vspace{-0.3in}
\label{f:windows}
\end{center}
%\end{comment}
\end{figure}

In this micro-benchmark, we compare native windowing support provided by S-Store within its EE to a comparable H-Store implementation without such built-in windowing support. As mentioned in Section \ref{sec:windows}, S-Store employs a tuple-wise ordering strategy and flags new tuples for ''staging'' as they arrive, keeping them inactive until the window slides.  The most fair strategy for H-Store is to use a similar technique, in which separate columns are maintained to indicate the absolute ordering of tuples, as well as a flag denoting which tuples are active within the window. If enough tuples have arrived to slide the window, then ''staged'' tuple flags are removed and expired tuples are deleted.

Figure \ref{f:window-mb} depicts the micro-benchmark we used in this experiment, which simply inserts new tuples into a tuple-based sliding window and maintains the window as tuples continue to arrive. As seen on the left-hand side of the figure, S-Store can implement this micro-benchmark simply by defining the window with its desired parameters and issuing insertion queries over it. An equivalent implementation in H-Store (shown on the right-hand side of Figure \ref{f:window-mb}) would require the maintenance of both a window and a metadata table, with a two-staged stored procedure to manage the window state using a combination of SQL queries and Java logic.

Figure \ref{f:window-size-exp} shows the transaction throughput achieved by S-Store and H-Store as window size varies. Our native windowing implementation grants us roughly a 2x throughput gain over the naive implementation of H-Store. This gain primarily comes from maintaining window statistics (total window size, number of tuples in staging, etc.) within the table metadata, allowing window slides and tuple expirations to happen automatically when necessary. It is also important to note that variations in window size provide a much larger contribution to the performance difference than window slide variations. This is because slide frequency only affects the frequency of two SQL queries associated with window management (shown in italic in Figure \ref{f:window-mb}).

% and \ref{f:window-slide-exp}


\subsection{Recovery Mechanisms} \label{sec:recovery-exp}

As described earlier in Sections \ref{sec:fault} and \ref{sec:recoverymechanisms}, S-Store provides two methods of recovery: {\em strong recovery}, in which all transactions are individually written to the command-log, and {\em weak recovery}, a version of upstream backup in which only border stored procedures are logged, and PE triggers allow interior stored procedures to be automatically activated during log replay. We now investigate the performance differences caused by logging overhead and recovery-time for these two methods.

For this experiment, we use the same micro-benchmark presented in Section \ref{sec:PE-triggers} (Figure \ref{f:PE-trigger-mb}), using a variable number of PE triggers within the sequence. Ordinarily in H-Store, higher performance is achieved during logging by grouping many transactions and committing them together, writing their log records to disk in batches. There are situations, however, when group committing is not possible, and each transaction must be logged and committed separately.  For trigger-heavy workloads, weak recovery can accomplish a similar run-time effect even without the use of group commit. As seen in Figure \ref{f:logginggraph}, without group commit, logging quickly becomes a bottleneck in the strong recovery case.  Each transaction is logged, so the workflow throughput quickly decreases as the number of transactions per workflow increases.  By contrast, weak recovery writes to the log for only the border transactions, allowing up to 4x the throughput as it writes a smaller fraction of log records to disk.

Figure \ref{f:recoverygraph} shows the striking result that weak recovery not only achieves better throughput during normal operation, but it also provides lower recovery time. Typically during recovery, the log is read by the client and transactions are submitted sequentially to the engine. Each transaction must be confirmed as committed before the next can be sent. Because weak recovery activates non-border transactions within the engine, the transactions can be confirmed without a round-trip to the client. As a result, latency stays roughly constant for weak recovery even for workloads with large numbers of PE triggers, while for strong recovery it increases linearly with the number of PE triggers.

\subsection{Leaderboard Maintenance} \label{sec:voter}

\begin{figure}[t]
\centering
\includegraphics[width=0.45\textwidth]{graphs/VoterDemo.png}
\vspace{-0.1in}
\caption{Leaderboard Maintenance}
\vspace{-0.2in}
\label{f:votergraph}
\end{figure}

This experiment compares S-Store's throughput performance aga\-inst that of H-Store on our leaderboard maintenance example previously described in Section \ref{sec:intro}. As was shown in Figure \ref{f:LB-workflow}, the workflow contains three separate stored procedures: one to validate and insert a new vote, a second to maintain the leaderboard, and a third to delete a candidate if necessary. In order to ensure the correctness of the result in the presence of shared tables, as well as to maintain consistency of the tables across the workflow, these three stored procedures must execute in sequence for each new vote. This benchmark practices all of the architectural additions of S-Store described in Section \ref{sec:implementation}.

As seen in Figure \ref{f:votergraph}, much like the results of our micro-benchmarks, S-Store maintains a significant throughput advantage over its OLTP predecessor. More specifically, as input rate in terms of votes/sec increases, both systems begin by displaying the same workflow throughput. However, as the rate goes beyond 500 workflows per second, H-Store is unable to keep up, while S-Store continues to successfully process nearly 3000 votes/sec. As expected, the main reason for this result is S-Store's use of PE triggers, which avoids redundant round-trips to the client. The streaming scheduler also ensures that transactions are run in proper sequence, while also allowing future transactions to be asynchronously queued. H-Store has no method of providing this queueing guarantee.  Thus, to prevent incorrect results, it must wait for the client to explicitly determine the next transaction to queue based on the result of the previous transaction. EE triggers and native windowing also contribute to the throughput increase, though to a lesser degree than PE triggers and the streaming scheduler.

\subsection{Modern Streaming Systems Comparison} \label{sec:storm-spark}

\begin{figure*}[t]
\begin{center}
\subfigure[Logging]{
\includegraphics[width=0.45\textwidth]{graphs/LoggingFTriggers.png}
\label{f:logginggraph}
}
\subfigure[Recovery]{
\includegraphics[width=0.45\textwidth]{graphs/RecoveryGraph.png}
\label{f:recoverygraph}
}
\vspace{-0.3in}
\caption{Recovery Mechanisms}
\label{f:logging-recovery}
\vspace{-0.2in}
\end{center}
\end{figure*}

In order to compare S-Store against state-of-the-art streaming systems, we chose to implement two variants of our leaderboard maintenance benchmark with Storm \cite{storm} and Spark Streaming \cite{zaharia13}. For the Storm implementation, we opted to use Trident\cite{trident-wiki}, an extension of Storm that supports stateful stream processing and exactly-once semantics. The following experiments compare the single-node performance of S-Store against Storm with Trident and Spark Streaming.

\subsubsection{Spark Streaming} \label{sec:spark}
As Spark Streaming does not include any notion of transactions and supports only time-based windows, we have simplified our benchmark to a single stored procedure which handles vote validation and recording while maintaining a time-windowed leaderboard (10 second windows, sliding every 1 second). The Spark benchmark uses Redis for vote state and RDDs for durability.

Like most streaming systems, Spark Streaming inherently has no concept of a true ACID transaction; we consider a Spark batch to be the proper analog to a transaction within S-Store. In our S-Store experiments, each vote is submitted as its own separate transaction (i.e., input batch size is a single tuple). We considered submitting only a single vote per Spark batch and minimizing the duration per batch in order to gain a similar level of isolation and atomicity to S-Store transactions. However, it quickly became apparent that Spark's performance would be extremely poor under these conditions. Spark Streaming primarily gains its performance through batch-processing and parallelization, so to remove both elements would provide an unfair comparison. 

\subsubsection{Storm with Trident} \label{sec:storm}
Ordinarily, Storm is stateless and guarantees either at-least-once or at-most-once semantics. Trident improves on Storm by offering much stronger transactional guarantees, adding stateful, incremental processing on top of any database \cite{trident-wiki}, as well as exactly-once processing. In Trident, we create two "Bolts" (Storm's equivalent of stored procedures): one that validates votes and another that maintains a leaderboard. As Trident has no notion of windows, sliding window functionality must be implemented with temporal state management and event triggering. State management for vote validation requires indexing; this is built using Memcached. Additionally, workflows are logged asynchronously using Storm's logging capabilities to ensure durability of processed streaming data. 

%We instead implemented the same benchmark in a way that allows multiple votes to be submitted in a single batch. Batch duration was set to 500 ms for this comparison, with each vote being validated individually (i.e., tuples are compared with all other votes, even with those inserted within the same batch).
\subsubsection{Comparison} \label{sec:sstore-comparison}
For S-Store, we ran experiments on the original transactional version of the benchmark (one vote per batch) with logging. As shown in the left-hand side of Figure \ref{f:sparkgraph}, S-Store and Trident perform comparably, but the state handling and interface that provides exactly-once guarantees impairs Storm's performance \cite{storm}. Spark exhibits surprisingly lower performance on this particular workload against S-Store. This is mainly caused by the validation process, in which a new tuple's phone number is verified as having not previously voted. This operation requires comparison to all previous votes that have been recorded, and happens each time a new vote arrives. Spark Streaming provides no method of indexing over state, and thus validation becomes extremely expensive as the total number of votes grows. S-Store, on the other hand, is able to handle this task easily via an index on the ''phone number'' column of the Votes table, providing a lookup rather than a table scan.

%Not RDD, it's a Redis pool. I've mentioned Redis in Spark's benchmark description, leaving this specifically vague-r.


In order to better compare against Spark's strengths, we created a second variant of the same benchmark that does not rely so heavily on indexed tables. This version is identical to the first, except that the vote validation against previous records is removed. The remaining portions of the workload are easily expressed as map-reduce operations (i.e., natural to express in Spark's programming model). As shown on the right-hand side of Figure \ref{f:sparkgraph}, neither S-Store nor Trident's performance show significant increase in throughput, proving that leaderboard maintenance remains the bottleneck for their performance. In the case of Trident, the lack of built-in windowing functionality curbs its overall performance. Spark Streaming performs over an order of magnitude better on this simplified workload than it does on the original. However, even after removing Spark's major bottleneck from the workload, S-Store is able to match both Spark and Storm's performance while providing much stronger transactional guarantees than either.

%Unlike S-Store and Spark, the lack of built-in windowing functionality on Trident streams curbs its overall performance. NOTE: removed mention of Spark's exactly-once processing guarantee
%RE:Note: What about built-in windows?


\begin{figure}[t]
\centering
\includegraphics[width=\columnwidth]{graphs/SparkStormGraph.png}
\vspace*{-12pt}
\caption{Voter w/ Leaderboard on Modern SDMSs}
\vspace{-0.2in}
\label{f:sparkgraph}
\end{figure}

\subsection{Multi-Core Scalability} \label{sec:multisite}

\begin{figure}[t]
\centering
\includegraphics[width=0.9\columnwidth]{graphs/multicore.png}
\vspace*{-12pt}
\caption{Multi-core Scalability on S-Store}
\vspace{-0.2in}
\label{f:multicore}
%\vspace{-0.1in}
\end{figure}

We are currently extending S-Store to operate on multiple nodes. This will allow us to scale to both multiple cores and machines by intelligently dividing the workload (both data and workflows) to be processed in parallel. While a complete solution is still in progress, S-Store is already able to leverage H-Store's distributed architecture described in Section \ref{sec:hstore} to parallelize workflow processing for highly partitionable workloads. Specifically, S-Store is able to partition an input stream onto multiple cores.  Each core runs TE's of the complete workflow in a serial, single-sited fashion for the input stream partition to which it is assigned. Our serial scheduling as well as both weak and strong recovery mechanisms apply to each core as described in Section \ref{sec:arch}.

In order to test S-Store's scalability on multiple cores for this type of workload scenario, we have implemented a subset of the Linear Road data stream benchmark \cite{arasu2004linear}. Linear Road simulates traffic on a specified number of "x-ways" and requires that notifications about tolls and accidents be dynamically delivered to vehicles based on their position reports. For the purposes of demonstrating partitioned stream workflows, it was sufficient to focus only on the queries over streaming position reports; thus, we excluded the historical queries of the benchmark. We simulated traffic data for 30 minutes, and evaluated performance by comparing the number of x-ways we are able to support while ensuring that all position reports are processed under a given latency threshold. Because we are running an abbreviated version of the benchmark, we set our latency threshold to be 1 second rather than the recommended 5 seconds.

We implemented this workload as a workflow that consists of two stored procedures. SP1 updates a vehicle's position and detects whether it has entered a new segment. If it has, SP1 sends a notification to the vehicle of any tolls or accidents recorded for upcoming road segments in the 1st minute, and charges the vehicle tolls from the previous segment. SP1 is also responsible for detecting stopped cars and accidents. At the beginning of each minute, SP1 triggers SP2, which calculates the tolls for the previous minute and records the current statistics for every x-way table into a historical statistics table. SP2 also checks all accidents to see whether the vehicles involved have been removed from the scene. Accidents and tolls are all calculated relative to the events happening on a single x-way, and thus we distribute the x-ways evenly across partitions. Because each partition runs transactions serially, we expect to achieve roughly linear scalability as we increase the number of cores.


Figure \ref{f:multicore} confirms our expectations.  S-Store is able to support 16 x-ways per core when a single partition is used, and scales roughly linearly, with about 5-10 percent drop-off per added core due to additional partition maintenance.  At the upper end of the graph (8 cores), 11 x-ways per core are able to be supported.  S-Store is best able to support a number of x-ways that is divisible by the number of cores that are available.  For instance, 3 cores are able to support 39 x-ways, but adding a 40th will disproportionately increase latency as the load between cores becomes asymmetric.

\begin{comment}

While a truly distributed system is still in development and is a topic for future research (Section \ref{sec:summary}), S-Store is currently able to utilize multi-core architectures in cases where the workload is highly partitionable.  As described in Section \ref{sec:hstore}, H-Store achieves scalability while maintaining data isolation by partitioning the data across multiple sites and serializing transaction executions on each site.  S-Store is able to leverage this architecture to parallelize streaming workflows.  When one stored procedure triggers another SP in a workflow, the downstream transaction execution is initialized on the same site as the first TE.  In essence, this chain of single-sited TEs allows multiple instances of the same workflow to run simultaneously in parallel.  Workflows partitioned in this way maintain all of the properties of multiple single-sited workflows, including serialized scheduling (Section \ref{sec:scheduler}) and both strong and weak recovery options (Section \ref{sec:recoverymechanisms}).

We are currently extending S-Store to operate on multiple nodes. This will allow us to scale to both multiple cores and machines by intelligently dividing the workload (both data and workflows) to be processed in parallel. While a complete solution is still in progress, S-Store is already able to leverage H-Store's distributed architecture described in Section \ref{sec:hstore} to parallelize workflow processing for certain types of workloads. More specifically, we can partition the input stream
onto multiple cores. Each core then runs TE's of the complete workflow in a serial, single-sited fashion for the input stream partition that is assigned to that core. Our serial scheduling as well as both weak and strong recovery mechanisms apply to each core as described in Section \ref{sec:arch}.

In order to test S-Store's scalability on multiple cores for this type of workload, we have implemented a subset of the Linear Road data stream benchmark \cite{arasu2004linear}. Linear Road simulates traffic on a specified number of "x-ways" and requires that notifications about tolls and accidents be dynamically delivered to vehicles based on their position reports. For the purposes of this experiment, it was sufficient to focus only on the queries over streaming position reports, thus we excluded the historical queries of the benchmark. We simulated traffic data for 30 minutes, and evaluated performance by comparing the number of x-ways we are able to support, while ensuring that all position reports are processed under a given latency threshold. Because we are running an abbreviated version of of the benchmark, we set our latency threshold to be 1 second rather than the recommended 5 seconds.

In implementing Linear Road for S-Store, we divide the workload into two stored procedures.  SP1 updates the vehicle's position, and detects whether it has entered a new segment.  If it has, SP1 sends a notification to the vehicle of any tolls or accidents recorded for upcoming road segments in the past minute, and charges the vehicle tolls from the previous segment.  SP1 is also responsible for detecting stopped cars and accidents.  At the beginning of each minute, SP1 triggers SP2, which calculates the tolls for the previous minute and records the current statistics for every x-way table into a historical statistics table.  SP2 also checks all accidents to see whether the vehicles involved have been removed from the scene.  Accidents and tolls are all calculated relative to the events happening on a single x-way, and thus we assign an equal number of x-ways to each partition.  Because each partition runs transactions serially, we expect to achieve close to linear scalability as we increase the number of cores.

We implemented this workload as a workflow that consists of two stored procedures. SP1 updates a vehicle's position and detects whether it has entered a new segment. If it has, SP1 sends a notification to the vehicle of any tolls or accidents recorded for upcoming road segments in the 1st minute, and charges the vehicle tolls from the previous segment.
SP1 is also responsible for detecting stopped cars and accidents. At the beginning of each minute, SP1 triggers SP2, which calculates the tolls for the previous minute and records the current statistics for every x-way table into a historical statistics table. SP2 also checks all accidents to see whether the vehicles involved have been removed from the scene. Accidents and tolls are all calculated relative to the events happening on a single x-way, and thus we assign an equal number of x-ways to each partition. Because each partition runs transactions serially, we expect to achieve roughly linear scalability as we increase the number of cores.

As shown in Figure \ref{f:multicore}, S-Store is able to support roughly 11 x-ways per core with very little performance drop-off.

\end{comment}

formal.tex
% Formal stuff for isolation

Consider a workflow $W$ of $n$ streaming transactions $T_i$, $1 \leq i \leq n$. $W$ is a directed acyclic graph $G=(V,E)$, where $V=\{T_1, \ldots, T_n\}$ and $E \subseteq V \times V$, where $(T_i, T_j) \in E$ means that $T_i$ must precede $T_j$.

Being a DAG, $G$ has at least one topological ordering. A topological ordering of $G$ is an ordering of its nodes $T_i \in V$ such that for every edge $(T_i, T_j) \in E$ we have $i < j$. Each topological ordering of $G$ is essentially some permutation of $V$.

Without loss of generality:
(i) Let us focus on one specific topological ordering of $G$ and call it $O$;
(ii) For ease of notation, let us simply assume that $O$ corresponds to the identity permutation such that it represents: $T_1 < T_2 < .. < T_n$.

$T_i$ represents a transaction definition $T_i(s_i, w_i, p_i)$, where
$s_i$ denotes all private stream inputs of $T_i$,
$w_i$ denotes all private window inputs of $T_i$,
$p_i$ denotes all public table partition inputs of $T_i$.

$T_i^j$ represents the $j^{\text{th}}$ transaction execution of $T_i$, $T_i^j(s_i.b_j, w_i, p_i)$, where $b_j$ denotes the  $j^{\text{th}}$ atomic batch of $s_i$.

A workflow $W$ is executed in rounds of atomic batches $1 \leq r < \infty$, such that:

For round $r=1$, atomic batch 1 from all streaming inputs into $W$ generates a sequence of TE's $T_i^1(s_i.b_1, w_i, p_i)$ for each $T_i$.
 
For round $r=2$, atomic batch 2 from all streaming inputs into $W$ generates a sequence of TE's $T_i^2(s_i.b_2, w_i, p_i)$ for each $T_i$. 

$\ldots$, and so on.

Note that this execution generates an unbounded schedule. However, as of a specific round $r=R$, we generate a bounded schedule that consists of $R*n$ TE's: $1 \leq r \leq R, 1 \leq i \leq n, T_i^r(s_i.b_r, w_i, p_i)$.

In the traditional ACID model of databases, any permutation of these $R*n$ TE's is considered to be a valid/correct, serial schedule.

In our model, we additionally have:

\begin{enumerate}

\item {\em Workflow order constraints:}
Consider the topological ordering $O$ of $W$ as defined above. Then for any given round $r$, it must hold that:

$T_1^r(s_1.b_r, w_1, p_1) < T_2^r(s_2.b_r, w_2, p_2) < \ldots < T_n^r(s_n.b_r, w_i, p_i)$ 

\item {\em Stream order constraints:}
For any given transaction $T_i$, as of any round $r=R$, the following must hold:

$T_i^1(s_i.b_1, w_i, p_i) < T_i^2(s_i.b_2, w_i, p_i) < \ldots < T_i^R(s_i.b_R, w_i, p_i)$

\end{enumerate}

Any bounded schedule of $W$ that meets the above two ordering constraints is a correct schedule.

{\bf OLTP Transactions:} Any OLTP transaction execution $T_i^j(p_i)$ can interleave anywhere in a correct bounded schedule of $W$. The resulting bounded schedule would still be correct.

{\bf Nested Transactions:} Any streaming transaction $T_i$ in workflow $W$ can be defined as a nested transaction that consists of children $T_{i1}, \ldots, T_{im}$. In this case, $T_{i1}, \ldots, T_{im}$ must obey the partial order defined for $T_i$ for every execution round $r$, $1 \leq r < \infty$. This means that no other streaming or OLTP transaction instance can interleave with $T_{i1}, \ldots, T_{im}$ for any given execution round.

hstore.tex
\section{H-Store Overview} \label{sec:hstore}

We build the XXX prototype on top of H-Store - an open-source, main-memory OLTP engine that was developed at Brown and MIT \cite{hstore} and formed the basis for the design of the VoltDB NewSQL database system \cite{voltdb}. This section will briefly describe the architecture and implementation of H-Store.

All transactions in H-Store must be predefined as stored procedures with input parameters. A {\em stored procedure} (SP) is a transaction definition that specifies the code for a transaction execution. The stored procedure code is a mixture of SQL and Java. A {\em transaction execution} (TE) is an instantiation of a transaction definition. It is what you get when you bind input parameters to real values and run the transaction. In general, a given stored procedure will, over time,  generate many transaction executions. Transaction executions are submitted to H-Store, and the H-Store scheduler executes them in whatever order it deems best. 

H-Store follows a typical distributed architecture in which there is a layer (called {\em the partion engine (PE)}) that is responsible for managing transaction distribution, scheduling, coordination, and recovery. The PE itself is built on top of another layer that is responsible for the local execution of SQL queries (called {\em the execution engine (EE)}). In the case of H-Store, the PE is written in Java and the EE is written in C++ (for easier control and tuning of the underlying system resources). A client program connects to the PE via a stored procedure execution request. If the stored procedure requires SQL processing, then the EE is invoked with these sub-requests.

An H-Store database is partitioned across multiple sites \cite{pavlo12}. A transaction is executed on the sites that hold the data that it needs. If the data is partitioned carefully, most transactions will only need data from a single site. Single-sited transactions are run serially on that node, thereby eliminating the need for fine-grained locks and latches.

H-Store provides recovery through a checkpointing and command-logging mechanism \cite{malviya14}. Periodically, the system creates a persistent snapshot or checkpoint of the current committed state of the database. To create a snapshot, the system must quiesce its running transactions. That is, it must delay new transactions and wait until all in-flight transactions commit or abort. When the system has quiesced, it persistently records the state of {\em all} tables. Thus, the snapshot captures a complete and consistent state of the database.

After a snapshot is complete, the system resumes normal operation. Every time it commits a transaction, it writes a command-log record containing the name of the corresponding stored procedure along with its input parameters. This command-log record must be made persistent before its transaction can commit. In order to minimize intereactions with the slow persistent store, H-Store offers a group-commit mechanism. Group-commit collects a batch of commit records and writes them as a group to persistent storage. This amortizes the I/O cost at the expense of delaying actual commits.

Note that updates can be done in-place, since a recovery action forces the system back to the last checkpoint, thereby, wiping out all changes that have been made since then. Thus, an undo log is unnecessary.

On recovery, the system's state is restored to the latest snapshot, and the command-log is replayed. That is, each command-log record allows the system to re-execute its associated stored procedure with the same arguments that it did before the failure.

It should be noted that replaying the command-log can require more work than using an Aries-style physical write-ahead log (WAL). H-Store is willing to accept this fact, since failure is relatively rare, and command-logging is lighter-weight than WAL during normal operation \cite{malviya14}.
55
implementation.tex
\section{Architecture \& Implementation} \label{sec:implementation}

We chose to build S-Store on top of the H-Store OLTP system \cite{hstore}. This allows us to inherit H-Store's support for high-throughput transaction processing, thereby eliminating the need to replicate this complex functionality. We also get important associated functionality like indexing, main-memory operation, and support for user-defined transactions that will be important for streaming OLTP applications.

In this section, we describe the H-Store architecture and the chan\-ges that were required to incorporate S-Store's hybrid computational model described in the previous section. An important design goal was to make use of H-Store's transaction model as much as possible, making as few changes and additions to the code-base as possible. Nevertheless, we believe that the architectural features that we have added to H-Store are conceptually applicable to any main-memory OLTP system in general.

\subsection{H-Store Overview} \label{sec:hstore}

H-Store is an open-source, main-memory OLTP engine that was developed at Brown and MIT \cite{hstore}, and formed the basis for the design of the VoltDB NewSQL database system \cite{voltdb}.
%This section will briefly describe the architecture and implementation of H-Store.

All transactions in H-Store must be predefined as stored procedures with input parameters.
%A {\em stored procedure} (SP) is a transaction definition that specifies the code for a transaction execution.
The stored procedure code is a mixture of SQL and Java.
%A {\em transaction execution} (TE) is an instantiation of a transaction definition. It is what you get when you bind input parameters to real values and run the transaction. 
Transaction executions are instantiated by binding input parameters of a stored procedure to real values and running it. In general, a given stored procedure will, over time,  generate many transaction executions. Transaction executions are submitted to H-Store, and the H-Store scheduler executes them in whatever order it deems best in order to provide ACID guarantees.

H-Store follows a typical distributed architecture in which there is a layer (called {\em the partion engine (PE)}) that is responsible for managing transaction distribution, scheduling, coordination, and recovery. The PE itself is built on top of another layer (called {\em the execution engine (EE)}) that is responsible for the local execution of SQL queries.
%In the case of H-Store, the PE is written in Java and the EE is written in C++ (for easier control and tuning of the underlying system resources).
A client program connects to the PE via a stored procedure execution request. If the stored procedure requires SQL processing, then the EE is invoked with these sub-requests.

An H-Store database is partitioned across multiple sites \cite{pavlo12}. A transaction is executed on the sites that hold the data that it needs. If the data is partitioned carefully, most transactions will only need data from a single site. Single-sited transactions are run serially on that site, thereby eliminating the need for fine-grained locks and latches.

H-Store provides recovery through a checkpointing and command-logging mechanism \cite{malviya14}. Periodically, the system creates a persistent snapshot or checkpoint of the current committed state of the database.
%To create a snapshot, the system must quiesce its running transactions. That is, it must delay new transactions and wait until all in-flight transactions commit or abort. When the system has quiesced, it persistently records the state of {\em all} tables. Thus, the snapshot captures a complete and consistent state of the database. After a snapshot is complete, the system resumes normal operation.
Furthermore,
every time H-Store commits a transaction, it writes a command-log record containing the name of the corresponding stored procedure along with its input parameters. This command-log record must be made persistent before its transaction can commit. In order to minimize interactions with the slow persistent store, H-Store offers a group-commit mechanism.
%Group-commit collects a batch of commit records and writes them as a group to persistent storage. This amortizes the I/O cost at the expense of delaying actual commits.

%Note that updates can be done in-place, since a recovery action forces the system back to the last checkpoint, thereby, wiping out all changes that have been made since then. Thus, an undo log is unnecessary.

On recovery, the system's state is restored to the latest snapshot, and the command-log is replayed. That is, each command-log record allows the system to re-execute its associated stored procedure with the same arguments that it did before the failure. Note that an undo-log is unnecessary, since the previous checkpoint will not contain uncommitted changes.

%It should be noted that replaying the command-log can require more work than using an Aries-style physical write-ahead log (WAL). H-Store is willing to accept this fact, since failure is relatively rare, and command-logging is lighter-weight than WAL during normal operation \cite{malviya14}.

\subsection{S-Store Extensions} \label{sec:arch}

\begin{figure*}[t]
\centering
\includegraphics[width=0.7\textwidth]{images/architecture3.png}
\caption{S-Store Architecture}
\vspace{-0.2in}
\label{f:architecture}
\end{figure*}

%S-Store adopts H-Store's distributed database architecture that we briefly summarized in Section \ref{sec:hstore}. Figure \ref{f:architecture} provides an overview of this architecture together with the extensions made by S-Store. Like H-Store, XXX is essentially a client-server system that consists of three fundamental layers (one for the client and two for the database server engine):
The high-level architecture of S-Store, directly adopted from H-Store, is shown in Figure \ref{f:architecture} and consists of four main components:

\begin{compactitem}

\item a \textbf{client} process that invokes stored procedures with their correct input parameters. This is essentially where an S-Store application runs.

\item a \textbf{stream injection module} that feeds streaming inputs into the S-Store engine. This module is responsible for managing all issues related to input handling such as preparing the atomic batches.

\item a \textbf{partition engine (PE)} that is responsible for breaking the client request into pieces that can be executed at the appropriate partition(s). There is one S-Store partition per core. The available DRAM for a node is divided equally among the partitions, and each stores a horizontal slice of the database.
%\footnote{Please remember that we restrict our focus to a single partition and do not discuss the distributed version of XXX in this paper. That implementation is underway and is the topic for a follow-on paper.}.

\item an \textbf{execution engine (EE)} that evaluates SQL expressions. This is called by the partition engine whenever a stored procedure contains SQL.

\end{compactitem}

%XXX directly inherits this base architecture. In addition, it makes a number of extensions to enable native stream processing support in the engine (shown in boldface in Figure \ref{f:architecture}).
S-Store makes a number of extensions to H-Store to enable stream processing in the engine (shown in boldface in Figure \ref{f:architecture}). These include: (i) management of inputs from streaming clients and complex workflows of stored procedures at the PE layer, (ii) management of stream- and window-based queries at the EE layer, (iii) management of in-memory stream and window state.
%In the following subsections, we describe the key features of our design that enabled these architectural extensions: Streams, Windows, and Triggers.
Furthermore, S-Store makes critical extensions to transaction scheduling and recovery mechanisms of the PE's transaction management part.

It is important to observe from Figure \ref{f:architecture} that, when H-Store executes a stored procedure, it requires a minimum of one round trip from the client to the partition engine, and one round trip from the partition engine to the execution engine if the stored procedure contains any SQL. In contrast, S-Store can trigger the next execution step within the appropriate layer using either {\em partition engine triggers (PE triggers)} or {\em execution engine triggers (EE triggers)} directly (see the red dashed horizontal arrows in Figure \ref{f:architecture}). This saves a fair amount of overhead, and, as we shall see later in Section \ref{sec:experiments}, it can even lead to a significant increase in overall transaction throughput.

In what follows, we describe the implementation of our architectural extensions in more detail.

\subsubsection{Streams} \label{sec:streams}

S-Store implements a stream as a time-varying, H-Store table. %This approach provides two advantages: (i) stream state is persistent and recoverable (via periodic snapshots of all table state + command-logging on persistent storage), and (ii) it can seamlessly interact with table state via joins.
Using this approach, stream state is persistent and recoverable. Since tables represent unordered data sets, the order of tuples in a stream is captured based on tuple metadata (i.e., timestamps/tuple-id's and batch-id's). An atomic batch of tuples is appended to the stream table when it is placed on the corresponding stream, and conversely, an atomic batch of tuples is removed from the stream table as it is consumed by a downstream transaction in the workflow. The presence of an atomic batch of tuples within a stream can activate either a SQL plan fragment or a downstream transaction, depending on what ``triggers" are attached to the stream (described in Section \ref{sec:triggers}). In the case of the latter, the current stream table serves as input for the corresponding downstream transaction instance.

\subsubsection{Windows} \label{sec:windows}

%As discussed in Section \ref{sec:streaming}, windows in S-Store operate very much like windows in previous streaming systems. They can be based either on a discrete timestamp or on a specific number of tuples, and they have a specified size and a slide amount.

Similarly to streams, windows are also implemented as time-varying, H-Store tables.
%An exact arrival order is maintained within tuple metadata.
As in streams, arrival order for tuples in a window is tracked via tuple metadata.

As opposed to most traditional streaming systems that processed windows incrementally, in S-Store, SQL queries over windows operate one full window at a time. For a sliding window, a new full window becomes available every time that window has slide-worth of new tuples. Therefore, when new tuples are inserted into a window, they are flagged as ``staged" until slide conditions are met. Staged tuples are not visible to any queries on the window, but are maintained within the window.
%The insertion of any tuple into the window has potential of triggering a slide, if the slide conditions are met.
Upon sliding, the oldest tuples within the window are removed, and the staged tuples are marked as active in their place. All window manipulation is done at the EE level, and output can be activated using an EE trigger (described in Section \ref{sec:triggers}).

Due to the invisible ``staging'' state of a window table as well as the transaction isolation rules discussed earlier in Section \ref{sec:isolation}, special scoping rules are enforced for window state. A window table is not allowed to be accessed in general by transaction executions other than those of the stored procedure that defined it. In fact, a window table must only be visible to future transaction executions of the stored procedure that contains it. As a consequence, one is not allowed to define PE triggers on window state, but only EE triggers. In other words, windows must be contained within the transaction executions of single stored procedures and must not be shared across other stored procedures in the workflow.
%These rules ensure that all window state will be eventually processed by preventing unexpected interactions with the table.  

\subsubsection{Triggers} \label{sec:triggers}

Triggers are the fundamental construct in S-Store that enables push-based, data-driven processing needed by our streaming transactions as well as by workflows. A trigger is associated with a stream or a window table. When new tuples are appended to such a table, downstream processing will be automatically activated depending on where in the system stack the trigger is located. The alternative to triggers would be to poll for newly-arriving tuples, which would be highly inefficient.

There are two types of triggers in S-Store:

\begin{compactitem}

\item {\em Partition engine (PE) triggers} can only be attached to stream tables, and are used to activate downstream stored procedures upon the insertion of a new atomic batch of tuples on the corresponding streams. As the name implies, PE triggers exist to create a push-based workflow within the partition engine by eliminating the need to return back to the client to activate downstream stored procedures. In Figure \ref{f:architecture}, the horizontal arrows between stored procedures inside the PE layer denote PE triggers.

\item {\em Execution engine (EE) triggers} can be attached to stream or window tables, and are used to activate SQL queries within the execution engine. These triggers occur immediately upon the insertion of an atomic batch of tuples in the case of a stream, and upon the insertion of an atomic batch of tuples that also cause a window to slide in the case of a window. The SQL queries are executed within the same transaction instance as the batch insertion which triggered them, and can also activate further downstream EE triggers. EE triggers are designed to eliminate redundant communication between EE and PE layers. In Figure \ref{f:architecture}, the horizontal arrows between SQL queries inside the EE layer denote EE triggers.

\end{compactitem}

S-Store provides automatic garbage collection mechanisms for tuples that expire from stream or window state, after any triggers associated with them have all been fired and executed.

\subsubsection{Streaming Scheduler} \label{sec:scheduler}

Being an OLTP database that implements the traditional ACID model, the H-Store scheduler can execute transaction requests in any order. On a single H-Store partition, transactions run in a serial fashion by design \cite{hstore}. H-Store serves transaction requests from its clients in a FIFO manner by default.

As we discussed in Section \ref{sec:isolation}, streaming transactions and workflows require transaction executions for dependent stored procedures to be scheduled in an order that is consistent with the workflow graph (i.e., not necessarily FIFO). Additionally,
%as shown in Section \ref{sec:isolation},
the application can specify (via defining nested transactions) additional isolation constraints, especially when shared table state among streaming transactions is involved.
The simplest solution to address these scheduling needs on a single partition is to adopt the most constrained approach in which the TE's in a workflow for a given input batch will always be executed in an order consistent with a topological ordering of the workflow. This will always ensure a correct, deterministic execution sequence. S-Store currently implements this approach by short-circuiting H-Store's FIFO scheduler in a way to fast-track stored procedures invoked via PE triggers to the front of the transaction request queue. This enables such TE's to be scheduled right away, preventing any other TE's from interleaving the workflow.

Although somewhat restrictive, we have found this approach to be practical in that it is amenable to a low-overhead implementation in H-Store and good enough to support all the S-Store use cases and benchmarks that we have studied (see Section \ref{sec:experiments}). As we are generalizing S-Store's implementation to the multi-node setting, we will also be extending its scheduler to consider other potential execution orders than FIFO and serial.

\subsubsection{Recovery Mechanisms} \label{sec:recoverymechanisms}

As described in Section \ref{sec:fault}, S-Store provides two different recovery options: (i) {\em strong recovery} is guaranteed to produce exactly the same state as was present before the failure, and (ii) {\em weak recovery} will produce a legal state that could have existed, but is not necessarily the exact state lost. Both of these options leverage periodic checkpointing and command-logging mechanisms of H-Store. However, they differ in terms of which transactions are recorded in the command-log during normal operation and how they are replayed during crash recovery. 

\vspace{0.05in}
\noindent
{\bf Strong Recovery.}
In strong recovery, as in H-Store, we log all transactions, both OLTP and streaming, in the command-log along with their input arguments. When a failure occurs, we replay the command-log in the order in which the transactions committed and were thus recorded in the command-log, starting from the latest snapshot. This will guarantee that the reads-from and the writes-to relationships between the transactions are strictly maintained. However, before this replay, we must first disable all PE-triggers so that the execution of a stored procedure does not redundantly trigger the execution of its successor(s) in the workflow. Once triggers are disabled, the snapshot is applied, and recovery from the command-log can begin. Because every transaction is logged in strong recovery, it is important that PE-triggers remain turned off to ensure that each interior transaction is only run once during the log replay. Failing to do this would create duplicate invocations, and thus potentially incorrect results.

When recovery is complete, we turn PE-triggers back on. At that point, we also check if there are any stream tables that contain tuples in them. For such streams, PE-triggers will be fired to activate their respective downstream transactions. Once those transactions have been queued, then the system can resume normal operation.

\vspace{0.05in}
\noindent
{\bf Weak Recovery.}
In weak recovery, the command-log need not record all stored procedure invocations, but only the ones that ingest streams from the outside (i.e., border stored procedures). We then use a technique similar to {\em upstream backup} \cite{hwang} to re-invoke other (i.e., interior) previously committed stored procedures. The basic idea behind upstream backup is to cache all the data at the inputs to a workflow so that in the event of a failure, the system can replay them in the same way that it did on first receiving them in the live system. Because the streaming stored procedures in an S-Store workflow have a well-defined ordering among them, the replay will necessarily create a correct execution history. While transactions may not be scheduled in the exact order that took place on the original run, some legal transaction order is ensured.

When recovering using weak recovery, we must first apply the snapshot, as usual. However, before applying the command log, S-Store must first check existing streams for data recovered by the snapshot, and fire any PE-triggers associated with those streams. This ensures that interior transactions that were run post-snapshot but not logged are re-executed. Once these triggers have been fired, S-Store can begin reading the log. Unlike for strong recovery, we do not need to turn off PE-triggers during weak recovery. In fact, we rely on PE-triggers for the recovery of all interior stored procedures, as these are not recorded in the command-log. This works fine as long as results are returned through committed tables.

Upstream backup is not a new technique. In fact, it was initially proposed as part of our Aurora/Borealis stream processing engine \cite{hwang}. However, we have found it easier to apply in a transactional setting. The fact that Aurora/Borealis did not have a transaction model made it difficult to know when it was safe to trim backup queues. In S-Store, however, we know that when a transaction commits, the input tuples associated with that transaction are no longer necessary. In other words, the existence of well-defined transactional units (atomic batches and stored procedures) simplifies the problem of queue trimming with upstream backup.
%In addition, we know that transactions will be scheduled using the same method as is used during normal run time. While transactions may not be scheduled in the exact order that took place on the original run, a different execution is possible, but some legal transaction order is ensured.

Weak recovery is a more light-weight alternative than strong recovery, since it need not log all committed transactions. Thus, it requires fewer writes to the command-log, at the expense of possibly longer recovery times. Section \ref{sec:recovery-exp} provides an experimental comparison of our strong and weak recovery mechanisms.

introduction.tex
\section{Introduction} \label{sec:intro}

A decade ago, the database research community focused attention on stream data processing systems. These systems \cite{stream, telegraphcq}, including our own system, Aurora/Borealis \cite{aurora, borealis}, were largely concerned with executing SQL-like operators on an unbounded and continuous stream of input data. The  main optimization goal of these systems was reducing the latency of results, since they mainly addressed what might be called monitoring applications \cite{lerner2003virtues, johnson2008query}. To achieve this, they were typically run in main memory, thereby avoiding the latency-killer of the disk.

While essentially all of the monitoring applications that we encountered had a need for archival storage, the system-level support for this was limited and ad hoc. That is, the systems were largely not designed with storage in mind; it was tacked on after the fact. Thus, there was no support for things like ACID transactions, leaving applications open to potential inconsistencies with weak guarantees for isolation and recovery semantics.

These first-generation streaming systems could be viewed as real-time analytics systems. After all, the input was made up of an infinite stream of new tuples. The notion of some of these tuples representing updates of previously viewed tuples was not made explicit in the model. This is fine if time is the key. In this case, if each tuple is given a unique timestamp, the update pattern is append-only. However, there are cases when the identifying attribute is something else. Consider a stock ticker application in which stock symbol is the key.  Here a new tuple for, say, IBM is really an update to the previously reported price. Traders want to see the current stock book as a consistent view of the 6000 stocks on the NYSE, with all prices reported in a consistent way.

We are beginning to see the rise of a second generation of streaming systems \cite{storm, zaharia13}. These systems do not enforce a relational view on their users. Instead, they allow users to create their own operators that are invoked and managed by a common infrastructure. Note that it is reasonable to have libraries of common operators, including relational operators, that manipulate tables. The infrastructure enforces some model of failure semantics (e.g., at-least-once or exactly-once processing), but still ignores needs of proper isolation and consistent storage in the context of updates.

Meanwhile, the advent of inexpensive, high-density RAM has led to a new-generation of distributed on-line transaction processing (OLTP) systems that store their data in main memory, thereby enabling very high throughput with ACID guarantees (e.g., \cite{voltdb, hekaton, hstore}). However, these systems lack the notion of stream-based processing (e.g., unbounded data, push-based data arrival, ordered processing, windowing).

Many applications need aspects of both streaming and transaction processing. In this paper, we propose to combine these two computational paradigms under a single system called S-Store. Applications that benefit from this kind of a hybrid system include those that use the streaming facilities to record persistent state or views in shared tables, and at the same time use the OLTP facilities to display a representation or summary of this state \cite{sstore-demo}. Obviously, one would want the displayed representation to show a consistent state of the world. Examples here include dashboards with multiple summary displays or leaderboards that show aggregate statistics derived from the shared persistent state. We describe a detailed example below.

\subsection{A Motivating Example} \label{sec:example}

One application domain that benefits greatly from a hybrid system such as S-Store is leaderboard maintenance. For example, consider an American Idol-like TV show, in which a number of candidates are presented to viewers for them to vote for the candidate that they find the most appealing. Each viewer may cast a single vote via text message. Suppose that a preferential voting scheme is used, where the candidate with the fewest number of votes will be removed from the running every 1000 votes, as it has become clear that s/he is the least popular. When this candidate is removed, votes submitted for him or her will be deleted, effectively returning the votes to the people who cast them. Those votes may then be re-submitted for any of the remaining candidates. This continues until a single winner is declared. During the course of the voting, each incoming vote needs to be validated and recorded. Furthermore, several leaderboards are maintained: one representing the top-3 candidates, another for the bottom-3 candidates, and a third one for the top-3 trending candidates of the last 100 votes. With each incoming vote, these leaderboards are updated with new statistics regarding the number of votes each candidate has received.

One can model this application with a workflow that consists of three logical transactional steps, as shown in Figure \ref{f:LB-workflow}: (i) one to validate new votes (i.e., check that the corresponding contestant exists, check that the corresponding viewer has not voted before, etc.) and record them; (ii) another to maintain the leaderboards; and (iii) a third to remove the lowest contestants and votes every 1000 votes, also updating the relevant leaderboards as necessary.

\begin{figure}[t]
\centering \includegraphics[width=\columnwidth]{images/3sp-voter.png}
\caption{Leaderboard Maintenance Workflow}
\vspace{-0.2in}
\label{f:LB-workflow}
\end{figure}

This workflow requires both stream-based processing and transactional state management. Votes continuously arrive as a stream, and the Votes and Contestants tables keep a record of all previously received valid votes together with the viewers who submitted them. Each vote's validation requires a read of the Votes and Contestants tables, followed by a write to the Votes table to record the newly validated vote. The leaderboard maintenance transaction maintains a sliding window over the most recent 100 valid votes in order to aggregate them into a count for trend computation. Additionally, it must maintain a total vote count for each running contestant in order to identify the top and bottom performers. Finally, the last transaction will trigger every 1000 incoming votes to update the Votes, Contestants, and Leaderboard tables for the removed candidate. State is shared among the three transactions, and all state accessed during this workflow (i.e., Votes, Contestants, Leaderboards, Trending Window) must be kept consistent.   In fact, the three transactions in this workflow must always run serially for each incoming vote in order to ensure correct behavior. Otherwise, votes might be validated/invalidated incorrectly and the wrong contestants might be removed, eventually leading to the wrong winner. Last but not least, all state must be maintained persistently and recovered correctly (i.e., in an ACID manner) in the event of a failure.

\subsection{Contributions and Outline} \label{sec:outline}

This paper introduces the design and implementation of S-Store, a single system for processing streams and transactions with well-defined correctness guarantees. Our approach to building such a system is to start with a fully transactional OLTP main-memory database system and to integrate additional streaming functionality on top of it. By doing so, we are able to leverage infrastructure that already addresses many of the implementation complexities of transaction processing. Also, this choice is very natural, since streaming systems largely run in main memory to achieve low latency. More specifically, this work makes the following key contributions:

\noindent
{\bf Model.}
We define a novel, general-purpose computational model that allows us to seamlessly mix {\em streaming transactions} with ordinary OLTP transactions. Stream processing adds additional semantics to an OLTP engine's operating model. In particular, stream processing introduces the notion of {\em order} to the transaction mix. That is, it is possible to say that one transaction must precede another, something that is missing from the non-deterministic semantics of a standard transaction model. Further, since streams are unbounded and arrive on a continuous basis, there is a need to add the necessary primitives for bounding computation on streams such as batch-based processing \cite{stream, streamsql} and windowing \cite{golab-sigrec03, secret}. Finally, streaming transactions support a push-based processing model, whereas OLTP transactions access state in a pull-based manner. Our hybrid model provides a uniform way to represent and access state for all transactions.

\noindent
{\bf Architecture and Implementation.}
We show how such a hybrid computational model can be cleanly and efficiently implemented on top of a state-of-the-art main-memory OLTP engine. While we have designed and implemented S-Store on top of H-Store \cite{hstore}, our architectural extensions are general enough to be applied to any main-memory OLTP engine. These include: (i) streams and windows represented as time-varying state, (ii) triggers to enable push-based processing over streams and windows, (iii) a streaming scheduler that ensures correct transaction ordering, and (iv) a variant on H-Store's recovery scheme \cite{malviya14} that is more appropriate for stream processing. Please note that the discussion in this paper is confined to the single-node case; multi-node S-Store is the topic for a follow-on paper.

\noindent
{\bf Performance.}
We provide a detailed study of S-Store's performance characteristics, including the surprising observation that S-Store can improve transaction throughput over H-Store \cite{hstore}, as well as outperforming Spark Streaming \cite{zaharia13} and Storm \cite{storm}, on a number of benchmarks including the leaderboard example shown in Figure \ref{f:LB-workflow}.

The rest of this paper is organized as follows: We first describe our computational model for transactional stream processing in Section \ref{sec:model}. Section \ref{sec:implementation} presents the design and implementation of the S-Store system, which realizes this model on top of the H-Store main-memory OLTP system \cite{hstore}. In Section \ref{sec:experiments}, we present an experimental evaluation of S-Store in comparison to H-Store as well as to two state-of-the-art stream processing systems - Spark Streaming \cite{zaharia13} and Storm \cite{storm}. We discuss related work in Section \ref{sec:related}, and finally conclude the paper with a summary and a discussion of future research directions in Section \ref{sec:summary}.

main.tex
\documentclass{vldb}

\usepackage{times}
\usepackage{graphicx}
\usepackage{amsmath}
\usepackage{balance}
\usepackage{url}
\usepackage{comment}
\usepackage{enumerate}
\usepackage{lipsum}
\usepackage{enumitem}
\usepackage[colorinlistoftodos, textwidth=4cm, shadow, backgroundcolor=yellow]{todonotes}
\usepackage{subfigure}
\usepackage{chngpage}
\usepackage{calc}
\usepackage{pbox}
\usepackage{array}

\newtheorem{definition}{Definition}

% to squeeze space between items in a list
\newcommand{\compactlist}{
  \begin{list}{$\bullet$}{
    \setlength{\itemsep}{0pt}
    \setlength{\parsep}{3pt}
    \setlength{\topsep}{3pt}
    \setlength{\partopsep}{0pt}
    \setlength{\leftmargin}{1.5em}
    \setlength{\labelwidth}{1em}
    \setlength{\labelsep}{0.5em}
  }
}
\newcommand{\compactend}{\end{list}}

\usepackage{paralist}

\begin{document}

\title{S-Store: Streaming Meets Transaction Processing}

\numberofauthors{1}
\author{
  \alignauthor
    John Meehan$^{1}$, Nesime Tatbul$^{2,3}$, Stan Zdonik$^{1}$, Cansu Aslantas$^{1}$,  \\
    Ugur Cetintemel$^{1}$,Jiang Du$^{4}$,Tim Kraska$^{1}$,Samuel Madden$^{3}$,David Maier$^{5}$,\\
    Andrew Pavlo$^{6}$,Michael Stonebraker$^{3}$,Kristin Tufte$^{5}$,Hao Wang$^{3}$ \\
    ~ \\
  \affaddr{
    $^{1}$Brown University \hbox{~~~~}
    $^{2}$Intel Labs \hbox{~~~~}
    $^{3}$MIT \hbox{~~~~}
    $^{4}$UToronto \hbox{~~~~}
    $^{5}$Portland State University \hbox{~~~~}
    $^{6}$CMU
  }
}

\maketitle

\begin{abstract}

Stream processing addresses the needs of real-time applications. Transaction processing addresses the coordination and safety of short atomic computations. Heretofore, these two modes of operation existed in separate, stove-piped systems. In this work, we attempt to fuse the two computational paradigms in a single system called S-Store. In this way, S-Store can simultaneously accommodate OLTP and streaming applications. We present a simple transaction model for streams that integrates seamlessly with a traditional OLTP system. We chose to build S-Store as an extension of H-Store, an open-source, in-memory, distributed OLTP database system. By implementing S-Store in this way, we can make use of the transaction processing facilities that H-Store already supports, and we can concentrate on the additional implementation features that are needed to support streaming. Similar implementations could be done using other main-memory OLTP platforms. We show that we can actually achieve higher throughput for streaming workloads in S-Store than an equivalent deployment in H-Store alone. We also show how this can be achieved within H-Store with the addition of a modest amount of new functionality. Furthermore, we compare S-Store to two state-of-the-art streaming systems, Spark Streaming and Storm, and show how S-Store matches and sometimes exceeds their performance while providing stronger transactional guarantees.

\end{abstract}

\input{introduction}

\input{transactions}

\input{implementation}

\input{experiments}

\input{related}

\input{conclusion}

\balance

\small

\bibliographystyle{abbrv}
\bibliography{sstore}

\end{document}
related.tex
\section{Related Work} \label{sec:related}

% first-generation streaming systems

In the early 2000's, there was a lot of interest in the database community for stream processing. The main goal of this work was to process continuous queries with low latency as data streamed into the system. This was largely inspired by the emergence of sensor-based applications. Many academic prototypes (Aurora / Bore\-alis \cite{aurora, borealis}, STRE\-AM \cite{stream}, TelegraphCQ \cite{telegraphcq}, NiagraCQ \cite{niagracq}) were built, and several commercial products were spawned as a result of this work (e.g., TIBCO StreamBase, CISCO Truviso, SAP Coral8, IBM InfoSphere Streams, Microsoft StreamInsight, Oracle CEP). With the exception of STREAM and Coral8, these systems did not support an explicit notion of transactions. STREAM did not directly claim to have transactions, but its execution model was based on logical timestamps which could be interpreted as transaction IDs. Batches of tuples with the same timestamp were executed atomically. While this could be used to provide isolation, recovery was not discussed. Furthermore, modeling transactions as the execution of an entire workflow (query) did not allow finer-grained transaction definitions. Similarly, Coral8 provided so-called ''atomic bundles'' as a configurable isolation/recovery unit embedded in its execution model, but did not provide any transactional guarantees beyond "at least once" for processing events through the entire workflow. Furthermore, none of these early systems considered integrating stream processing with traditional OLTP-style query processing.

% transactions and streams

Fault tolerance issues have been investigated as stream processing systems have been moved into distributed settings \cite{borealis, telegraphcq}. A few fundamental models and algorithms have been established by this work \cite{hwang, balazinska, shah}, including the upstream backup technique that we leverage in our weak recovery mechanism.

There have also been several efforts in addressing specific transactional issues that arise in stream processing settings. For example, Golab et al. have studied the concurrency control problem that arises when a sliding window is advanced (write operation) while it is being accessed by a query (read operation) \cite{golab-edbt06}. This work proposes sub-windows to be used as atomic access units and two new isolation levels that are stronger than conflict serializability. Such a problem never arises in S-Store, since window state is accessed by a single TE at a time (and never by TEs of different SPs). As another example, Wang et al. have considered concurrency issues that arise when adding active rule support to CEP engines in order to monitor and react to streaming outputs \cite{wang-vldb11}. In this case, the rules may require accessing state shared with other queries or rules. This work defines a stream transaction as a sequence of system state changes that are triggered by a single input event, and proposes a timestamp-based notion of correctness enforced through appropriate scheduling algorithms. S-Store investigates transactional stream processing in a more general context than active CEP engines.

Botan et al.'s work was the first to recognize the need for an explicit transaction model to support queries across both streaming and stored data sources \cite{botan-edbt12}. This work proposed to extend the traditional page model \cite{weikum-book} to include streams of events (as time-varying relations) and continuous queries (as a series of one-time queries activated by event arrivals). As a result, each one-time query execution corresponds to a sequence of read/write operations, and operations from one or more such sequences can be grouped into transaction units as required by the application semantics. Transactions must then be executed in a way to ensure conflict serializability and event arrival ordering. Thus, this work focused on the correct ordering of individual read/write operations for a single continuous query, and not so much on transaction-level ordering for complex workflows like we do. As such, S-Store proposes a more general transaction model.

%Fernandez et al. have proposed a technique for integrated management of scale-out and fault-tolerance in public cloud infrastructures \cite{seep}. The key idea is to expose the state of streaming operators to the system such that it not only becomes externally visible to the system, but also manageable by the system in a generic fashion using a number of basic primitive scale-out and fault-tolerance operations such as checkpoint, backup, restore, and partition. This approach enables treating operator recovery as a special case of scale-out as well as recovering from failures in a parallel fashion.

% new-generation streaming systems

Recently, a new breed of stream processors has emerged. Unlike the majority of the earlier-generation systems, these do not adopt a select-project-join operator environment. Instead, they expect the user to supply their own operators (UDF's), and the system controls their execution in a scalable fashion over a cluster of compute nodes. Typically, these systems provide fault tolerance and recoverability, but do not support fully-ACID transactions. Representatives include Storm \cite{storm} and Spark Streaming \cite{zaharia13}.
  
Storm has been originally built to support Twitter's real-time computation needs. Like our workflows, computations are represented as {\em topologies}, i.e., directed graphs of operators called {\em bolts}.
%Strea\-aming tuples are shuffled using various partitioning strategies as they get processed through the bolts in a parallel fashion.
Storm provides two types of semantic guarantees: {\em at least once} and {\em at most once}. In case of {\em at least once} semantics, each tuple is assigned a unique message-id and its lineage is tracked. For each output tuple $t$ that is successfully delivered by a topology, a backflow mechanism is used to acknowledge the tasks that contributed to $t$ with the help of a dedicated {\em acker bolt} added to the corresponding topology. The data source must hold the tuple until a positive ack is received and the tuple can be removed (similar to upstream backup \cite{hwang}). If an ack is not received within a given timeout period, then the source will replay the tuple again. Storm can only provide the weaker {\em at most once} semantics when the acknowledgment mechanism is disabled. Trident provides a higher-level programming abstraction over Storm which provides a stronger, {\em exactly once} processing guarantee based on automatic replication \cite{trident-wiki}. While these guarantees ensure some level of consistency against failures, they are not sufficient to support atomicity and isolation as in the case of ACID guarantees. Furthermore, Storm focuses on purely streaming topologies and thus lacks support for dealing with persistent state and OLTP transactions.

Spark Streaming extends the Spark batch processing engine with a new stream processing model called discretized streams (D-Streams) \cite{zaharia13}. The basic idea is to divide (mostly analytical) computations into a series of stateless, deterministic transformations over small batches of input tuples based on their arrival time intervals. Somewhat similar to STREAM, tuples are processed atomically within the time interval that they arrive in. All state in Spark Streaming is stored in in-memory data structures called Resilient Distributed Datasets (RDDs). RDDs have two key properties: they are immutable and partitioned. Immutability enables efficient asynchronous checkpointing as well as maintaining different versions of state, which are tracked using lineage graphs. This kind of a computation model allows Spark Streaming to recover from failures in a parallel fashion without the need for full replication (except replicating new input and periodic checkpointing to reduce recovery time), and from stragglers by running speculative copies of slow tasks. The common RDD-based storage and processing model of the Spark engine also facilitates integrating streaming, batch, and interactive computations. Like Storm+Trident, Spark Streaming provides an {\em exactly once} consistency semantics, i.e., a much weaker form of guarantee than ACID. Furthermore, the RDD-based state management model incurs high overhead for transactional workloads that require many fine-grained update operations. First, since RDDs are immutable, each update requires creating a new RDD. Second, the lineage graph also gets bigger as each operation needs to be logged. In addition, the D-Stream model also has several limitations. For example, the time interval-based batching model hinders defining fine-grained isolation units (e.g., a single tuple) or tuple-based windowing operations.

%We also provide an experimental comparison of our S-Store system with both Storm and Spark Streaming in Section \ref{sec:spark}.

transactions.tex
\section{The Computational Model} \label{sec:model}

In this section, we describe our computational model for transactional stream processing. This model inherits traditional, well-accepted models of OLTP and stream processing, and fuses them within one coherent model which will then allow us to support hybrid workloads (independent OLTP transactions and streaming transactions) with well-defined transactional guarantees.
%In what follows, it should be noted that we use the terms {\emph transaction} and {\emph{stored procedure} interchangeably.

We now describe our overall abstract models of OLTP and streaming, and how we blended them together. Our goal is to design S-Store in such a way that ordinary OLTP transactions can operate concurrently with streaming transactions. We assume that both kinds of transactions can share state and at the same time produce correct results. We assume three different kinds of state: (i) public shared tables, (ii) windows, and (iii) streams. Furthermore, we make a distinction between {\em OLTP transactions} (only access (i)) and {\em streaming transactions} (access (iii) + optionally (ii) or (i)).
%Both types of transactions are predefined as parametric stored procedures, but they are instantiated rather differently. The former involve input values that are assigned by the application when the application explicitly invokes them (pull). The latter are invoked as new input data becomes available (push). 

For OLTP transactions, we simply adopt the traditional ACID model that has been well-described in the literature \cite{weikum-book}. A {\em database} consists of unordered, bounded collections (i.e., sets) of tuples. A {\em transaction} represents a finite unit of work (i.e., a finite sequence of read and write operations) performed over a given database. Each transaction has a definition and possibly many executions (i.e., instances). We assume that all transactions are predefined as {\em stored procedures} with input parameters, each of which can be instantiated with specific parameter values through explicit client invocations (``pull"). In order to maintain integrity of the database in the face of concurrent transaction executions and failures, each transaction is executed with {\em ACID guarantees}.

Streaming transactions are also predefined as parametric stored procedures, but they are instantiated rather differently, i.e. as new input data becomes available (``push"). We describe streaming transactions in more detail next.

\subsection{Streaming Transactions and Workflows} \label{sec:streaming}

Our stream data model is very similar to many of the stream processing systems of a decade ago \cite{aurora, stream, telegraphcq}. A {\em stream} is an ordered, unbounded collection of tuples. Tuples have a timestamp \cite{stream} or batch-id \cite{streamsql, secret} that specifies simultaneity. Tuples with the same timestamp or batch-id logically occur as a group at the same time and, thus, should be processed as a unit.

A {\em window} is a finite, contiguous subsequence of a stream. Windows can be defined in many different ways \cite{golab-sigrec03, secret}, but for the purposes of this work, we will restrict our focus to the most common type - sliding windows. A {\em sliding window} is a window which has a fixed size and a fixed slide, where the slide specifies the distance between two consecutive windows and can be less than or equal to size (called a {\em tumbling window} if the latter). A sliding window is said to be {\em time-based} if its size and slide are defined in terms of tuple timestamps, and {\em tuple-based} if its size and slide are defined in terms of the number of tuples.

Stream processing systems commonly define computations over streams as workflows. Early streaming systems focused on relational-style operators as computations (e.g., Filter, Join), whereas current systems support more general user-defined computations \cite{storm, zaharia13}. Following this trend and consistent with our OLTP model, we assume that computations over streams are expressed as workflows of user-defined stored procedures. More formally, a {\em workflow} is a directed acyclic graph (DAG), in which nodes represent streaming transactions and edges represent an execution ordering. If there is an edge between node $N_i$ and node $N_j$, there is also a stream that is output for $N_i$ and input for $N_j$. We say $N_i$ precedes $N_j$ and is denoted as $N_i \prec N_j$.

Given the unbounded nature of streams, we restrict streaming transaction instances to operate over non-overlapping ``atomic batch\-es" of stream tuples. An {\em atomic batch} is a finite, contiguous subsequence of a stream that must be processed atomically  (e.g., tuples with the same timestamp, from a common time interval, that belong to the same sales order, etc.).

Contiguous atomic batches of tuples arrive on a stream at the input to a workflow from push-based data sources. We adopt the data-driven execution model of streams, where arrival of a new atomic batch causes a new invocation for all the streaming transactions that are defined over the corresponding stream. This is in contrast to the on-demand execution model of OLTP transactions which are invoked by explicit requests from clients.

%We require transactions to be pre-defined as stored procedures (SPs) with input parameters in the context of a streaming workflow. They are defined once, but they are executed many times, once for each input batch.
Streaming transactions are defined as stored procedures once, but they are executed many times, once for each input batch. We refer to each such execution of a transaction definition as a {\em transaction execution} (TE). A TE essentially corresponds to an atomic batch and its subsequent processing by a stored procedure. For example, in Figure \ref{f:workflow}, a workflow with two stored procedures are defined, but each of those are executed twice for two contiguous atomic batches, yielding a total of four transaction executions.

Given a workflow, it is also useful to distinguish between {\em border transactions} (those that ingest streams from the outside, e.g., $SP_1$ in Figure \ref{f:workflow}) and {\em interior stored procedures} (others, e.g., $SP_2$ in Figure \ref{f:workflow}). Border stored procedures are instantiated by each newly arriving atomic batch, and each such execution may produce a group of output stream tuples labeled with the same batch-id as the input that produced them. These output tuples become the atomic batch for the immediately downstream interior stored procedures, and so on.

%We assume that for a given workflow, any of the topological orderings for its graph gives a correct execution ordering for that workflow's stored procedures. Enforcing a particular serial ordering was not needed in earlier streaming systems, but it can be critical in a transactional setting in which the commit of one transaction triggers the start of one or more other transactions, and table state might be shared among streaming and OLTP transactions.

\begin{figure}[t]
\centering
\includegraphics[width=0.8\columnwidth]{images/TE-image.png}
\caption{Transaction Executions in a Workflow}
\vspace{-0.2in}
\label{f:workflow}
\end{figure}

In the rest of this paper, we use the terms ``transaction'' and ``stored procedure'' interchangeably to refer to the definition of a transaction, whereas ``transaction execution'' to refer to a specific instance of that definition.

\subsection{Correct Execution} \label{sec:isolation}

A standard OLTP transaction mechanism guarantees the isolation of one transaction's operations from other transactions'. When a transaction $T$ commits successfully, all of $T$'s writes are installed and made public. During the execution of $T$, all of $T$'s writes remain private.

S-Store adopts such standard transaction semantics as a basic building block for its streaming transactions; however, the ordering of stored procedures in the workflow definition as well as the inherent order in streaming data puts additional constraints on allowable transaction execution orders. As an example, consider the workflow shown in Figure \ref{f:workflow}, where $SP_1$ precedes $SP_2$. We use $TE_{i,j}$ to indicate the $j$th transaction execution of stored procedure $SP_i$. The figure illustrates four transaction executions that correspond to the processing of two atomic input batches (batch1 and batch2) through the complete workflow. These TE's can be ordered in one of two possible ways:
$[TE_{1,1}, TE_{2,1}, TE_{1,2}, TE_{2,2}]$ or $[TE_{1,1}, TE_{1,2}, TE_{2,1}, TE_{2,2}]$. Any other orderings would not lead to a correct execution. This is in contrast to most OLTP transaction processors which would accept any serializable schedule (e.g., one that is equivalent to any of the 4! possible serial execution schedules if these were 4 independent OLTP transactions).

Note that we make no ACID claims for the workflow as a whole. The result of running a workflow is to create an ordered execution of ACID transactions.

\begin{comment}

The following two rules must hold on any transaction execution ordering:

\begin{enumerate}
  \item $TE_{i,k} \prec TE_{j,k},  \forall i < j$
  \item $TE_{i,j} \prec TE_{i,k},  \forall j < k$
\end{enumerate}

The scheduler is free to pick any transaction execution ordering that conforms to these rules. This is in contrast to most OLTP transaction processors which are free to choose any ordering as long as the result is serializable (i.e., equivalent to any of the serial execution schedules).

In the example of Figure \ref{f:workflow}, the first rule ensures that for a given input batch, $SP_1$ is executed before $SP_2$, and the second rule ensures that a given stored procedure executes its atomic input batches in their arrival order. Thus, two legal orderings for this example would be: \\

$TE_{1,1}; TE_{2,1}; TE_{1,2}; TE_{2,2}$ , or \\

$TE_{1,1}; TE_{1,2}; TE_{2,1}; TE_{2,2}$ \\

\end{comment}

Furthermore, in streaming applications, the state of a window must be shared differently than other stored state. To understand this, consider again the simple workflow shown in Figure \ref{f:workflow}. Let us assume for simplicity that the transaction input batch size for $SP_1$ is one tuple. Further, suppose that $SP_1$ constructs a window of size 2 that slides by 1 tuple, i.e., two consecutive windows in $SP_1$ overlap by 1 tuple. This means that window state will carry over from $TE_{1,1}$ to $TE_{1,2}$. For correct behavior, this window state must not be publicly shared with other TE's. That is, the state of a window can be shared among transaction executions of a given stored procedure, but should not be made public beyond that. Returning to Figure \ref{f:workflow}, when $TE_{1,1}$ commits, the window in $TE_{1,1}$ will slide by one and will then be available to $TE_{1,2}$, but not to $TE_{2,1}$. This approach to window visibility is necessary because of the nature of streaming applications. In some sense, it is this way of sharing window state that is the basis for continuous operation. Windows evolve and, in some sense, ``belong'' to a particular stored procedure. Thus, a window's visibility should be restricted to the transaction executions of its owning stored procedure.

We will now describe what constitutes a {\em correct execution} for a workflow of streaming transactions more formally. Consider a workflow $W$ of $n$ streaming transactions $T_i$, $1 \leq i \leq n$. $W$ is a directed acyclic graph $G=(V,E)$, where $V=\{T_1, \ldots, T_n\}$ and $E \subseteq V \times V$, where $(T_i, T_j) \in E$ means that $T_i$ must precede $T_j$ (denoted as $T_i \prec T_j$). Being a DAG, $G$ has at least one topological ordering. A topological ordering of $G$ is an ordering of its nodes $T_i \in V$ such that for every edge $(T_i, T_j) \in E$ we have $i < j$. Each topological ordering of $G$ is essentially some permutation of $V$.

Without loss of generality:
(i) Let us focus on one specific topological ordering of $G$ and call it $O$;
(ii) For ease of notation, let us simply assume that $O$ corresponds to the identity permutation such that it represents: $T_1 \prec T_2 \prec .. \prec T_n$.

$T_i$ represents a transaction definition $T_i(s_i, w_i, p_i)$, where
$s_i$ denotes all private stream inputs of $T_i$ (at least one),
$w_i$ denotes all private window inputs of $T_i$ (optional),
$p_i$ denotes all public table partition inputs of $T_i$ (optional).
Similarly, $T_i^j$ represents the $j^{\text{th}}$ transaction execution of $T_i$ as $T_i^j(s_i.b_j, w_i, p_i)$, where $b_j$ denotes the $j^{\text{th}}$ atomic batches of all streams in $s_i$.

A workflow $W$ is executed in rounds of atomic batches $1 \leq r < \infty$, such that for any round $r$, atomic batch $r$ from all streaming inputs into $W$ generates a sequence of TE's $T_i^r(s_i.b_r, w_i, p_i)$ for each $T_i$.
Note that this execution generates an {\em unbounded schedule}. However, as of a specific round $r=R$, we generate a {\em bounded schedule} that consists of all $R*n$ TE's: $1 \leq r \leq R, 1 \leq i \leq n, T_i^r(s_i.b_r, w_i, p_i)$.

In the traditional ACID model of databases, any permutation of these $R*n$ TE's would be considered to be a valid/correct, serial schedule. In our model, we additionally have:

\begin{compactenum}

\item {\em Workflow order constraint:}
Consider the topological ordering $O$ of $W$ as we defined above. Then for any given execution round $r$, it must hold that:

$T_1^r(s_1.b_r, w_1, p_1) \prec \ldots \prec T_n^r(s_n.b_r, w_n, p_n)$ 

\item {\em Stream order constraint:}
For any given transaction $T_i$, as of any round $r=R$, the following must hold:

$T_i^1(s_i.b_1, w_i, p_i) \prec \ldots \prec T_i^R(s_i.b_R, w_i, p_i)$

\end{compactenum}

Any bounded schedule of $W$ that meets the above two ordering constraints is a {\em correct schedule}.

If $G$ has multiple topological orderings, then the workflow order constaint must be relaxed to accept any of those orderings for any given execution round of $W$.

%Note that workflows are not ACID, but their components are.

\begin{comment}

Expanding on this idea a bit, each transaction T in a workflow has two implicit arguments, one for the window that may get passed to it by the corresponding earlier TE of T, and a second argument that corresponds to the batch of tuples that are consumed on the input to T. Thus, when we write $TE_{i,j}$, we really mean $TE_{i,j}(W, S)$. Rewriting the first order for Figure \ref{f:workflow} given above in this expanded form gives us:

\begin{center}
$TE_{1,1}(\Phi_W, Batch_1); TE_{2,1} (\Phi_W, S_1);$\\
$TE_{1,2}(W_1, Batch_2); TE_{2,2}(W_2, S_2)$
\end{center}

where $Batch_1$ and $Batch_2$ are two consecutive tuple batches that are consumed by $T_1$ at it's input. $\Phi_W$ is an empty window that initializes the sequence of TEs that are generated by some transaction definition, and $S_1$ and $S_2$ are streams that connect the various TEs.

Note that we make no ACID claims for the workflow as a whole. The result of running a workflow is to create an ordered execution of ACID transactions. The above sequence is certainly that. It includes two parameters that are not public and that are simply inputs to the stored procedure (transaction), like any other.

\end{comment}

\subsection{Hybrid Transaction Schedules} \label{sec:hybrid}

S-Store's computational model allows OLTP and streaming transactions to co-exist as part of a common transaction execution schedule. This is particularly interesting if those transactions access public shared tables.

Given previous section's formal description of a correct schedule for a workflow $W$ that consists of streaming transactions, any OLTP transaction execution $T_i^j(p_i)$ (defined on one or more public table partitions $p_i$) is allowed to interleave {\em anywhere} in such a schedule. The resulting schedule would still be correct.

\begin{figure}[t]
\centering
\includegraphics[width=0.4\columnwidth]{images/OLTPTransaction.png}
\caption{Nested Transaction Example}
\vspace{-0.2in}
\label{f:OLTP}
\end{figure}

We have also extended our transaction model to include {\em nested transactions}. Fundamentally, this allows the application programmer to build higher-level transactions out of smaller ones, giving him/her the ability to create coarser isolation units among stored procedures, as illustrated in Figure \ref{f:OLTP}. In this example, two streaming transactions, $SP_1$ and $SP_2$, in a workflow access a shared table $T$. $SP_1$ writes to the table and $SP_2$ reads from it. If another OLTP transaction also writes to $T$ in a way to interleave between $SP_1$ and $SP_2$, then $SP_2$ may get unexpected results. Creating a nested transaction with $SP_1$ and $SP_2$ as its children will isolate the behavior of $SP_1$ and $SP_2$ as a group from other transactions (i.e., other OLTP or streaming). Note that nested transactions can also be useful for isolating multiple instances of a given streaming workflow from one another. For example, defining our leaderboard maintenance workflow described in Section \ref{sec:example} as a nested transaction would facilitate enforcing the correct execution ordering.

More generally, an S-Store nested transaction consists of two or more stored procedures with a partial order defined among them \cite{db-book}. The stored procedures within a nested transaction must execute in a way that is consistent with that partial order.
%(in a similar way that stored procedures are allowed to execute in a particular order in our streaming workflows as discussed in Section \ref{sec:isolation}).
A nested transaction will commit, if and only if all of its stored procedures commit. If one or more stored procedures abort, the whole nested transaction will abort.

Nested transactions fit into our formal model of streaming transactions in a rather straight-forward way. More specifically, any streaming transaction $T_i$ in workflow $W$ can be defined as a nested transaction that consists of children $T_{i1}, \ldots, T_{im}$. In this case, $T_{i1}, \ldots, T_{im}$ must obey the partial order defined for $T_i$ for every execution round $r$, $1 \leq r < \infty$. This means that no other streaming or OLTP transaction instance will be allowed to interleave with $T_{i1}, \ldots, T_{im}$ for any given execution round.

\subsection{Fault Tolerance} \label{sec:fault}

In the face of failure, S-Store must recover its state such that any committed transactions remain stable and at the same time, any uncommitted transactions are not allowed to have any effect on the publicly shared tables. For a streaming workflow, if $TE_{i,j}$ precedes $TE_{m,n}$, then $TE_{i,j}$ must commit before $TE_{m,n}$. Also, if $TE_{i,j}$ precedes $TE_{i,j+1}$ and they share a window $w$, $w$ must be restored to the state that it had before $TE_{i,j+1}$ began executing. A TE that had started, but had not committed should be undone and it should be restarted with the proper stream input from its predecessor.

S-Store strives to produce {\em exactly-once} executions which is to say that the output will not contain any duplicates, but will produce all output tuples that would have been produced if there were no failures (no lost results). However, if the workflow definition allows multiple transaction execution orderings or if the transactions within a workflow contain any non-deterministic operations, we provide an additional recovery option that we call {\em weak recovery}. Weak recovery will produce a correct result in the sense that it will produce results that could have been produced before the failure, but not necessarily the one that was in fact being produced.

We also are willing to accept a mechanism that is slow on recovery if it is fast during normal operation. This is based on the observation that we expect failures to be rare. In the next section, we will explain the recovery mechanisms that S-Store uses to implement this fault tolerance model.

conclusions.tex
\section{Conclusions} \label{s:conclusions}
introduction.tex
\section{Introduction} \label{s:introduction}

Managing high-speed data streams generated in real time is an integral part of
today’s big data applications. In a wide range of domains from social media to 
financial trading, there is a growing need to seamlessly support incremental
processing as new data is generated. At the same time, the system must ingest
some or all of this data into a persistent store for on-demand transaction or
analytical processing. In particular, for applications with high-velocity data
updating a large persistent state, there is need to manage streaming and
non-streaming state side by side, in a way to ensure transactional integrity
and performance. Examples include leader-board maintenance, multi-player
online games, real-time financial trading, and online advertising.

It has been more than a decade since the first stream processing engines
were built. As these research prototypes moved into the commercial world,
the customer applications that they were used for were rarely pure stream
processing. They often required storage as a way to record the past for
comparison with the present. Thus, the obvious solution was to bolt on a
DBMS, but the solutions were very unsatisfying because they ignored the
need for consistency between streaming results and stored state.

Today’s stream processing systems (stream databases, CEP engines,
distributed real-time computation frameworks such as \cite{streambase},
\cite{oracle-cep}, \cite{storm}) still lack the required transactional
robustness, while OLTP databases do not provide native support for
data-driven processing. In this work, our goal is to fill this gap by
building a single system that can scale to support stream and transaction
processing under the same hood. We believe that modern distributed main-memory
OLTP platforms such as H-Store \cite{hstore} provide a suitable
foundation for building such a system, since: first, they are more
light-weight than their traditional disk-based counterparts; second, like 
streaming engines, they offer lower latency via in-memory processing;
and last but
not least, they provide strong support for state and transaction management.
Thus, we introduce S-Store, a streaming OLTP system that realizes our goal
by extending H-Store with streams.

\subsection{Running Example}

\todo[inline]{
Describe the voter example.
}

\subsection{Contributions}

\todo[inline]{
Summarize contributions and outline of the paper.
}
paper.tex
\documentclass{vldb}
\usepackage{graphicx}
\usepackage{balance}
\usepackage{url}
\usepackage{comment}
\usepackage{enumerate}
\usepackage[colorinlistoftodos, textwidth=4cm, shadow, backgroundcolor=yellow]{todonotes}
\usepackage{subfigure}

% to squeeze space between items in a list
\newcommand{\compactlist}{
\begin{list}{$\bullet$}{
\setlength{\itemsep}{0pt}
\setlength{\parsep}{3pt}
\setlength{\topsep}{3pt}
\setlength{\partopsep}{0pt}
\setlength{\leftmargin}{1.5em}
\setlength{\labelwidth}{1em}
\setlength{\labelsep}{0.5em}
}
}
\newcommand{\compactend}{\end{list}}

\begin{document}

\title{S-Store: A Streaming OLTP System for \\ Big Velocity Applications}

\numberofauthors{1}
\author{
\alignauthor
Nesime Tatbul$^{1,2}$, Ugur Cetintemel$^{3}$, Daehyun Kim$^{1}$,
Tim Kraska$^{3}$, Sam Madden$^{2}$, \\ John Meehan$^{3}$, Andy Pavlo$^{4}$,
Mike Stonebraker$^{2}$, Hawk Wang$^{2}$, Stan Zdonik$^{3}$,
... \\
~ \\
\affaddr{
$^{1}$Intel Labs \hbox{~~~~}
$^{2}$MIT \hbox{~~~~}
$^{3}$Brown \hbox{~~~~}
$^{4}$CMU
}
}

\maketitle

\begin{abstract}
\end{abstract}

\listoftodos

\input{introduction}

\input{architecture}

\input{transactions}

\input{implementation}

\input{experiments}

\input{related}

\input{challenges}

\input{conclusions}

\balance

\bibliographystyle{abbrv}
\bibliography{sstore}

\end{document}
architecture.tex
\section{Architectural Overview} \label{s:architecture}

\todo[inline]{
We may want to detail the H-Store description if needed.
}

S-Store extends the architecture of H-Store \cite{hstore}. In a nutshell,
H-Store is a high-performance, in-memory, distributed OLTP system designed
for shared-nothing clusters. It targets OLTP workloads with short-lived
transactions, which are pre-defined as parametric stored procedures
(i.e., SQL queries embedded in Java programs) that get instantiated by client
requests at run time. Data is carefully partitioned and transactions that
can be executed on a single partition are run serially on that partition,
eliminating the need for locks and latches. H-Store's database designer
strives to minimize the number of distributed transactions to reduce the
overhead of coordination across the partitions \cite{pavlo12}.
Fault tolerance is achieved based on a technique that combines command logging
and periodic snapshots \cite{malviya14}.

S-Store builds on H-Store’s architecture, reusing its core data management
primitives such as tables and stored procedures. In addition, it introduces
a number of new primitives for supporting stream processing natively.
S-Store’s key extensions include:

\compactlist

\item {\bf New constructs for streams, windows, triggers, and workflows:}
Streams differentiate continuously streaming state from regular stored state;
windows define finite chunks of state over (possibly unbounded) streams;
triggers are used to indicate computations to be invoked for newly generated
data; and workflows refer to pipelines of dependent transactions triggered
by common streaming input(s).

\item {\bf Stream-oriented transaction model:}
We define a streaming transaction in S-Store as a stored procedure with a
streaming input that is invoked by newly arriving stream tuple(s) on that
input.
%Thus, streaming transactions are data-driven, whereas regular OLTP
%transactions are operation-driven.

\item {\bf Uniform state management:}
H-Store’s in-memory tables are used for representing all states including
streams and windows, making state access both efficient and
transactionally safe. Furthermore, S-Store provides automatic garbage
collection mechanisms for tuples that expire from stream or window state.

\item {\bf Data-driven processing via triggers:}
Special insert triggers are defined on stream or window state in order to
enable push-based, data-driven processing in S-Store. There are two types of
triggers: engine triggers at the query level (SQL) and front-end triggers
at the stored procedure level (Java). The former enable continuous processing
for a given transaction, while the latter enable composing workflows of
transactions.

\compactend

\missingfigure[figwidth=\columnwidth]{
Need an architecture figure with all important components,
showing the ones added by S-Store.
}

As shown in Figure \ref{f:3layers},
S-Store follows H-Store's three-layer transaction execution stack
that consists of the client application, the partition engine,
and the execution engine. The native stream processing extensions described
above help S-Store use this stack in the most efficient way by embedding
the required functionality in the proper engine layers, thereby avoiding
redundant computations, communication across the layers, and the need to
poll for new data.
\todo[inline]{Detail the comparison.}

\begin{figure*}[t]
\begin{center}
\subfigure[H-Store]{
    \includegraphics[width=0.45\textwidth]{3layers-hstore.png}
    \label{f:3layers-hstore}
}
\subfigure[H-Store]{
    \includegraphics[width=0.48\textwidth]{3layers-sstore.png}
    \label{f:3layers-sstore}
}
\caption{Three Layers of Transaction Execution}
\label{f:3layers}
\end{center}
\end{figure*}

Figure \ref{f:voterwinsstore} shows how our running example would be
implemented with S-Store.
\todo[inline]{Describe the figure.}

\begin{figure}[t]
\centering
\includegraphics[width=\columnwidth]{voterwinsstore.png}
\caption{Streaming Voter with S-Store}
\label{f:voterwinsstore}
\end{figure}

Although discussed within the context of the H-Store OLTP engine in this paper,
we believe that the architectural additions described above generally apply
when extending any in-memory OLTP engine with streams.
The two-layer engine is the only H-Store-specific part of our design,
which essentially leads to having two levels of triggers in the engine.
We believe that all of our extensions are conceptually needed in any OLTP
engine and trigger is the key construct for data-driven processing
and to avoid the need for polling and redundant communication,
even in a single-layer engine.
\todo[inline]{Incorporate Andy's input:}
- Main reasons for the two-layered design: It's easier for customers to write
stored procedures in Java, while it's better to keep all of the storage +
memory management in C++ and avoid the GC.

- PE: Anything dealing with the network, query planning, scheduling,
txn management, stored procedures.

- EE: Low-level query operator execution, storage, indexes.

- H-Store vs. other main-memory OLTP:
The coarse-grained partition locks, single-threaded execution engines,
logical logging, and being optimized only for stored procedures.
implementation.tex
\section{Implementation} \label{s:implementation}
challenges.tex
\section{Research Challenges} \label{s:challenges}

\todo[inline]{
This part needs to be rewritten.
}

Building an integrated system for stream and transaction system such as
S-Store raises a number of interesting research challenges. Following is
a list of problems that are in our current research agenda:

\compactlist

\item Workload partitioning:
S-Store takes its foundation from a distributed main-memory OLTP system
that achieves high transactional performance by carefully partitioning
its data and transaction workload onto a cluster of nodes, each of which
itself is running one or more partition engine instances with its own
shard of data. This is achieved using an automatic designer that decides on
the data partitions in a way to minimize the number of distributed
transactions required for a given type of workload.
In the presence of streaming inputs and complex workflows of transactions
that execute in a push-based manner, the designer needs to additionally
decide how to partition these streaming inputs as well as
where to run different transactions that comprise a given workflow.
We have been working on a new cost model to guide the designer in making
these decisions given the new forms of parallelism opportunities
and data dependency constraints in S-Store workloads.

\item Transaction management:
Our research in this area is focusing on two key questions:
when do we need transactional semantics for streams and how
do we define a transaction when streaming data is involved?
S-Store currently follows a basic stream-oriented transaction model
that uses newly arriving tuples to trigger transaction requests and
to determine transactional boundaries on continuous data. We have been
working on generalizing this model in a way to support a variety of
application semantics in a flexible and performant way.

\item Handling mixed and dynamic workloads:
S-Store can potentially support mixed workloads with regular OLTP
transactions, streaming OLTP transactions, as well as adhoc transactions.
Furthermore, workload conditions can change over time.
It is an interesting question how to prioritize the scheduling of such
mixed and complex workloads as well as support more elastic workload
partitioning under dynamic settings.

\item Performance analysis and hardware optimization:
Like H-Store, S-Store has great potential to optimize its transactional
performance further by leveraging modern hardware architectures
(e.g., multiple cores, non-volatile memory). Our future work includes
performing a rigorous software performance analysis of S-Store in
a way to identify bottlenecks and fix them exploiting modern hardware.

\compactend
experiments.tex
\section{Experiments} \label{s:experiments}

\subsection{Goals and Setup}

\subsection{Streaming TPC-E}

- goals: to show that we can support complex/realistic benchmarks;
to show performance benefits over H-Store.

- describe Streaming TPC-E.

- describe H-Store and S-Store implementations.

- results with H-Store and S-Store.

\subsection{Streaming Voter}

- goals: to show performance benefits over H-store on a simpler benchmark
(paper's running example);
to compare against Spark-Streaming and Twitter-Storm
(with different transactional guarantees).

- describe Spark-Streaming and Twitter-Storm implementations.

- results with H-Store, S-Store, Spark-Streaming (exactly once),
Twitter-Storm (at least once), Twitter-Storm+Trident (exactly once)

- if feasible, results with Spark and Twitter giving equivalent guarantees
as S-Store.

\subsection{Streaming WordCount}

- goals: to compare against Spark-Streaming and Twitter-Storm on one of their
own benchmarks; both on single and multiple partitions.

- describe Streaming WordCount.

- describe implementations on S-Store, Spark, Storm.

- results (single-node + multi-node).

\subsection{Discussion}

\begin{comment}

\section{Preliminary Results} \label{s:experiments}

\begin{figure}
\centering
\includegraphics[width=3.0in]{svoter2}
\caption{Streaming Voter Benchmark}
\label{f:svoter2}
\end{figure}

We have compared S-Store's transaction throughput performance against
H-Store's on a streaming version of the Voter Benchmark \cite{benchmarks}.
This benchmark originally models a leader-board application where
a number of contestants participate in a talent show and users call in to
vote on their favorite contestants. Each phone call invokes a new transaction
that updates the total number of votes for each contestant after a number of
validity checks according to the voting rules. Valid votes must be permanently
recorded in the system as well as updating a number of statistical views
over these (e.g., total number of votes per contestant, per state,
and per voting phone number; top-k contestants, etc.).
We have added two streaming extensions to this benchmark:
a stream of votes as input and a sliding window aggregate computation.
We have then implemented this streaming version with S-Store as well as
with H-Store. The S-Store implementation uses the new stream processing
constructs embedded into the execution engine layer such as streams,
windows, engine triggers, whereas the H-Store implementation simulates the
same functionality using the available constructs of H-Store.
Figure \ref{f:svoter2} shows how each of these variants perform for a
tuple-based window of size 100 as the window slide varies
between 1 and 100 (tumbling window). In all of these experiments, S-Store
outperforms H-Store by a factor of about 50\%. The performance gap slightly
decreases as slide increases, since in this case, H-Store has to make
fewer additional computations and calls to the execution engine for
simulating the equivalent windowing behavior.

We are currently extending our experimental study to benchmark S-Store
against H-Store and other alternative stream processing system (e.g.,
\cite{storm}, \cite{zaharia13-short}) using more complex benchmarks
such as a streaming version of the TCP-E OLTP benchmark \cite{tpc-e}.

\end{comment}
transactions.tex
\section{S-Store Transaction Model} \label{s:transactions}

\subsection{Overview}

In S-Store, all {\em state} is kept in in-memory tables like in H-Store.
In addition to regular tables, one can also define tables to represent
stream state and window state.

S-Store supports regular {\em OLTP transactions} (that only involve access to
regular tables) as well as {\em streaming transactions} (that involve access to
streams or windows along with regular tables).
OLTP transactions are pre-defined as parametric stored procedures
(i.e., SQL queries embedded in Java programs) that get instantiated
by client requests at run time.
Streaming transactions are defined in the same manner except that they get
instantiated by newly arriving data on streaming inputs rather than by client
transaction requests. In other words, streaming transactions are
{\em data-driven} while OLTP transactions are operation-driven.
Furthermore, S-Store allows multiple streaming transactions to be composed
into a {\em transactional workflow}.

\subsection{Streaming Transactions}

A stored procedure that defines a streaming transaction involves a sequence of
SQL commands that get triggered by new tuple insertions into stream or window
state. Each new ``batch" of tuples inserted into the input stream of this
stored procedure instantiates a new transaction that executes this sequence.
A {\em batch} is a sequence of stream tuples that logically belong together
to be processed as an atomic unit and can be defined in an application-specific
way. For example, tuples with a common timestamp might be specified as a batch,
or tuples that include sales orders from a common business transaction might be
specified as a batch.

\begin{figure}[t]
\centering
\includegraphics[width=\columnwidth]{txn-example.png}
\caption{Streaming transaction example}
\label{f:txn-example}
\end{figure}

The streaming transaction shown in Figure \ref{f:txn-example} selects
tuples with values above 10 from a stream, constructs a window out
of them (assume a window defined as a tuple-based tumbling window of size 2),
computes an average of the values in each window, and then the results are
written to a table. More specifically, input batch (a,b) initiates a
transaction instance T1, where:

\begin{enumerate}[(i)]

\setlength{\itemsep}{0pt}
\setlength{\parsep}{3pt}
\setlength{\topsep}{3pt}
\setlength{\partopsep}{0pt}
\setlength{\leftmargin}{1.5em}
\setlength{\labelwidth}{1em}
\setlength{\labelsep}{0.5em}

\item the first SQL statement writes (a,b) into stream S,

\item insert trigger defined on S fires the second SQL statement that selects
tuples with values above 10 (say, (a) qualifies) and writes them into stream
S',

\item insert trigger defined on S' fires the third SQL statement writing
(a) into window W,

\item window insert trigger defined on W fires the last SQL statement if there
is a full window, which then computes an average for that window and writes
the result into table T.

\end{enumerate}

T1 {\em commits} when there are no more SQL statements or triggers to execute.
Then the system can initiate the next transaction instance T2 for the next
input batch (c,d,e), and so on. This entails that subsequent instances of a
streaming transaction executed on a given node are inherently serial.

Note that while each transaction instance is executed {\em atomically} with
respect to the input batches that triggers it, not all SQL statements are
necessarily executed in every instance. For example, T1 will only execute up
to step (iv) since there is no full window to trigger the last statement yet.
A similar situation could arise if none of the tuples in a given batch
qualified the filter predicate in step (ii), committing the transaction
instance right after (ii).

Stream state is garbage collected after all relevant triggers are executed.
So is window state. While no stream state is left behind after a transaction
instance commits, this is not necessarily the case for window state.
For example, T1 will insert (a) into W, but the window insert trigger on W
will not yet fire, i.e., (a) has to be kept in window state W. This means
that state from a transaction instance (T1) can carry over to the subsequent
ones (T2). Furthermore, for correct execution of the streaming transaction,
such state must be protected from {\em side effects} of other transactions.
For example, another OLTP or streaming transaction should not be able to see
or modify any state kept in W between subsequent instantiations of the
streaming transaction (i.e., after T1 commits and before T2 starts).
Such a restriction is not needed for regular table state, e.g., another OLTP
transaction can read/modify table T without causing any side effects on our
streaming transaction's instances.

To protect state that carries over across subsequent instances of a streaming
transaction, we use a {\em nested transaction model}. All such instances are
treated as if they were subtransactions of a common transaction. Any state
that should only be visible to them must be explicitly specified
(W in our example). All other state is treated in the usual way.

Note that subtransactions execute in a serial fashion on a given node and
their order of execution is determined by the input batches, i.e., no
{\em concurrency issues} arise among these subtransactions. On the other hand,
an interesting concurrency scenario might arise within a given subtransaction,
if there were multiple triggers defined on a common stream or window state,
firing at the same time and leading to multiple SQL statements downstream
that might have conflicting access to shared state. In S-Store, we process
the triggers of a transaction instance in a pre-defined order (depth-first)
and therefore this scenario never occurs.

Streaming transactions offer interesting opportunities for
{\em speculative execution}. For example, the batch-based definition of a
streaming transaction leads to the question whether a transaction instance
should be scheduled for execution as soon as the first element of a new
batch arrives (a) or after the whole batch has arrived (a,b).
The first approach may block the execution engine if the input is slow and
the rest of the batch does not arrive right away for the transaction instance
to commit (long-running transaction). On the other hand, the second approach
does not allow making progress with the new tuples as soon as they arrive.
If there is no regular table state (externally visible) involved in a
streaming transaction, then we can safely take the first approach and
execute the transaction instance exploiting the fact that all state is
anyway protected in this case due to our nested transaction model.
On the other hand, if regular table state is involved, we can also
speculatively execute up until the step that involves access to the
table state. In this case, we would only complete the rest of the
execution after we have received the whole batch. In the meantime,
other transactions would be allowed to access the table state.
We currently implement the second approach in S-Store and plan to
investigate speculative transaction execution opportunities
as part of our future work.

\subsection{Workflows of Streaming Transactions}

In S-Store, multiple streaming transactions can be composed into a
transactional workflow. In this case, additional data dependency scenarios
arise across instances of different transactions that are part of the same
workflow (in addition to the ones across instances of the same transaction
as we have discussed in the previous section). We can capture them using
another level of transaction nesting in our model.

\begin{figure}[t]
\centering
\includegraphics[width=0.5\columnwidth]{workflow-example.png}
\caption{Streaming workflow example}
\label{f:workflow-example}
\end{figure}

For example,
the workflow in Figure \ref{f:workflow-example} consists of three streaming
transactions X, Y, and Z. $S_{xy}$ represents the stream that flows from
X to Y, and $S_{yz}$ represents the stream that flows from Y to Z.
Furthermore, X and Y both access a common regular table T.
We consider X, Y, and Z to be subtransactions of a common transaction for the
whole workflow. This way we can protect states $S_{xy}$ and $S_{yz}$ from the
access of all OLTP and streaming transactions that are external to this
workflow.
Assume that the whole workflow runs on the same node, i.e., instances
of X, Y, and Z do not execute in parallel. While instances of each transaction
execute serially among themselves ($X1 \rightarrow X2 \rightarrow X3;
Y1 \rightarrow Y2 \rightarrow Y3; Z1 \rightarrow Z2 \rightarrow Z3$),
they can interleave in multiple different ways (though not completely
arbitrarily) across instances of different transactions.
For example, the following would both be possible orderings:

(i) $X1 \rightarrow Y1 \rightarrow Z1 \rightarrow X2 \rightarrow Y2 \rightarrow
Z2 \rightarrow X3 \rightarrow Y3 \rightarrow Z3$

(ii) $X1 \rightarrow X2 \rightarrow Y1 \rightarrow X3 \rightarrow Z1
\rightarrow Y2 \rightarrow Y3 \rightarrow Z2 \rightarrow Z3$

The question is whether all orderings are ``serializable" with respect to
the data dependencies within the workflow. If there is no shared non-streaming
state like T, this would certainly be the case. Otherwise, depending on the
nature of how T is shared across X and Y, we may want to disallow orderings
like in (ii), since it may be that effects of processing batch (a,b)
through X1 must be seen by Y1 first, before X2 overwrites them
by processing batch (c,d,e)...
related.tex
\section{Related Work} \label{s:related}

Botan et al. have proposed to extend the traditional model to include 
streaming data and continuous queries based on a unified transaction
model \cite{botan-edbt12}... logical model, tested on a stream
processing system, stream-oriented transactions, but no nested transactions...

Zaharia et al.'s batch-based model and exactly-once consistency guarantee
\cite{zaharia13}...

Google MillWheel \cite{millwheel}...

Twitter Storm/Trident, at-least-once or exactly-once, different programming
model \cite{storm}...
main.bbl
\begin{thebibliography}{10}

\bibitem{trident-wiki}
{Trident Tutorial}.
\newblock \url{https://storm.apache.org/documentation/Trident-tutorial.html}.

\bibitem{voltdb}
{VoltDB}.
\newblock \url{http://www.voltdb.com/}.

\bibitem{aurora}
D.~Abadi et~al.
\newblock {Aurora: A New Model and Architecture for Data Stream Management}.
\newblock {\em {VLDB Journal}}, 12(2), 2003.

\bibitem{borealis}
D.~Abadi et~al.
\newblock {The Design of the Borealis Stream Processing Engine}.
\newblock In {\em {CIDR}}, 2005.

\bibitem{arasu2004linear}
A.~Arasu et~al.
\newblock {Linear Road: A Stream Data Management Benchmark}.
\newblock In {\em {VLDB}}, 2004.

\bibitem{stream}
A.~Arasu et~al.
\newblock {STREAM: The Stanford Data Stream Management System}.
\newblock In {\em {Data Stream Management: Processing High-Speed Data
  Streams}}, 2004.

\bibitem{balazinska}
M.~Balazinska et~al.
\newblock {Fault-tolerance in the Borealis Distributed Stream Processing
  System}.
\newblock {\em {ACM TODS}}, 33(1), 2008.

\bibitem{secret}
I.~Botan et~al.
\newblock {SECRET: A Model for Analysis of the Execution Semantics of Stream
  Processing Systems}.
\newblock {\em {PVLDB}}, 3(1), 2010.

\bibitem{botan-edbt12}
I.~Botan et~al.
\newblock {Transactional Stream Processing}.
\newblock In {\em {EDBT}}, 2012.

\bibitem{sstore-demo}
U.~Cetintemel et~al.
\newblock {S-Store: A Streaming NewSQL System for Big Velocity Applications
  (Demonstration)}.
\newblock {\em {PVLDB}}, 7(12), 2014.

\bibitem{telegraphcq}
S.~Chandrasekaran et~al.
\newblock {TelegraphCQ: Continuous Dataflow Processing for an Uncertain World}.
\newblock In {\em {CIDR}}, 2003.

\bibitem{niagracq}
J.~Chen et~al.
\newblock {NiagaraCQ: A Scalable Continuous Query System for Internet
  Databases}.
\newblock In {\em {SIGMOD}}, 2000.

\bibitem{hekaton}
C.~Diaconu et~al.
\newblock {Hekaton: SQL Server's Memory-Optimized OLTP Engine}.
\newblock In {\em {SIGMOD}}, 2013.

\bibitem{weikum-book}
G.~W. et~al.
\newblock {\em {Transactional Information Systems: Theory, Algorithms, and the
  Practice of Concurrency Control and Recovery}}.
\newblock {Morgan Kaufmann}, 2001.

\bibitem{golab-edbt06}
L.~Golab et~al.
\newblock {On Concurrency Control in Sliding Window Queries over Data Streams}.
\newblock In {\em {EDBT}}, 2006.

\bibitem{golab-sigrec03}
L.~Golab and T.~Ozsu.
\newblock {Issues in Data Stream Management}.
\newblock {\em {ACM SIGMOD Record}}, 32(2), 2003.

\bibitem{hwang}
J.-H. Hwang et~al.
\newblock {High-Availability Algorithms for Distributed Stream Processing}.
\newblock In {\em {ICDE}}, 2005.

\bibitem{streamsql}
N.~Jain et~al.
\newblock {Towards a Streaming SQL Standard}.
\newblock {\em {PVLDB}}, 1(2), 2008.

\bibitem{johnson2008query}
T.~Johnson et~al.
\newblock {Query-aware Partitioning for Monitoring Massive Network Data
  Streams}.
\newblock In {\em {SIGMOD}}, 2008.

\bibitem{hstore}
R.~Kallman et~al.
\newblock {{H}-{S}tore: A High-Performance, Distributed Main Memory Transaction
  Processing System}.
\newblock {\em {PVLDB}}, 1(2), 2008.

\bibitem{lerner2003virtues}
A.~Lerner and D.~Shasha.
\newblock {The Virtues and Challenges of Ad Hoc + Streams Querying in Finance}.
\newblock {\em IEEE Data Engineering Bulletin}, 26(1), 2003.

\bibitem{malviya14}
N.~Malviya et~al.
\newblock {Rethinking Main Memory OLTP Recovery }.
\newblock In {\em {ICDE}}, 2014.

\bibitem{pavlo12}
A.~Pavlo et~al.
\newblock {Skew-Aware Automatic Database Partitioning in Shared-Nothing,
  Parallel {OLTP} Systems}.
\newblock In {\em {SIGMOD}}, 2012.

\bibitem{shah}
M.~A. Shah et~al.
\newblock {Highly Available, Fault-tolerant, Parallel Dataflows}.
\newblock In {\em {SIGMOD}}, 2004.

\bibitem{db-book}
A.~Silberschatz et~al.
\newblock {\em {Database System Concepts}}.
\newblock {McGraw-Hill}, 2010.

\bibitem{storm}
A.~Toshniwal et~al.
\newblock {Storm @Twitter}.
\newblock In {\em {SIGMOD}}, 2014.

\bibitem{wang-vldb11}
D.~Wang et~al.
\newblock {Active Complex Event Processing over Event Streams}.
\newblock In {\em {VLDB}}, 2011.

\bibitem{zaharia13}
M.~Zaharia et~al.
\newblock {Discretized Streams: Fault-tolerant Streaming Computation at Scale}.
\newblock In {\em {SOSP}}, 2013.

\end{thebibliography}
