This application claims priority to Provisional Application 61/911,545 filed 12/04/2013, the content of which is incorporated by reference.

BACKGROUND

To become more efficient, effective, and competitive, enterprises are expecting ever increasing benefits from data analytics. To meet this demand, data analytics platforms are including more data sources, which may be both internally and externally available. These data sources are often stored in distributed data stores. Data analytics applications or data scientists query the data from these distributed stores and merge and join the data to generate coherent analysis reports. With continuously increasing data sizes, querying and joining data from distributed sources can generate a significant amount of data traffic on the network, an issue that is exacerbated if the network is shared with other applications as well. Therefore, optimizing queries that access the distributed data stores, and specifically optimizing their network utilization, is likely to be an important problem to address in order to deliver improved query performance and query service differentiation.

Distributed data processing is supported by products from almost all major database system vendors nowadays. However, for decades, network has always been a major concern for performance management of distributed relational databases. Distributed queries suffer from bad performance in terms of query execution time when they encounter network resource contention. The main cause is due to the fact that a distributed query optimizer treats the underneath network as a black-box: it is unable to monitor it. Therefore, a traditional distributed query optimizer may select a bad query execution plan without dynamic network resource usage information.

In the past, people in database community expend considerable effort to work around the network rather than work with the network. For example, most of the distributed query optimizers consider the underneath network as a black-box and assume a constant parameter for the available network bandwidth. Some of the distributed query optimizers select and execute the plan that has the least cost albeit the network condition changes overtime. Although other distributed query optimizers make efforts to react to expected delays by scrambling, the decisions in their algorithm are either heuristic-driven which is prone to making poor scrambling decisions in some cases or inaccurate due to poor state of estimation for remote date access.

SUMMARY

In one aspect, systems and methods are disclosed for selecting a query plan in a database by monitoring network state information and flow information; and selecting an adaptive plan for execution with a query manager that receives the network state information and flow information, including: receiving a query, parsing the query, generating and optimizing a global query plan; dividing the global query plan into local plans; sending the local plans to

corresponding data store sites for execution with separate threads; and orchestrating data flows among the data store sites and forwarding a final result to a user.

Implementations of the method can include one or more of the following.

1. Creating a monitoring framework for collecting the current network bandwidth usage information.

2. Creating a cost model as a function of the available network bandwidth for distributed query plans in relational distributed databases.

3. Creating a query optimizer in relational distributed databases to adaptive ly select the best query plan with the shortest query execution time.

Advantages of the system may include one or more of the following. The system provides better performance: because the query optimizer will select the best query plan adaptively according to the dynamic network resource usage, query execution time is shorter. With greater visibility into the network's state, a distributed query optimizer could make more accurate cost estimates for different query plans and make better informed decisions. Moreover, as the optimizer could have some control of the network's future state, a distributed query optimizer could request and reserve the network bandwidth for a specific query plan and thereby improve query performance and query service differentiation. BRIEF DESCRIPTION OF THE DRAWINGS

FIG. 1 shows an exemplary network monitoring process.

FIG. 2 shows an exemplary adaptive plan selection process.

FIG. 3 shows an exemplary method for adaptive query plan selection in distributed relational database management system based on software-defined network.

FIG. 4 shows an exemplary system for adaptive query plan selection in distributed relational database management system based on software-defined network.

DESCRIPTION

FIGS. 1-4 show a system that works with software-defined networking (SDN) and enables a distributed query optimizer to achieve such visibility into and control of the network's state. Given dynamic network bandwidth usage information which is provided by software- defined network, the system how to select the best query plan among candidate query execution plans which can offer the shortest query execution time.

By decoupling the system that makes decisions about where traffic is sent (the control plane) from the underlying systems that forward traffic to the selected destination (the data plane), network services can be managed through an abstraction of lower level functionality. Thus, SDN raises the possibility that it is for the first time feasible and practical for distributed query optimizers to carefully monitor and even control the network. Our goal in this paper is to begin the exploration of this capability, and to try to gain insight into whether it really is a promising new development for distributed query optimization. SDN can indeed be effectively exploited for the performance management of analytical queries in distributed data store environments. Our system can analyze and show the opportunities SDN provides for distributed query optimization.

