Writing Federated Queries

In the previous section we provide two example queries to provide us the answers to:

• What are the top ten airline carriers with the most flights?

• What are the top ten airports that had the most departures?

By querying our airline data source, we were able to obtain the results. However, it was somewhat difficult to extract meaning from two of the columns. In the following examples we will show how we can provide more meaning to the results by combining with additional data in a relational database. We choose to use PostgreSQL in our examples, but it will be applicable for any relational database.

As with the airline data, our GitHub repository includes the setup for creating and loading tables in a relational database as well as configuring the Presto connector to access it. We’ve chosen to configure Presto to query from a PostgreSQL database that contains additional airline data. One table, carrier, in PostgreSQL provides a mapping of the airline code to the more descriptive airline name. We will use this additional data with our first example query. Let’s take a look at table carrier in PosrtgreSQL.

presto:public> SELECT * FROM carrier LIMIT 10;
 code | description 
 ------+----------------------------------------------
 02Q | Titan Airways 
 04Q | Tradewind Aviation 
 05Q | Comlux Aviation, AG 
 06Q | Master Top Linhas Aereas Ltd. 
 07Q | Flair Airlines Ltd. 
 09Q | Swift Air, LLC 
 0BQ | DCA 
 0CQ | ACM AIR CHARTER GmbH 
 0GQ | Inter Island Airways, d/b/a Inter Island Air 
 0HQ | Polar Airlines de Mexico d/b/a Nova Air 
 (10 rows)

This looks promising. You’ll see that there is a column code along with a description of the code. Using this information, we can use our first example query for the flights_orc table and modify it to join with the data in the PostgreSQL carrier table.

presto:ontime> SELECT f.uniquecarrier, c.description, count(*) AS ct 
 FROM hive.ontime.flights_orc f, 
 postgres.public.carrier c
 WHERE c.code = f.uniquecarrier
 GROUP BY f.uniquecarrier, c.description
 ORDER BY count(*) DESC
 LIMIT 10;
 uniquecarrier | description | ct 
 ---------------+----------------------------+----------
 WN | Southwest Airlines Co. | 24096231 
 DL | Delta Air Lines Inc. | 21598986 
 AA | American Airlines Inc. | 18942178 
 US | US Airways Inc. | 16735486 
 UA | United Air Lines Inc. | 16377453 
 NW | Northwest Airlines Inc. | 10585760 
 CO | Continental Air Lines Inc. | 8888536 
 OO | SkyWest Airlines Inc. | 7270911 
 MQ | Envoy Air | 6877396 
 EV | ExpressJet Airlines Inc. | 5391487 
 (10 rows)

Viola! Now that we have written a single SQL query to federate data from S3 and PostgreSQL, we’re able to provide more valuable results of the data for one to extract meaning. Instead of having to know or separately look up the airline codes, the descriptive airline name is in the results.

In the query we wrote, you’ll see in bold the places we added columns from the postgres table carrier. We also used fully qualified names when referencing the tables. When utilizing the USE command to set the default catalog and schema, a non qualified table name is linked to that catalog and schema. However, anytime you need to query outside for the catalog and schema, the table name must be qualified. Otherwise Presto will try to find it within the default catalog and schema, and will return an error. If you are referring to a table within the default catalog and schema, it is not required to fully qualify the table name. However, it’s recommended to as best practice whenever referring to data sources outside of the default scope.

Next, let’s look at another table airport in PostgreSQL. This table will be used as part of federating our second example query.

presto:public> SELECT code, name, city FROM airport LIMIT 10;
 code | name | city 
 ------+--------------------------+----------------------
 01A | Afognak Lake Airport | Afognak Lake, AK 
 03A | Bear Creek Mining Strip | Granite Mountain, AK 
 04A | Lik Mining Camp | Lik, AK 
 05A | Little Squaw Airport | Little Squaw, AK 
 06A | Kizhuyak Bay | Kizhuyak, AK 
 07A | Klawock Seaplane Base | Klawock, AK 
 08A | Elizabeth Island Airport | Elizabeth Island, AK 
 09A | Augustin Island | Homer, AK 
 1B1 | Columbia County | Hudson, NY 
 1G4 | Grand Canyon West | Peach Springs, AZ 
 (10 rows)

Looking at this PostgreSQL, you’ll see that there is a column code that can be used to join with our second query on the flight_orc table. Doing this will allow us to extend the additional information in the airport table to our query to provide a more details an understandable report.

presto:ontime> SELECT f.origin, c.name, c.city, count(*) AS ct
 FROM hive.ontime.flights_orc f, 
 postgres.public.airport c
 WHERE c.code = f.origin
 GROUP BY origin, c.name, c.city
 ORDER BY count(*) DESC
 LIMIT 10;
 origin | name | city | ct 
 --------+-----------------------------------+-----------------------+---------
 ATL | Hartsfield-Jackson Atlanta International | Atlanta, GA | 8867847 
 ORD | Chicago O'Hare International | Chicago, IL | 8756942 
 DFW | Dallas/Fort Worth International | Dallas/Fort Worth, TX | 7601863 
 LAX | Los Angeles International | Los Angeles, CA | 5575119 
 DEN | Denver International | Denver, CO | 4936651 
 PHX | Phoenix Sky Harbor International | Phoenix, AZ | 4725124 
 IAH | George Bush Intercontinental/Houston | Houston, TX | 4118279 
 DTW | Detroit Metro Wayne County | Detroit, MI | 3862377 
 SFO | San Francisco International | San Francisco, CA | 3825008 
 LAS | McCarran International | Las Vegas, NV | 3640747 
 (10 rows)

