Shasta: Interactive reporting at scale

Workload

Power the the UI with both OLAP + OLTP functionalities in the same UI, e.g., the AdWords Web UI.
- Users can select columns, filters, and segmentation in the UI, while still expecting sub-second latency.
Because users expect near real-time experience, queries have to go from the primary storage to user facing views.
- Transitional data stores are optimized for writes.
- Underlying OLAP data stores and OLTP data stores are diverse
Interfaces and APIs are often scoped to an individual user’s business data, so that typical Shasta queries process only a modest subset of data in underlying data stores.

Pain points

Reporting query is complex, often joining 50+ tables from different data stores. For instance, computing the main table in the AdWords UI “Campaigns” tab involves joining and aggregating approximately 20 F1 tables and 20 Mesa tables
- Queries were way too large to be expressed gracefully in SQL. These complex SQLs are hard to reuse
Write-friendly schemas need significant transformation logic to be read-friendly. This “concept gap” tends to be particularly wide for OLAP functionality.
- Classic ETL + pre-computation is not feasible due to freshness requirements
- The necessary offline pipelines tend to make systems more stateful and operationally complex.
- Transactionally updated materialized views on the other hand increase the cost of writes.

Architecture

Does not use pre-computation or materialized view of intermediate results.
Translate a client request to a single SQL and send the SQL to F1. F1 then uses its distributed query engine to join between different storage systems
RVL is sql like, with support for view params and UDFs
Added UDF server to F1 to support UDFs in RVL
Added in-memory, read-only TableCache between F1 storage and query engine. For other storages such as Mesa or BigTable, such cache had questionable RoI.

Sequence of actions

FE server sends a query, which has view name, columns to query, and view params such as user ID, feature ID. The query open contains a timestamp for snapshot versioning
The query is accepted by the view gateway, binds the view params, and generates a SQL.
- RVL compiler uses RVL code + table metadata to generate SQL
The request handler then executes the SQL query on F1 and forwards the query results back to the Shasta client.
- The SQL is executed by the F1 query engine, which also contacts the UDF server and TableCache. The query engine contacts Mesa and BigTable storage directly.

RVL

RVL uses information in the schema to automatically aggregate data when columns are projected.
The support for implicit aggregation has parallels to the MDX language for querying OLAP data cubes.

Query Language

The syntax and semantics are similar to SQL, with one fundamental difference: RVL automatically determines how to aggregate query results.
the metadata of each column may optionally specify an implicit aggregation function. If a column has an implicit aggregation function, we refer to that column as an aggregatable column. Otherwise, it is a grouping column.
In the special case where all columns are grouping columns, the behavior of RVL is consistent with relational algebra on sets
RVL also supports syntax to modify or remove implicit aggregation functions assigned to the columns of a relation
SELECT, FROM, WHERE, JOIN, and UNION: The behavior of these operations is similar to SQL, except that implicit aggregation is applied after each operation to yield a set of unique rows

Sample 1

Employee table: EmpId, DeptId, BldgId, Salary[SUM]
Building table: BldgId, CityId, Capacity[SUM]

Q0 = SELECT *
      FROM Employee LEFT JOIN Building USING (BldgId);
Q1 = SELECT DeptId, Salary FROM Q0;

generates

 SELECT DeptId, SUM(Salary) FROM Employee
 GROUP BY DeptId;

The join with the Building table can be pruned since none of the Building columns are required.

Sample 2

 Q2 = SELECT CityId, Salary, Capacity FROM Q0;

generates

SELECT CityId, SUM(Salary) AS Salary,
        SUM(Capacity) AS Capacity
 FROM
   (SELECT BldgId, SUM(Salary) AS Salary
    FROM Employee GROUP BY BldgId)
   LEFT JOIN Building USING (BldgId))
 GROUP BY CityId;

We aggregate requested columns before joining
Building to employee is 1 to many, so we have to make sure each building capacity is summed once, instead of once per employee.

View Templates

RVL embeds its query language in view templates.
- A view template specifies a dynamic query using replaceable parameters.
- Within each template, a query can be built as a sequence of subquery assignment statements. Each subquery in a template may refer to template parameters and instantiate other view templates.

Motivations

The view parameters may change the tables used in joins or the placement of filters in the query. RVL needs to be more dynamic in order to capture this wide range of possible query structures.
A typical Shasta view would require 100s of lines of code, and expressing that as a single large query can be difficult to read and maintain.
Allow large queries to be constructed dynamically from smaller pieces, which can be composed and reused by multiple Shasta views. Idea similar to Spark SQL’s DataFrame API
RVL view templates may use control structures written as if/else blocks to dynamically choose between two or more subqueries.

Parameters