The system adaptively selects the optimal query plan based on the information provided by the network before the query execution. This method observes the status of the network and reacts by adapting the query execution plan to one that yields better performance.

A distributed query processor can be used to deliver differentiated query service to the users with different priorities. One method allows for network traffic prioritization and the second method provides the capability of reserving a certain amount of bandwidth for specific queries and making use of that guaranteed bandwidth during query optimization. These methods achieve run-time query service differentiation in shared and highly utilized networks, which was not possible before.

A method to model dynamic communication costs is used. We integrate the model into a distributed query optimizer along with an existing computational cost model and show its effectiveness. In one embodiment, a distributed data store environment is built using multiple instances of open source databases running on an SDN network with commercial OpenFlow enabled switches. Experimental results confirm our expectations and clearly show the benefits of the SDN technologies.

FIG. 1 shows an exemplary network monitoring process. The process receives as input the network state information including flows, network topology (hosts, switches, ports), queues, links and their capabilities (101). The process updates flow information (in one embodiment using OpenFlow protocol) (102). The flow information is summarized and sent to an adaptive optimizer (103). Operations 101-104 are repeated for all monitoring intervals (104).

FIG. 2 shows an exemplary adaptive plan selection process. In 201, the process receives as inputs global flow information, query with candidate plans, and cost models. In 202, the process estimates the cost for each candidate plan using the global flow information based on the cost model. In 203, the process selects the best plan that has the lowest cost and executes the plan. In 204, operations 201-203 are repeated for each incoming queries.

FIG. 3 shows an exemplary method 300 for adaptive query plan selection in distributed relational database management system based on software-defined network. The first step is the monitoring process. It monitors all the traffic of the flows in the openflow switches based on openflow protocol.

The second step is the adaptive plan selection. Here we propose a cost model to calculate the cost for a candidate plan based on the network status. And, based on the cost, the best plan that has the lowest cost is selected and executed.

The first part is network monitoring 302 which uses open flow protocol to monitor network status in 304 and updates global status in 305. In 304, the system uses openflow protocol to monitor network status. Before software-defined network is invented, network is treated as a black-box and it is impossible to observe network status in prior art. The second part is an adaptive plan selection and execution in 303. The operation 303 uses the plan generator to generate candidate plans in 306. Operation 303 then estimates the cost for each candidate plan using the global flow information based on the cost model in 307 and then selects the best plan with the lowest cost and executes the plan in 308. In 307, the system uses cost model which is able to estimate the cost for a candidate plan using the global flow information. Previous work assumes that network cost is a fixed parameter. As a result, each candidate plan also has a fixed cost. In 308, the system adaptively selects the best plan that has the lowest cost from all the candidate plans. Previous work assumes a static best plan based on the cost calculation.

We have the following considerations: (1) Relational and SQL: For concreteness and the simplicity of the presentation, we assume in this paper that the stores are relational databases and that SQL is used to query the databases. (2) Analytical workloads: We consider data intensive analytical workloads as we expect that they are the most likely to benefit from the SDN technologies due to their heavy use of the interconnection network. (Transactional systems are unlikely to consume prolonged, high network bandwidth, as queries are typically very short and involve smaller amounts of data transfer.) Continuing this observation, the queries we consider are mostly read-only, consuming large amounts of network bandwidth. (3) Shared network: We also observe that many data analytics applications run on shared networks along with other applications that use the same network, sometimes competing for the network resources, which is consistent with many real world scenarios.

FIG. 4 shows the overall system architecture. The evaluation system is mainly composed of a user site, a master site, several data store sites, and an SDN component, which consists of an OpenFlow controller and OpenFlow switches. The unit of distribution in the system is a table and each table is either stored at one data store or can be replicated to more than one data stores. A user or application program submits the query to the master site for compilation. The master site coordinates the optimization of all SQL statements. We assume that only the data store sites store the tables. The master and the data stores run off-the-shelf, modified database servers (PostgreSQL, in our case). A query manager runs on the master site, which consists of a distributed query processor and a network information manager (NIM). The distributed query processor presents an SQL API to users. It also maintains a global view of the meta-data for all the tables in the databases. The query manager communicates with the OpenFlow controller to (1) receive network resource usage information, and update the information in NIM accordingly; and (2) send the control commands to the OpenFlow controller. The basic operation of the system is as follows: when the query manager receives a query, it parses the query, generates, and optimizes a global query plan. The global query plan is divided into local plans. The local plans are sent to corresponding data store sites for execution via separate threads. The query manager orchestrates the necessary data flows among the data store sites. The query manager also forwards the final results from the master to the user.