Presto! As with our first example, we can provide more meaningful information by federating across two disparate data sources in order to combine and provide more meaningful information. Here we are able to add in the name of the airport instead of the user relying on airport codes that are hard to intrepret.

Extract, Transform, Load

Extract Transform Load (ELT) is a term used to describe the technique of copying data from data sources and landing it into another data source. Often there is a middle step of transforming the data from the source in preparation for the destination. This may include dropping columns, feature engineering, joining in data, pre aggregations and other ways to prepare and make it suitable for querying the destination.

Presto is not intended to be a full fledged ETL tool comparable to a commercial solution. However, it can assist by avoiding the need for ETL. Because Presto can query from the data source, there may no longer be a need to move the data. Presto will query the data where it lives to alleviate the complexity of managing the ETL process.

It may still be desired to do some type of ETL transformations. Perhaps you want to query on pre aggregated data or you don’t want to put a stress on the underlying system. By using the CREATE TABLE AS or INSERT SELECT constructs you can move data from one data source into another such as HDFS or S3.

Federated Architecture

Now that we’ve gone through some examples of query federation from an end user perspective, let’s discuss the architecture of how this works. We will build some of the concepts we learned about in {{Chapter X}} on the Presto architecture.

Presto is able to coordinate the hybrid execution of the query across the data sources involved in the query. In the example earlier, we were querying between distributed storage and PosgreSQL. For distributed storage via the Hive connector, Presto reads the data files directly whether it’s from HDFS, S3, Azure Blob Storage, etc. For a relational database connector such as the PostgreSQL connector, Presto will rely on PostgreSQL to do part of the execution. Let’s use our query from earlier, but to make it more interesting let’s add a new predicate that refers to a column in the PostgreSQL airport table.

SELECT f.origin, c.name, c.city, count(*) AS ct
 FROM hive.ontime.flights_orc f, 
 postgres.public.airport c
 WHERE c.code = f.origin AND c.state = ‘AK’ 
 GROUP BY origin, c.name, c.city
 ORDER BY count(*) DESC
 LIMIT 10;

The logical query plan will resemble something similar the following figure. You will see the plan consists of scanning both the flights_orc and airport tables. Both inputs are fed into the join operator. But before the airport data is fed into the join, a filter is applied because we only want to look at the results for airports in Alaska. After the join, the aggregation and grouping operation is applied. And then finally the TopN operator performs the order by and limit combined

Figure 9-1. This logical query plan in Presto shows how data is federated across two data sources using the Hive and PostgreSQL connector.

In order for Presto to retrieve the data from PostgreSQL, it will send a query via JDBC. For example, in the naive approach the following query is sent to PostgreSQL.

SELECT * FROM public.airport

However, Presto is smarter than this, and the Presto optimizer will try to reduce the amount of data transferred between system. In this example, Presto will only query the columns it needs from the PostgreSQL table as well as push down the predicate into the SQL that is sent to PostgreSQL.

So now the query sent from Presto to PostgreSQL will push more processing to PostgreSQL.

SELECT code, city, name FROM public.airport WHERE state = ‘AK’

As JDBC connector to PostgreSQL returns data to Presto, Presto will continue processing the data for the part that will be executed in the Presto query engine.

Figure 9-2. Using the PostgreSQL connector, Presto will push down filter processing into PostgreSQL helping improve query performance by retrieving the data Presto needs.

Some simpler queries such as SELECT * FROM public.airport would be entirely pushed down into the underlying data source such that more the query execution happens outside of Presto and Presto acts as a passthrough.

Currently, more complex SQL pushdown is not supported. For example, Aggregations or Joins that involve only the RDBMS data could be pushed into the RDBMS to eliminate data transfer to Presto. The Presto community has been discussing this and will be something Presto will support in time.

Security in Federation

We will learn more about Presto security in general in {{Chapter X}}. For query federation, both user authentication (“who are you?”) and use authorization (“what can you do?”) are important pieces. There are multiple points for authentication (“AuthN”) and authorization (“AuthZ).

Figure 9-3. Presto is designed to be a trusted system and authenticates to the relational databases. {{Chapter X}} discusses Presto Security in greater detail.

The user will first authenticate to the coordinator. And because Presto connector will be issuing requests to the data sources there will be some authentication required as well. The client can authenticate to the Presto coordinator over LDAP or Kerberos. Authorization is enforced by Presto’s system access control which enforces authorization at the global level.

Authentication from the connectors to the data sources are connector implementation dependent. In many connector implementations, Presto authenticates as a service users. Therefore any users that runs a query using such a connector, the query will be executed in the underlying system as that service users. To provide a more fine grained access control, you could create several different service users with different permissions. You can then have multiple configurations of the connector for each service user. And finally, restrict access to these connectors by levering the system access control for catalogs.