A view template may be referenced in the FROM clause of RVL queries by passing values for its input parameters
RVL text: A string containing valid RVL syntax can be bound to a view template parameter, and that parameter can be referenced in places where it would be valid to inject the RVL syntax string. For example, a template parameter bound to the string “X,Y” could be referenced in a SELECT clause, and the template parameter reference will behave exactly as if “X,Y” were written directly in the query.
- RVL text values do not allow for arbitrary code injection
- In order to make view templates less error-prone, an RVL text value is only allowed to contain a few specific syntactic forms, such as scalar expressions, expression lists, and base table names
Nested dictionary: A template parameter can be bound to a dictionary of name-value pairs, where the values can either be another dictionary, or RVL text. Intuitively, a nested dictionary is a collection of RVL text parameters with hierarchical names.
Subquery: A template parameter can be bound to an RVL subquery, and referenced anywhere a table can be referenced. A subquery value differs from RVL text values, in the sense that subquery values are substituted in a purely logical manner which is independent of the syntax used to create the subquery. In contrast, an RVL text value is purely a text injection, which allows any variables in the RVL text to be interpreted based on the context where the parameter is referenced.

 view FilterUnion<input_table, params> {
   T1 = SELECT $params.column_name FROM $input_table;
   T2 = SELECT $params.column_name FROM Employee;
   T = T1 UNION T2;
   return SELECT * FROM T
          WHERE $params.column_name >= $params.min_value;
}

input_table can be bound to a table name or subquery
params must be bound to a nested dictionary.

main view template

A view template can be designated as an entry point in the RVL code
RVL provides an API to invoke a main view template, with a nested dictionary as a parameter.

main OutputValues<params> {
   b = SELECT * FROM Building;
   all_values = SELECT * from FilterUnion<@b, $params>;
   output all_values AS result;
}

The output statement specifies a table to produce as the final result when the main view template is invoked, as well as an alias for that table.
If there are multiple output statements, the aliases must be unique so that the Shasta view gateway can distinguish the results. Multiple output statements can reference the same RVL subquery by name which is useful when applications need to display multiple pivots of the same shared view computation. To achieve consistency between different data pivots within a view query, RVL guarantees that each named subquery is only executed once.

Compiler

the RVL compiler first resolves references to view templates and named subqueries, producing an algebraic representation of an RVL query plan that includes all outputs of the invoked main view template. The RVL compiler performs some transformations to optimize and simplify the query plan before translating it to SQL
The RVL compiler optimizes query plans using a rule-based engine. Each rule uses a different strategy to simplify the plan based on algebraic structure, without estimating cost. In practice, rule-based optimization is sufficient because the only goal is to simplify the generated SQL, rather than determine all details of query execution. We avoid using cost-based optimization because a cost model would tie the RVL compiler to a specific SQL engine and make it less generic.

Column Pruning

When computing aggregate values after joining, column pruning can reorder join and aggregation steps, so that aggregations happens before joins whenever possible. This reordering simplifies the job for F1’s query optimizer.

SELECT CityId, Salary, Capacity
FROM Employee LEFT JOIN Building USING (BldgId);

Before column pruning

 SELECT CityId, SUM(Salary) AS Salary,
        SUM(Capacity) AS Capacity
 FROM
   (SELECT BldgId, CityId,
           SUM(Salary) AS Salary, Capacity
    FROM Employee LEFT JOIN Building USING (BldgId)
    GROUP BY BldgId, CityId, Capacity)
 GROUP BY CityId;

After

SELECT CityId, SUM(Salary) AS Salary,
        SUM(Capacity) AS Capacity
 FROM
   (SELECT BldgId, SUM(Salary) AS Salary
    FROM Employee GROUP BY BldgId)
   LEFT JOIN Building USING (BldgId))
 GROUP BY CityId;

Filter Pushdown

Filter pushdown can improve the effectiveness of the column pruning optimization. For example, if there is a filter on a column which is not part of the final result, the filter will prevent column pruning from removing the column before the filter is applied, and pushed down filter means the column can be pruned early.

Left Join Pruning

A user can add many left joins to their view templates to fetch columns which might not be required, and they can be confident that the RVL compiler will know which joins can be skipped.
if a left join does not require any of the columns from its right input, the right input can be removed from the query plan

QUERY EXECUTION ON F1

Shasta represents a novel use of F1’s query engine as it places heavily distributed query execution at the heart of latency sensitive and business-critical applications. While Shasta queries are typically scoped to a modest subset of data in underlying databases (e.g., a single advertiser’s campaign configurations and performance metrics), the combination of re- mote input data, diverse data stores, data freshness requirements, and complexity of query logic make it challenging to achieve low latencies reliably using centralized execution,
DAG-structured query plans
External Data Sources: Using a plugin framework for federated querying, the F1 query engine supports several data sources. Central to the plugin framework is the abstraction of a ScanOperator. For every supported data source, a ScanOperator implementation translates between the data source API and the F1 SQL runtime
RVL code can invoke user-defined functions (UDFs) written in procedural languages such as C++ or Java. Motivations include sharing critical procedural code between RVL definitions and other systems processing the same data, re-use of subtle legacy code, and expressivity.
In order to be able to register with the F1 query engine, UDF server binaries must implement a standard RPC API defined by F1.

Workload

Pain points

Architecture

Sequence of actions

RVL

Query Language

Sample 1

Sample 2

View Templates

Motivations

Parameters

main view template

Compiler

Column Pruning

Filter Pushdown

Left Join Pruning

QUERY EXECUTION ON F1

References