In order to keep the programming simple, how data is stored and accessed via the network should be transparent to users. We map the table names used by the users, which we call the print names, to internal System Wide Names, SWN. An SWN has the form T ^s which denotes that a copy of table T is stored at site S . For convenience, if there is a single copy of table T , we also denote the site that has this copy as S _T . The system uses a distributed catalog. The catalogs at each data store site maintain the information about the tables in the database, including the replicas stored at that site. The catalog at the master site keeps the information indicating where each table is currently stored and this entry is updated if a table is moved.

After name resolution, a set of candidate plans P are generated. Each plan is a tree such that each node of the tree is a physical operator, such as a sequential scan, sort, or hash join. A physical operator can be either blocking or nonblocking. An operator is blocking if it cannot produce any output tuples without reading all of its input. For instance, the sort operator is a blocking operator.

There are two cost models that can be used to estimate the cost of a plan. The classic cost model, which estimates the total resource consumption of a query, is useful for maximizing the overall throughput of a system. The response time model, which estimates the total response time of a query, is useful for minimizing query execution time. We use the response time model in this paper.

The optimizer estimates query execution cost by aggregating the cost estimates of the operators in the query plan. To distinguish blocking and non-blocking operators, this cost model considers both the start _cost and total _cost of each operator: start _cost (sc ) is the cost before the operator can produce its first output tuple; total _cost (tc ) is the cost after the operator generates all of its output tuples. Note that the cost of an operator includes the cost of its child operators. The run_cost ( rc ) is defined as rc = tc -sc . The total cost of a query plan P , denoted as C , is the total _cost of the root operator.

There are generally two kinds of operators in a distributed query execution plan, (1) local operators, 0 _L , which do not involve shipping data over the network; and (2) network operators, 0 _N , which do involve data shipping over the network. For example, in Figure 3(b), the scan, hash, and hashjoin operators are local operators, while the function scan (func scan) operator is a network operator.

Based on the cost models of local and network operators, we summarize how we estimate the cost C _p for a plan P as follows. Here each brace means a dependency relationship.

and the data transfer rate, i.e., real-time bandwidth consumption for 0 _N denoted as C(U) ₀ .

C(U) ₀ ^ further depends on the upper bound bandwidth consumption for 0 _N (i.e., UB ₀ ), the available bandwidth for user U for 0 _N (i.e., A(U) ₀ ), and the reserved bandwidth for 0 _N by user U . Generally speaking, we define a network traffic matrix as a | S | x | S | matrix where | S | is the total number of sites. The rows of the matrix correspond to the source sites while the columns correspond to the destination sites. Cap denotes the port capacity, which is a constant lGbps in our setting, and all the elements in the matrix should be less than Cap . The available bandwidth matrix for user U is a network traffic matrix, denoted as A(U) . If we assume that network operator 0 _N involves data shipping from S _src to S _dst , then the available bandwidth for 0 _N , denoted as A(U) ₀ is the value at row S _src and column S _dst of A(U) .

N Compared with a traditional distributed query optimizer and executor, the query optimizer and executor in our system have the following distinguishing features:

1. A traditional distributed query optimizer generally models the network as a FIFO queue with a constant bandwidth. However, because the total cost C _P depends on A(U) in our system, our optimizer can adapt to the dynamic network status when choosing the best plan.

2. In traditional distributed query processing, once the best query plan is selected, it will be executed. If many lower priority queries are saturating the network, a traditional distributed query processing can do nothing to expedite an incoming important query. However, our query optimizer can "protect" the important queries by either giving them higher priority to use network bandwidth than the lower priority queries or by reserving and using the reserved network bandwidth.

SDN is an approach to networking that decouples the control plane from the data plane. The control plane is responsible for making decisions about where traffic is sent, while the data plane forwards traffic to the selected destination. This separation allows network administrators and application programs to manage network services through abstraction of lower level functionality by using software APIs. From a DBMS point of view, the abstraction and the control APIs allow the DBMS to (1) inquire about the current status and performance of the network, and (2) control the network with directives, for example, with bandwidth reservations.

OpenFlow is a standard communication interface among the layers of an SDN architecture, which can be thought of as an enabler for SDN. An OpenFlow controller communicates with an OpenFlow switch. An OpenFlow switch maintains a flow table, with each entry defining a flow as a certain set of packets by matching on 10 tuple packet information. When a new flow arrives, according to the OpenFlow protocol, a ^" Packetln" message is sent from the switch to the controller. The first packet of the flow is delivered to the controller. The controller looks into the 10 tuple packet information, determines the egress (exiting) port and sends a ^" FlowMod" message to the switch to modify a switch flow table. More specifically, APIs in the OpenFlow switch enable us to attach the new flow to one of the physical transmitter queues behind each port of the switch. When an existing flow times out, according to OpenFlow protocol, a ^" FlowRemoved" message is delivered from the switch to the controller to indicate that a flow has been removed. There are already OpenFlow controllers and switches that implement the OpenFlow standard from the major vendors in the industry. In our studies we also use actual commercial products from one of those vendors, NEC.

For example, we show a commercial OpenFlow switch NEC PFS5240 and three data store sites S _{0 l 2} connected to the switch at port 0,1,2 in Figure 4. There is a receiver and a transmitter behind each port of the switch and there are 8 transmission queues q8 to ql inside a transmitter. When a new flow Flow ₀ (from S ₀ to S ₂ ) under user U 's name arrives, a "Packetln" message is sent from the switch to the controller. The controller looks into the 10 tuple packet information, determines the egress ports (i.e., 2) and one of the transmission queues (e.g., q8) according to the user's priority U _pri and sends a ^" lowMod" message to the switch to modify a switch flow table. The following packets in the same flow will be sent through the same transmission queue q8 of the egress ports (i.e., 2) to site S ₂ . If no user information is specified, a default queue ( q4) will be used.

The OpenFlow API is used to implement our performance management methods. The network information manager (NIM) updates and inquires information about the current network state by communicating with the OpenFlow controller. The network information includes the network topology (hosts, switches, ports), queues, and links, and their capabilities. The runtime uses the information to translate the logical actions to a physical configuration, and to host the switch information such as its ports' speeds, configurations, and statistics. It is important to keep this information up-to-date with the current state of the network as an inconsistency could lead to under-utilization of network resources as well as bad query performance. In the NIM, we define a Flow as a four tuple:

Flow : := [src , dst , queue , rate ]

Here src and dst mean the ingress and egress ports of the switch for the flow, respectively, queue means the egress queue of the flow, and rate means the traffic rate. For example, we can have two flows, Flow ₀ = [0, 2, q8, 200Mbps] and Flow ₁ = [1, 2, ql,

200Mbps] as shown in Figure 4. Flow ₀ means that the flow is from port 0 (S ₀ ) to q8 of port 2 (

S ₂ ) and the rate is 200Mbps. The distributed query processor sends an inquiry to the network information manager to inquire A(U) ₀ , i.e., the available bandwidth for network operator 0 _N for user U . More specifically, it is calculated as

A(U) ₀ = Cap - ∑ Flow.rate (1)

Flow .dst=0 ^ .dst

Generally, we are interested in the flows that could compete with 0 _N at the transmitter. These flows should share the same destination port with 0 _N , i.e., Flowdst = 0 _N .dst . We sum up all these flows and the remaining bandwidth is assumed to be the available bandwidth for 0 _N . Note that A(U) ₀ as calculated by the above formula is a very rough estimation of the available bandwidth for 0 _N as there are various factors that we do not take into consideration, e.g., interaction between different flows with different internet protocols UDP and TCP.

For example, assume that we have two flows, Flow ₀ and Flow ₁ , and a network operator 0 _N . 0 _N 's destination port is also port 2 and 0 _N uses the default queue q4 as shown in Figure 4. Because there is no defined network traffic differentiation at this moment, all the queues q8, q4, ql have the same priority. Then A(U) ₀ =lG-(200M+200M)=624Mbps.

Our distributed query processor can communicate with the OpenFlow controller to leverage the OpenFlow APIs to pro-actively notify the switch to give certain priority to or make a reservation for specific flows. The main mechanism in the OpenFlow switch to implement these methods is the transmission queues. We show two examples using a priority queue (PQ) and a weighted fair queue (WFQ) in our system while the other options could also be possible. For example, combining PQ and WFQ could be considered to resolve more difficult network resource contention situations, which could be a future work.

In this case, we set the queues within the switch as priority queues (PQ). If more than one queue has queued frames, PQ sends frames in the order of queue priority. During the transmission, this configuration gives higher-priority queues absolute preferential treatment over lower-priority queues. If any port is set as PQ, then the queues from the highest priority to the lowest priority are q8,q7,...,ql . Under this setting, the calculation of the available bandwidth for 0 _N should be changed accordingly:

A(U) ₀ = Cap - ∑ Flow.ra.te (2)

Here Flow .queue .pri means the priority of queue and U.pri means the priority of user U ( 0 _N 's priority is the same as the user's priority who submits the query). Compared with (1), besides sharing the same destination port with 0 _N , the competing flows should have equal or higher priority than 0 _N , i.e., Flow. queue. pri≥ U.pri .

For example, assume that we have two flows, Flow ₀ and Flow _l , and a network operator 0 _N as shown in Figure 4. 0 _N 's destination port is also port 2 and 0 _N is assigned by OpenFlow controller to use queue q4 according to the user U 's priority. Because q4 has higher priority than ql and lower priority than q8, only Flow ₀ will compete with 0 _N . Thus, A(U) ₀ =1G-

200M=824Mbps. We can see that the available bandwidth for 0 _N is 200Mbps more than the case when no network traffic differentiation is applied (624Mbps). Because the cost of 0 _N depends on A(U) ₀ , the distributed query optimizer selects the query plan accordingly.

In this case, we set the port within the switch as weighted fair queues. After setting the weight (minimum guaranteed bandwidth) on every queue, the switch sends the amount of frames equivalent to the minimum guaranteed bandwidth from each queue to begin with. Under this setting, the calculation of the available bandwidth for 0 _N should be changed accordingly:

A(U) ₀ = Max(Cap - ∑ Flow.rate, R(U) ₀ )

Here R(U) ₀ ^ is the bandwidth reservation for 0 _N by user U . For example, assume that we have two flows, Flow ₀ and Flow _x , and a network operator 0 _N as shown in Figure 4. We assume that the user makes an 800Mbps bandwidth reservation for 0 _N and the other users do not make any bandwidth reservations. By calculation, A(U) ₀ is equal to the bandwidth reservation (i.e., 800Mbps). We can see that the available bandwidth for 0 _N is more than the case when no network traffic differentiation is applied (624Mbps). Similar to the previous cases, this method computes A(U) ₀ value, which affects the cost of 0 _N , and in turn, the plan selection of the distributed query optimizer. Note that WFQ works in a work conserving mode in this switch. That is, although 0 _N is guaranteed 800Mbps, if 0 _N does not use 800Mbps, the other flow can use the remaining bandwidth. If 0 _N indeed uses the capacity and also the other flows also use up the maximum capacity, the system guarantees the reserved capacity for 0 _N and serves the other flows with the remaining capacity by throttling them as necessary.

The system leverages software-defined networking for the performance management of analytical queries in distributed data stores in a shared networking environment. The system utilizes greater visibility into the network's state and makes more informed decisions to adaptively pick the best plan. The system can control the priority of network traffic or make network bandwidth reservations according to different users' priorities, thereby differentiating the query service. The instant methods exhibit significant potential for the performance management of analytical queries in distributed data stores. The system enhances distributed data intensive computing by combing SDN and distributed database technologies.

While the foregoing is directed to embodiments of the present invention, other and further embodiments of the invention may be devised without departing from the basic scope thereof, and the scope thereof is determined by the claims that follow.